/**
* The `proact-streams` module provides stateless streams to the ProAct.js API.
* FRP reactive streams.
*
* @module proact-streams
* @main proact-streams
*/
// PRIVATE
var StreamUtil = {
go: function (event, useTransformations) {
if (this.listeners.change.length === 0) {
return this;
}
if (useTransformations) {
try {
event = P.Actor.transform(this, event);
} catch (e) {
StreamUtil.triggerErr.call(this, e);
return this;
}
}
if (event === P.Actor.BadValue) {
return this;
}
return ActorUtil.update.call(this, event);
},
triggerMany: function () {
var i, args = slice.call(arguments), ln = args.length;
for (i = 0; i < ln; i++) {
this.trigger(args[i], true);
}
return this;
},
trigger: function (event, useTransformations) {
if (useTransformations === undefined) {
useTransformations = true;
}
return StreamUtil.go.call(this, event, useTransformations);
},
triggerErr: function (err) {
return ActorUtil.update.call(this, err, 'error');
},
triggerClose: function (data) {
return ActorUtil.update.call(this, data, 'close');
}
};
/**
* <p>
* Constructs a `ProAct.Stream`.
* The stream is a simple {{#crossLink "ProAct.Actor"}}{{/crossLink}}, without state.
* </p>
* <p>
* The streams are ment to emit values, events, changes and can be plugged into other `Actors`.
* For example it is possible to connect multiple streams, to merge them and to separate them,
* to plug them into properties.
* </p>
* <p>
* The reactive environment consists of the properties and the objects containing them, but
* the outside world is not reactive. It is possible to use the `ProAct.Streams` as connections from the
* outside world to the reactive environment.
* </p>
* <p>
* The transformations can be used to change the events or values emitetted.
* </p>
* <p>
* `ProAct.Stream` is part of the proact-streams module of ProAct.js.
* </p>
*
* @class ProAct.Stream
* @extends ProAct.Actor
* @constructor
* @param {String} queueName
* The name of the queue all the updates should be pushed to.
* <p>
* If this parameter is null/undefined the default queue of
* {{#crossLink "ProAct/flow:property"}}{{/crossLink}} is used.
* </p>
* <p>
* If this parameter is not a string it is used as the
* <i>source</i>.
* </p>
* @param {ProAct.Actor} source
* A default source of the stream, can be null.
* @param {Array} transforms
* A list of transformation to be used on all incoming chages.
*/
function Stream (queueName, source, transforms) {
if (queueName && !P.U.isString(queueName)) {
transforms = source;
source = queueName;
queueName = null;
}
P.Actor.call(this, queueName, transforms);
this.sourceNumber = 0;
if (source) {
this.into(source);
}
}
ProAct.Stream = ProAct.S = Stream;
P.U.ex(P.S, {
fromString: function (str, args) {
throw new Error('Stream.fromString is not implemented!');
}
});
ProAct.Stream.prototype = P.U.ex(Object.create(P.Actor.prototype), {
/**
* Reference to the constructor of this object.
*
* @property constructor
* @type ProAct.Stream
* @final
* @for ProAct.Stream
*/
constructor: ProAct.Stream,
/**
* Creates the <i>event</i> to be send to the listeners on update.
* <p>
* Streams don't create new events by default, the event is the source.
* </p>
*
* @for ProAct.Stream
* @protected
* @instance
* @method makeEvent
* @param {ProAct.Event} source
* The source event of the event. It can be null
* @return {ProAct.Event}
* The event.
*/
makeEvent: function (source) {
return source;
},
/**
* Creates the <i>listener</i> of this stream.
*
* @for ProAct.Stream
* @protected
* @instance
* @method makeListener
* @return {Object}
* The <i>listener of this stream</i>.
*/
makeListener: function () {
if (!this.listener) {
var stream = this;
this.listener = function (event) {
if (stream.trigger) {
stream.trigger(event, true);
} else {
StreamUtil.trigger.call(stream, event, true);
}
};
}
return this.listener;
},
/**
* Creates the <i>error listener</i> of this stream.
* <p>
* The listener pushes the incoming event into `this Stream` by default.
* </p>
*
* @for ProAct.Stream
* @protected
* @instance
* @method makeErrListener
* @return {Object}
* The <i>error listener of this stream</i>.
*/
makeErrListener: function () {
if (!this.errListener) {
var stream = this;
this.errListener = function (error) {
if (stream.triggerErr) {
stream.triggerErr(event);
} else {
StreamUtil.triggerErr.call(stream, error);
}
};
}
return this.errListener;
},
/**
* Creates the <i>closing listener</i> of this stream.
*
* Pushes a closing notification into the stream by default.
*
* @for ProAct.Stream
* @instance
* @protected
* @method makeCloseListener
* @return {Object}
* The <i>closing listener of this stream</i>.
*/
makeCloseListener: function () {
if (!this.closeListener) {
var stream = this;
this.closeListener = function (data) {
if (stream.triggerClose) {
stream.triggerClose(data);
} else {
StreamUtil.triggerClose.call(stream, data);
}
};
}
return this.closeListener;
},
/**
* Defers a `ProAct.Actor` listener.
* <p>
* For streams this means pushing it to active flow using {{#crossLink "ProAct.Flow/push:method"}}{{/crossLink}}.
* If the listener is object with 'property' field, it is done using {{#crossLink "ProAct.Actor/defer:method"}}{{/crossLink}}.
* That way the reactive environment is updated only once, but the streams are not part of it.
* </p>
*
* @for ProAct.Stream
* @protected
* @instance
* @method defer
* @param {Object} event
* The event/value to pass to the listener.
* @param {Object} listener
* The listener to defer. It should be a function or object defining the <i>call</i> method.
* @return {ProAct.Actor}
* <i>this</i>
*/
defer: function (event, listener) {
if (!listener) {
return;
}
if (listener.property) {
P.Actor.prototype.defer.call(this, event, listener);
return;
}
var queueName = (listener.queueName) ? listener.queueName : this.queueName;
if (P.U.isFunction(listener)) {
P.flow.push(queueName, listener, [event]);
} else {
P.flow.push(queueName, listener, listener.call, [event]);
}
},
/**
* Creates a new `ProAct.Stream` instance with source <i>this</i> and mapping
* the passed <i>mapping function</i>.
*
* ```
* var mapped = stream.map(function (v) {
* return v * v;
* });
*
* mapped.on(function (v) {
* console.log(v); // squares
* });
* ```
*
* @for ProAct.Stream
* @instance
* @method map
* @param {Object} mappingFunction
* Function or object with a <i>call method</i> to use as map function.
* @return {ProAct.Stream}
* A new `ProAct.Stream` instance with the <i>mapping</i> applied.
*/
map: function (mappingFunction) {
return new P.S(this).mapping(mappingFunction);
},
/**
* Creates a new `ProAct.Stream` instance with source <i>this</i> and filtering
* the passed <i>filtering function</i>.
*
* ```
* var filtered = stream.filter(function (v) {
* return v % 2 === 1;
* });
*
* filtered.on(function (v) {
* console.log(v); // odds
* });
* ```
*
* @for ProAct.Stream
* @instance
* @method filter
* @param {Object} filteringFunction
* The filtering function or object with a call method, should return boolean.
* @return {ProAct.Stream}
* A new `ProAct.Stream` instance with the <i>filtering</i> applied.
*/
filter: function (filteringFunction) {
return new P.S(this).filtering(filteringFunction);
},
/**
* Creates a new `ProAct.Stream` instance with source <i>this</i> and accumulation
* the passed <i>accumulation function</i>.
*
* ```
* var acc = stream.accumulate(0, function (p, v) {
* return p + v;
* });
*
* acc.on(console.log); // sums
* ```
*
* @for ProAct.Stream
* @instance
* @method accumulate
* @param {Object} initVal
* Initial value for the accumulation. For example '0' for sum.
* @param {Object} accumulationFunction
* The function to accumulate.
* @return {ProAct.Stream}
* A new `ProAct.Stream` instance with the <i>accumulation</i> applied.
*/
accumulate: function (initVal, accumulationFunction) {
return new P.S(this).accumulation(initVal, accumulationFunction);
},
/**
* Creates a new `ProAct.Stream` instance that merges this with other streams.
* The new instance will have new value on value from any of the source streams.
*
* ```
* var merged = stream1.merge(stream2);
* ```
*
* Here if `stream1` emits:
* 1--2---3----5-----X
*
* and `steam2` emits:
* ----A-----B-----C-----D--X
*
* `merged` will emit:
* 1--2A--3--B-5---C-----D--X
*
* @for ProAct.Stream
* @instance
* @method merge
* @param [...]
* A list of streams to be set as sources.
* @return {ProAct.Stream}
* A new `ProAct.Stream` instance with the sources this and all the passed streams.
*/
merge: function () {
var sources = [this].concat(slice.call(arguments)),
result = new P.S();
return P.S.prototype.into.apply(result, sources);
},
/**
* Links source actors into this `ProAct.Stream`. This means that <i>this stream</i>
* is listening for changes from the <i>sources</i>.
*
* The streams count their sources and when the sources are zero, they become inactive.
*
* ```
* var stream1 = ProAct.stream();
* var stream2 = ProAct.stream();
* var stream = ProAct.stream();
*
* stream.into(stream1, stream2);
* stream.on(function (v) {
* console.log(v);
* });
*
* ```
*
* Now if the any of the source streams is emits,
* the notification will be printed on the output.
*
* @for ProAct.Stream
* @instance
* @method into
* @param [...]
* Zero or more source {{#crossLink "ProAct.Actor"}}{{/crossLink}}s to set as sources.
* @return {ProAct.Stream}
* <b>this</b>
*/
into: function () {
ProAct.Actor.prototype.into.apply(this, arguments);
this.sourceNumber += arguments.length;
return this;
},
/**
* Checks if <i>this</i> can be closed.
*
* Uses the number of the active sources to decide if `this stream` is ready to be closed.
* If the active sources are zero - it can.
*
* @for ProAct.Stream
* @protected
* @instance
* @method canClose
*/
canClose: function () {
this.sourceNumber -= 1;
return this.sourceNumber <= 0;
}
});
// Methods added to the ProAct.Actor from the proact-streams module.
P.U.ex(P.Actor.prototype, {
/**
* Turns this `ProAct.Actor` to a {{#crossLink "ProAct.Stream"}}{{/crossLink}}.
*
* In reality this method creates a new `Stream` with source this.
*
* @for ProAct.Actor
* @instance
* @method toStream
*/
toStream: function () {
return new P.S(this.queueName, this);
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It skips the first `n` updates incoming from `this`.
*
* source : --3---4---5--4---3---4---5--|->
* skip(3): -------------4---3---4---5--|->
*
* @for ProAct.Actor
* @instance
* @method skip
* @param {Number} n The number of notifications to skip.
*/
skip: function (n) {
var i = n, self = this;
return this.fromLambda(function (stream, event) {
if (event === ProAct.Actor.Close) {
stream.close();
return;
}
i--;
if (i < 0) {
self.offAll(stream.lambda);
stream.into(self);
stream.trigger(event);
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It skips notifications from its source, while a condition is true.
*
* ```
*
* source.skipWhile(function (v) {
* return v % 2 === 1;
* });
*
* // source :
* // --3---5---2--4---3---4---5--|->
* // skipWhile:
* // ----------2--4---3---4---5--|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method skipWhile
* @param {Function} condition
* A condition function, which is called for each of the incoming values
* While it returns true, the elements are skipped,
* after it returns false for the first time, the current and all the following values are emitted.
*/
skipWhile: function (condition) {
var self = this,
cond = condition ? condition : function (e) {
return e;
};
return this.fromLambda(function (stream, event) {
if (event === ProAct.Actor.close) {
stream.close();
return;
}
if (!cond(event)) {
self.offAll(stream.lambda);
stream.into(self);
stream.trigger(event);
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It skips dublicating elements, comming one after another.
*
* ```
*
* source.skipDuplicates();
*
* // source :
* // --3---5---5--4---3---3---5--|->
* // skipDuplicates:
* // --3---5------4---3-------5--|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method skipDuplicates
* @param {Function} comparator
* A function used to compare the elements.
* If nothing is passed it defaults to comparing using `===`.
*/
skipDuplicates: function (comparator) {
var last = undefined,
cmp = comparator ? comparator : function (a, b) {
return a === b;
};
return this.fromLambda(function (stream, event) {
if (!cmp(last, event)) {
stream.trigger(event);
last = event;
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It emits the difference between the last update and the current incomming update from the source.
*
* ```
*
* source.diff(0, function(prev, v) {
* return v - prev;
* });
*
* // source :
* // --3---5------6---|->
* // diff:
* // --3---2------1---|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method diff
* @param {Object} seed
* A value to pass the `differ` function as previous on the inital notification from the source.
* @param {Function} differ
* Creates the difference, receives two params - the previous update and the current.
* It can be skipped - the default `differ` function returns array with two elements - the previous and the curren updates.
*/
diff: function(seed, differ) {
var last = seed,
fn = differ ? differ : function (last, next) {
return [last, next];
};
return this.fromLambda(function (stream, event) {
if (event === ProAct.Actor.close) {
stream.close();
return;
}
if (last === undefined) {
last = event;
return;
}
stream.trigger(differ(last, event));
last = event;
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It takes the first `limit` updates incoming from `this`.
*
* source : --3---4---5--4---3---4---5--|->
* skip(3): --3---4---5--|->
*
* @for ProAct.Actor
* @instance
* @method take
* @param {Number} limit The number of notifications to emit.
*/
take: function (limit) {
var left = limit;
return this.fromLambda(function (stream, event) {
left--;
if (left >= 0) {
stream.trigger(event, true);
}
if (left <= 0 && stream.state === ProAct.States.ready) {
stream.close();
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* It emits notifications from its source, while a condition is true.
*
* ```
*
* source.takeWhile(function (v) {
* return v % 2 === 1;
* });
*
* // source :
* // --3---5---2--4---3---4---5--|->
* // takeWhile:
* // --3---5--|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method takeWhile
* @param {Function} condition
* A condition function, which is called for each of the incoming values
* While it returns true, the elements are emitted,
* after it returns false for the first time, the stream created by takeWhile closes.
*/
takeWhile: function (condition) {
return this.fromLambda(function (stream, event) {
if (condition.call(null, event)) {
stream.trigger(event);
} else {
stream.close();
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* The logic of the stream is implemented through the passed `lambda` parameter.
*
* TODO The first parameter of the lambda should be called something else and not stream.
*
* ```
* source.fromLambda(function (stream, notification) {
* stream.trigger(notification);
* });
*
* // Just forwards notifications..
*
* ```
*
* @for ProAct.Actor
* @instance
* @method fromLambda
* @param {Function} lambda
* A function, with two arguments - the returned by this function stream and notification.
* For every update comming from `this`, the lambda is called with the update and the stream in it.
* Has the `trigger`, `triggerErr` and `triggerClose` methods.
*/
fromLambda: function (lambda) {
var stream = new ProAct.Stream(this.queueName),
listener = function (e) {
stream.trigger = StreamUtil.trigger;
stream.triggerErr = StreamUtil.triggerErr;
stream.triggerClose = StreamUtil.triggerClose;
lambda.call(null, stream, e);
stream.trigger = undefined;
stream.triggerErr = undefined;
stream.triggerClose = undefined;
};
this.onAll(listener);
stream.lambda = listener;
return stream;
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* For every update incomming from the source, a new `Actor` is created using the `mapper`
* function. All the updates, emitted by the streams, returned by the `mapper` are emitted by the
* `Actor` created by `flatMap`
*
*
* ```
* source.flatMap(function (v) {
* return ProAct.seq(100, [v, v +1 ]);
* });
*
* // source:
* // -1---2----4-----3-----2-----1---->
* // flatMap
* // -1-2-2-3--4-5---3-4---2-3---1-2-->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method flatMap
* @param {Function} mapper
* A function that returns an `ProAct.Actor` using the incomming notification.
*/
flatMap: function (mapper) {
return this.fromLambda(function (stream, e) {
if (e !== P.Actor.Close) {
var actor = mapper ? mapper.call(null, e) : e;
stream.into(actor);
}
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* For every update incomming from the source, a new `Actor` is created using the `mapper`
* function. ALl the updates, emitted by the streams, returned by the `mapper` are emitted by the
* `Actor` created by `flatMap`. The number of the currently active sources is limited by the
* passed `limit`. All the sources created after the limit is reached are queued and reused as sources later.
*
*
* @for ProAct.Actor
* @instance
* @method flatMapLimited
* @param {Function} mapper
* A function that returns an `ProAct.Actor` using the incomming notification.
* @param {Number} limit
* The number of the currently active sources.
*/
flatMapLimited: function (mapper, limit) {
var queue = [], current = 0, addActor = function (stream, actor) {
if (!actor) {
return;
}
if (current < limit) {
current++;
stream.into(actor);
actor.onClose(function () {
current--;
actor.offAll(stream.makeListener());
addActor(stream, queue.shift());
});
} else {
queue.push(actor);
}
};
return this.fromLambda(function (stream, e) {
var actor = mapper ? mapper.call(null, e) : e;
addActor(stream, actor);
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* For every update comming from `this`, a new `ProAct.Actor` is created using the logic
* passed through `mapper`. This new `Actor` becomes the current source of the `ProAct.Stream`,
* returned by this method. The next update will create a new source, which will become
* the current one and replace the old one. This is the same as {{#crossLink "ProAct.Actor/flatMapLimited:method"}}{{/crossLink}},
* with `limit` of `1`.
*
* ```
* source.flatMapLast(function (v) {
* return ProAct.seq(100, [v, v + 1, v + 2, v + 3]);
* });
*
* // source:
* // -1---2----4-----3-----2-----1----|->
* // flatMapLast
* // -1-2-2-3-44-5-6-3-4-5-2-3-4-1-2-3-4-|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method flatMapLast
* @param {Function} mapper
* A function that returns an `ProAct.Actor` using the incomming notification.
*/
flatMapLast: function (mapper) {
var oldActor;
return this.fromLambda(function (stream, e) {
var actor = mapper ? mapper.call(null, e) : e;
if (oldActor) {
oldActor.offAll(stream.makeListener());
}
oldActor = actor;
stream.into(actor);
});
},
/**
* Creates a new {{#crossLink "ProAct.Stream"}}{{/crossLink}} with source - `this`.
* For every update comming from `this`, a new `ProAct.Actor` is created using the logic
* passed through `mapper`. The first such `Actor` becomes the source of the `Actor`, returned by this
* method. When it finishes, if a new `Actor` is emitted, it becomes the source.
*
* ```
* source.flatMapLast(function (v) {
* return ProAct.seq(100, [v, v + 1, v + 2, v + 3]);
* });
*
* // source:
* // -1---2----4-----3-----2-----1----|->
* // flatMapFirst
* // -1-2-3-4--4-5-6-7-----2-3-4-5-|->
*
* ```
*
* @for ProAct.Actor
* @instance
* @method flatMapFirst
* @param {Function} mapper
* A function that returns an `ProAct.Actor` using the incomming notification.
*/
flatMapFirst: function (mapper) {
var oldActor;
return this.fromLambda(function (stream, e) {
if (oldActor && oldActor.state !== ProAct.States.closed) {
return;
}
var actor = mapper ? mapper.call(null, e) : e;
if (oldActor) {
oldActor.offAll(stream.makeListener());
}
oldActor = actor;
stream.into(actor);
});
}
});
P.S.prototype.t = P.S.prototype.trigger;
P.S.prototype.tt = P.S.prototype.triggerMany;
proact.js