VimUnDoNô]sɥT~/#SqG6!{KKKKTE_ Tz/function ReceivedChannelValue(id, err, value) {5_ Tz.function eceivedChannelValue(id, err, value) {5_ Tz-function ceivedChannelValue(id, err, value) {5_ Tz,function eivedChannelValue(id, err, value) {5_ Tz+function ivedChannelValue(id, err, value) {5_ Tz*function vedChannelValue(id, err, value) {5_ Tz)function edChannelValue(id, err, value) {5_  Tz(function dChannelValue(id, err, value) {5_ Tz'function ChannelValue(id, err, value) {5_ Tz&function ChannelValue(d, err, value) {5_ Tz%function ChannelValue(, err, value) {5_ Tz this.id = id;5_  Tz this.id = d;5_ Tz this.id = ;5_ Tz this.id = chan;5_ Tz this.i = chan;5_Tz6ReceivedChannelValue.prototype.resolve = function () {5_Tz5eceivedChannelValue.prototype.resolve = function () {5_Tz4ceivedChannelValue.prototype.resolve = function () {5_Tz3eivedChannelValue.prototype.resolve = function () {5_Tz2ivedChannelValue.prototype.resolve = function () {5_Tz1vedChannelValue.prototype.resolve = function () {5_Tz0edChannelValue.prototype.resolve = function () {5_Tz/dChannelValue.prototype.resolve = function () {5_Tz; self.put(new ReceivedChannelValue(id, err, value));5_Tz: self.put(new eceivedChannelValue(id, err, value));5_Tz9 self.put(new ceivedChannelValue(id, err, value));5_Tz8 self.put(new eivedChannelValue(id, err, value));5_Tz7 self.put(new ivedChannelValue(id, err, value));5_Tz6 self.put(new vedChannelValue(id, err, value));5_ Tz5 self.put(new edChannelValue(id, err, value));5_! Tz4 self.put(new dChannelValue(id, err, value));5_ "!"Tz3 self.put(new ChannelValue(id, err, value));5_!#""Tz2 self.put(new ChannelValue(d, err, value));5_"$#"Tz1 self.put(new ChannelValue(, err, value));5_#%$Tz< self.fill(new ReceivedChannelValue(id, err, value));5_$&%Tz; self.fill(new eceivedChannelValue(id, err, value));5_%'&Tz: self.fill(new ceivedChannelValue(id, err, value));5_&('Tz9 self.fill(new eivedChannelValue(id, err, value));5_')(Tz8 self.fill(new ivedChannelValue(id, err, value));5_(*)Tz7 self.fill(new vedChannelValue(id, err, value));5_)+*Tz6 self.fill(new edChannelValue(id, err, value));5_*,+Tz5 self.fill(new dChannelValue(id, err, value));5_+-,#Tz4 self.fill(new ChannelValue(id, err, value));5_,.-#Tz3 self.fill(new ChannelValue(d, err, value));5_-/.#Tz2 self.fill(new ChannelValue(, err, value));5_.0/'Tz,Channel.prototype.resolver = function (id) {5_/10'Tz+Channel.prototype.resolver = function (d) {5_021&Tz+Channel.prototype.receive = function (id) {5_132&Tz*Channel.prototype.receive = function (d) {5_243* T )+z-function MergedChannelValue(ch, err, value) {5_354* T )+z,function ergedChannelValue(ch, err, value) {5_465* T )+z+function rgedChannelValue(ch, err, value) {5_576* T )+z*function gedChannelValue(ch, err, value) {5_687*T )*)function edChannelValue(ch, err, value) {5_798*T )* this.chan = ch;5_8:9*T )* this.err = err;5_9;:*T )* this.val = value;5_:<;*T )* return this;5_;=<*T )*}5_<>=*T )*5_=?>E TDFsL channel.put(new MergedChannelValue(ch, err, value), reader);5_>@?E TDFsK channel.put(new ergedChannelValue(ch, err, value), reader);5_?A@E TDFsJ channel.put(new rgedChannelValue(ch, err, value), reader);5_@BAE TDFsI channel.put(new gedChannelValue(ch, err, value), reader);5_ACBE TDFsH channel.put(new edChannelValue(ch, err, value), reader);5_BDCE TDFsG channel.put(new dChannelValue(ch, err, value), reader);5_CEDH TGIsD channel.put(new MergedChannelValue(ch, null, null));5_DFEH TGIsC channel.put(new ergedChannelValue(ch, null, null));5_EGFH TGIsB channel.put(new rgedChannelValue(ch, null, null));5_FHGH TGIsA channel.put(new gedChannelValue(ch, null, null));5_GIHH TGIs@ channel.put(new edChannelValue(ch, null, null));5_HJIH TGIs? channel.put(new dChannelValue(ch, null, null));5_IKJ>T#>At// >@s5_JKwTDu8// A channel is a queue with a read-end and a write-end. 0) {$ var q = this._queue.shift();) sendValue(q._value, q._callback);& sendValue(q._value, callback); } else {1 callback && this._pending.push(callback); }};Q// Places a value into the channel. The callback will be called when the value is// consumed from the read-end.4Channel.prototype.put = function (value, callback) {# if (this._pending.length > 0) {& var p = this._pending.shift();# sendValue(value, callback); sendValue(value, p); } else {3 this._queue.push(new CBV(callback, value)); }};*// Does any ending actions on the channel.*// The protocol is to have a channel "end"*// by a null value being placed on it. The1// end() method is simply to perform any pending 3// ending actions. The default action is to replace,// the end() function with the original end.(Channel.prototype.end = function end() { this.end = end;};8// Returns a channel that will give you values that come>// on this channel, without actually reading from the channel.:// That is, multiple taps on a channel will get its values9// fanned out. If a channel argument is given, the tapped$// values will go into that channel.)Channel.prototype.tap = function (chan) {( var tapChan = chan || new Channel(); var self = this; if (!this._taps) { this._taps = [tapChan]; var put = this.put;/ this.put = function (value, callback) {B for (var c = 0, cN = this._taps.length; c < cN; ++c) {) this._taps[c].put(value); }! if (value === null) {/ while (this._taps.length > 0) {( this._taps[0].end(); }" this._taps = null; this.put = put; }+ if (this._pending.length > 0) {: // Put only if there are takers. Otherwise< // just drop the value. If we don't do this,= // the value will simply pile up if only taps1 // are being used on the channel.0 put.call(this, value, callback); } }; } else {! this._taps.push(tapChan); } var end = tapChan.end; tapChan.end = function () {: self._taps.splice(self._taps.indexOf(tapChan), 1); end.call(this); }; return tapChan;};.// For an end-point channel, applies the given.// function to values received on the channel.4// The second argument to the function is a callback;// that should be called once the processing has completed.4// It is alright to call the callback synchronously.6// It only makes sense to have one processing function8// for a channel. The fn is called with the value as the5// first argument and a loop continuation callback as// the second argument.+Channel.prototype.process = function (fn) { var self = this;" function receive(err, value) { fn(value, loop); } function loop(err) { if (!err) { self.take(receive); } } loop(null); return this;};9// Binds a channel to the given named method of the given>// class. If the class has an init() method, it will be called4// with `options.initArgs` to instantiate an object.B// The given `options.methodName` of the resultant object will be E// invoked with the message as the first argument, and a continuationE// callback (a la `process()`) as the second argument. The methodName// defaults to `receive`.//D// If `options.spawn` is `true`, then the handler is called with the;// message only and the channel returns to processing other;// messages immediately without waiting for the handling to // finish.//J// Calling bind on an already bound channel replaces the previous binding.4Channel.prototype.bind = function (klass, options) {# var self = this, receive, loop; self._boundClass = klass;I self._boundMethodName = (options && options.methodName) || 'receive';> self._boundInitArgs = (options && options.initArgs) || [];; self._boundSpawn = (options && options.spawn) || false; if (!self._bound) {' receive = function (err, msg) {R var handler = new self._boundClass(); // new is not expected to throw. if (handler.init) { try {O handler = handler.init.apply(handler, self._boundInitArgs); } catch (e) {% return loop(err); } }# if (self._boundSpawn) { nextTick(loop);4 handler[self._boundMethodName](msg); } else {: handler[self._boundMethodName](msg, loop); } }; loop = function (err) { if (!err) {# self.take(receive); } else {$ self._bound = false; } }; self._bound = true; loop(null); } return this;};)function ChannelValue(chan, err, value) { this.chan = chan; this.err = err; this.val = value; return this;}.ChannelValue.prototype.resolve = function () { if (this.err) { throw this.err; } else { return this.val; }};;// Makes a callback that will receive the value produced by:// some process and place the result into the channel. The7// "id" exists to identify the one producing the value.5// The "id", "err" and "val" are all available on the // channel.)Channel.prototype.receive = function () { var self = this;" return function (err, value) {5 self.put(new ChannelValue(self, err, value)); };};:// Like receive, but results in the channel being 'filled'8// with the value received on the callback. Once a value8// is received, all subsequent take operations will give4// the same value, and puts will result in an error.*Channel.prototype.resolver = function () { var self = this;" return function (err, value) {6 self.fill(new ChannelValue(self, err, value)); };};+// Answers "will read succeed immediately?")Channel.prototype.canRead = function () {@ return this._queue.length > 0 && this._pending.length === 0;};,// Answers "will write succeed immediately?"*Channel.prototype.canWrite = function () {@ return this._pending.length > 0 || this._queue.length === 0;};?// Answers "how many values have been placed into the channel?"B// Positive values give the number of values available right away.>// Negative values give the number of pending take operations.)Channel.prototype.backlog = function () {5 return this._queue.length - this._pending.length;};@// Makes a new channel whose values are transformed by the given>// function "f". `cond(value)` is a function that specifies a >// condition until which the mapping will continue. The mapper// is not expected to throw.&Channel.prototype.map = function (f) {" var ch2 = Object.create(this); var take = this.take;$ ch2.take = function (callback) {/ take.call(this, function (err, value) {1 callback(err, err ? null : f(value)); }); }; return ch2;};3// Makes a new channel and pipes the values in this2// channel to it. Only the values that satisfy the.// predicate function 'f' are piped and others3// are dropped. The filter function is not expected // to throw.)Channel.prototype.filter = function (f) {" var ch2 = Object.create(this); var take = this.take;$ ch2.take = function (callback) {/ take.call(this, function (err, value) { if (err) { % callback(err, null); " } else if (f(value)) {% callback(err, value); } else {4 ch2.take(callback); // Value dropped } }); }; return ch2;};3// Makes a new channel, reduces the values produced2// continuously and sends the output to the taker.(// The reducer is not expected to throw.2Channel.prototype.reduce = function (initial, f) {" var ch2 = Object.create(this); var take = this.take; var result = initial;$ ch2.take = function (callback) {/ take.call(this, function (err, value) { if (err) {$ callback(err, null); } else {* result = f(result, value);' callback(null, result); } }); }; return ch2;};9// Makes a new channel and pipes the values put into this// channel in groups of N. (Channel.prototype.group = function (N) { if (N <= 0) {S throw new Error('Groups need to be at least 1 in size. Given "' + N + '"'); }4 return this.reduce([], function (group, value) {K return (group.length === N) ? [value] : (group.push(value), group);7 }).filter(function (g) { return g.length === N; });};.function resolve(thing, recursive, callback) { var unresolved = 0;# if (thing instanceof Channel) {A unresolved += resolveChannel(thing, recursive, callback);( } else if (thing instanceof Array) {? unresolved += resolveArray(thing, recursive, callback);) } else if (thing instanceof Object) {@ unresolved += resolveObject(thing, recursive, callback); } else {# sendValue(thing, callback); } return unresolved;}7function resolveChannel(channel, recursive, callback) { if (recursive) {4 channel.take(function receiver(err, value) {8 Channel.resolve(value, recursive, callback); }); } else { channel.take(callback); } return 1;}1function resolveArray(arr, recursive, callback) { var unresolved = 0;* for (var i = 0; i < arr.length; ++i) {@ unresolved += resolve(arr[i], recursive, (function (i) {2 return function receiver(err, value) { if (recursive) {8 resolve(value, recursive, receiver); } else {# arr[i] = value;! --unresolved;+ if (unresolved === 0) {, callback(null, arr); } } }; }(i))); } return unresolved;}2function resolveObject(obj, recursive, callback) { unresolved = 0;+ Object.keys(obj).forEach(function (k) {P unresolved += resolve(obj[k], recursive, function receiver(err, value) { if (recursive) {4 resolve(value, recursive, receiver); } else { obj[k] = value; --unresolved;' if (unresolved === 0) {( callback(null, obj); } } }); }); return unresolved;}// back when all entries have been resolved. If 'recursive' isE// true, then if the value received on a channel is itself a channel,6// it is recursively waited on until final resolution.Channel.resolve = resolve;;// Temporarily switches the channel to a mode where it will:// collect the next N items into a group and pass it on to// the callback.//// Use within task like this -// var ch = new Channel(); // ...// x <- ch.takeN(10);2Channel.prototype.takeN = function (N, callback) { var group = []; var self = this;" function receive(err, value) { if (err) {, return sendError(err, callback); } if (value !== null) { group.push(value);# if (group.length < N) {# self.take(receive); } else {+ sendValue(group, callback); } } else {' sendValue(group, callback); } } self.take(receive);};3// Takes as many values as it can without blocking.2Channel.prototype.takeSome = function (callback) { var bl = this.backlog(); if (bl > 0) {( return this.takeN(bl, callback); } sendValue([], callback);};,// Keeps this channel alive until a value is // received from the given chan.+Channel.prototype.until = function (chan) { var done = false; var self = this; var tapChan = chan.tap();( tapChan.take(function (err, value) { done = true; tapChan.end(); self.end(); }); var ch = this.tap(); var take = ch.take;# ch.take = function (callback) { if (done) {& sendValue(null, callback); } else {& take.call(this, callback); } }; return ch;};function noop() {}8// Switches the channel to a state where every time some5// reader takes a value from the channel, they'll get6// `value` delivered immediately. This makes a channel5// behave somewhat like a promise, where until `fill`7// is called, asking for a value will cause a wait, but6// once `fill` is called somewhere, `take` will always// succeed with a single value.+Channel.prototype.fill = function (value) { if (this.backlog() > 0) {[ throw new Error('Channel::fill cannot be used after Channel::put has been called'); } var origPut = this.put;% this.take = function (callback) {# sendValue(value, callback); };2 this.put = function (ignoredValue, callback) {& sendError('filled', callback); }; this.fill = noop;2 // If takers are already waiting, satisfy them // immediately. while (this.backlog() < 0) {" origPut.call(this, value); } return this;};3// Sends the elements of the given array one by one0// to the channel as readers request values. The1// callback will be called when the last value is // accepted.7Channel.prototype.stream = function (array, callback) { var i = 0, self = this; function next() { if (i < array.length) {' self.put(array[i++], next); } else {' sendValue(array, callback); } } next();};:// Sets up the channel to receive events of the given type:// from the given domElement. (Works only in the browser.);// `domElement` can either be a string which is taken to be7// a querySelector specifier, an array of DOM nodes, or:// a single DOM node. `eventName` is a string like 'click',// which gives an event category to bind to.//4// Note: If you want a channel to not receive events5// too frequently, you can first debounce the channel+// before listening for events, like this -//// ch = new Channel();7// ch.debounce(100).listen('.textarea', 'change');//9// The above code will make sure that consecutive change 9// events are separated by at least 100ms. The debounce()5// method call produces a wrapper channel object that6// acts as a gatekeeper to the original channel object9// 'ch'. So, while the above way will result in debounced6// actions, you can subsequently call `ch.listen()` to9// bypass debouncing on the same channel. Readers reading7// `ch` will receive events from the debounced elements,// as well from the elements bound directly.=Channel.prototype.listen = function (domElement, eventName) { var self = this; var elements = null;) if (typeof domElement === 'string') {9 elements = document.querySelectorAll(domElement);# } else if (domElement.length) { elements = domElement; } else { elements = [domElement]; } function listener(event) { self.put(event); event.stopPropagation(); }6 for (var i = 0, N = elements.length; i < N; ++i) {: elements[i].addEventListener(eventName, listener); } var end = this.end; this.end = function () {: for (var i = 0, N = elements.length; i < N; ++i) {A elements[i].removeEventListener(eventName, listener); } end.call(this); }; return this;};8// Makes a new channel that receives the values put into:// all the given channels (which is an array of channels).=// The value produced by a merged channel is a wrapper object:// that has three fields - "chan" giving the channel that 7// produced the value, "val" giving the value and "ix" =// giving the index of the channel in the array that produced?// the value. The merged channel will received a wrapped object;// that will pass on both values as well as errors from theD// channels being merged. This permits custom error handling instead// channel's error. Not all errors and channels need be equal.//// for this change is that now the "piper" function is exposedA// as the .add() method of the merged channel, to enable addition=// of new channels to the merged stream on the fly. To remove;// a channel from a merged stream, simply send a null value // to it.//;// Breaking change: MergedChannelValue is now ChannelValue.%Channel.merge = function (channels) { var channel = new Channel(); function piper(ch) {% function writer(err, value) {! if (value !== null) {F channel.put(new ChannelValue(ch, err, value), reader); } else {V // Indicate that the channel is finished. The reader can discard this.> channel.put(new ChannelValue(ch, null, null)); } }% function reader(err, value) { ch.take(writer); } reader(null, true); } channel.add = piper;( channels && channels.forEach(piper); return channel;};6// It is sometimes useful to also have a value sent to7// an existing channel after a timeout expires. If some4// other process is supposed to write a value to the6// channel and it is taking too long, the value passed9// to the .timeout() call can be tested against to decide6// whether a timeout occurred before the process could// do its thing.2Channel.prototype.timeout = function (ms, value) {- setTimeout(timeoutTick, ms, this, value); return this;};6// Makes a "timeout" channel, which'll deliver a value1// a given interval after the channel is created.(Channel.timeout = function (ms, value) {. return (new Channel()).timeout(ms, value);};&function timeoutTick(channel, value) { channel.put(value);}// `stop()` method is called on the channel. Calling `start()`6// will have an effect only when the clock is stopped.Channel.clock = function (ms) { var channel = new Channel(); channel._timer = null;" channel._timeInterval_ms = ms; channel._counter = 0; channel.start = clockStart; channel.stop = clockStop; return channel;};function clockTick(clock) { clock.put(clock._counter++);}#function clockStart(startCounter) { if (!this._timer) {? startCounter = arguments.length < 1 ? 1 : startCounter;% this._counter = startCounter;J this._timer = setInterval(clockTick, this._timeInterval_ms, this); }}function clockStop() { if (this._timer) {# clearInterval(this._timer); this._timer = null; }}4// Returns a wrapped interface to channel which will2// debounce the values placed on it - i.e. it will3// reject put() operations that occur within a time+// of "ms" milliseconds between each other.,Channel.prototype.debounce = function (ms) {! var ch = Object.create(this); ch._channel = this;! ch._debounceInterval_ms = ms; ch._timer = null; ch.put = debouncingPut; return ch;};'function realPut(ch, value, callback) { ch._timer = null;% ch._channel.put(value, callback);})function debouncingPut(value, callback) { if (this._timer) {" clearTimeout(this._timer); this._timer = null; }X this._timer = setTimeout(realPut, this._debounceInterval_ms, this, value, callback); return this;}1// Wraps the given channel with an interface such1// that put() operations will immediately succeed2// as long as fewer than N values have been placed// on the channel.)Channel.prototype.buffer = function (N) {! var ch = Object.create(this); ch._channel = this; ch._bufferLength = N; ch.put = bufferedPut; ch.take = bufferedTake; return ch;};'function bufferedPut(value, callback) {. if (this.backlog() < this._bufferLength) {! this._channel.put(value);# sendValue(value, callback); } else {+ this._channel.put(value, callback); }}!function bufferedTake(callback) {! this._channel.take(callback);/ if (this.backlog() >= this._bufferLength) {4 var q = this._queue[this._bufferLength - 1];) sendValue(q._value, q._callback); q._callback = null; }}>// Every time a bucket's level falls below the low water mark,>// it waits for the bucket to get full again before delivering@// values to the takers. This is useful when values are expected@// to arrive at a channel roughly periodically, but the rate at ?// which they get processed can fluctuate a bit. The buffering C// helps with the fluctuation and the "low water mark" helps ensure// maintenance of the buffer.>Channel.prototype.bucket = function (fullSize, lowWaterMark) {! var ch = Object.create(this); ch._channel = this; ch._bufferLength = fullSize;/ ch._bucketLowWaterMark = lowWaterMark || 0; ch._suspendedTakes = []; ch.waitingTillFull = true; ch.take = bucketTake; ch.put = bucketPut; return ch;};+function bucketProcSuspendedTakes(bucket) {/ while (bucket._suspendedTakes.length > 0) {B bufferedTake.call(bucket, bucket._suspendedTakes.shift()); }L bucket.waitingTillFull = bucket.backlog() <= bucket._bucketLowWaterMark;}function bucketTake(callback) { if (this.waitingTillFull) {2 if (this.backlog() > this._bufferLength) { // Full reached.) this.waitingTillFull = false; this.take(callback); } else {0 this._suspendedTakes.push(callback); } } else {, this._suspendedTakes.push(callback);' bucketProcSuspendedTakes(this); }}%function bucketPut(value, callback) {, bufferedPut.call(this, value, callback); if (this.waitingTillFull) {2 if (this.backlog() > this._bufferLength) { // Full reached.+ bucketProcSuspendedTakes(this); } } else {' bucketProcSuspendedTakes(this); }}8// If more than N values have been placed into a channel8// and a writer tries to place one more value, sometimes4// we want the new value to be dropped in order that4// processing requirements don't accumulate. This is3// the purpose of `droppingBuffer` which wraps the .// parent channel's `put` to do this dropping.//9// A channel with a droppingBuffer will never block a put // operation.1Channel.prototype.droppingBuffer = function (N) {! var ch = Object.create(this); ch._channel = this; ch._bufferLength = N; ch.put = droppingPut; return ch;};'function droppingPut(value, callback) {. if (this.backlog() < this._bufferLength) {! this._channel.put(value);# sendValue(value, callback); } else { // Drop the value." sendValue(null, callback); }}2// In the same situation as with `droppingBuffer`,1// at other times, we want the more recent values1// to take precedence over the values already in 1// the queue. In this case, we want to expire the/// old values and replace them with new values.&// That is what `expiringBuffer` does.//7// A channel with an expiringBuffer will never block a // put operation.1Channel.prototype.expiringBuffer = function (N) {! var ch = Object.create(this); ch._channel = this; ch._bufferLength = N; ch.put = expiringPut; return ch;};'function expiringPut(value, callback) {2 while (this.backlog() >= this._bufferLength) { this.take(); } this._channel.put(value); sendValue(value, callback); return this;}8// Makes a "fanout" channel that can be "connect()"ed to>// other channels to whom the values that come on this channel?// will be copied. Do not call a fanout channel's "take" method>// explicitly. Instead connect other channels to it to receive@// values. Since it may take time to setup connections, you have>// to call ch.start() explicitly to begin piping values to the0// connections, lest some values get missed out.(Channel.prototype.fanout = function () {! var ch = Object.create(this);$ ch.connect = fanoutConnect;' ch.disconnect = fanoutDisconnect;" ch.start = fanoutStart; ch._channel = this; ch._connections = []; ch._started = false; return ch;};function fanoutConnect() {7 for (var i = 0, N = arguments.length; i < N; ++i) {& this.disconnect(arguments[i]);- this._connections.push(arguments[i]); } return this;}function fanoutDisconnect() {! if (arguments.length === 0) { this._connections = []; return this; } var N, i, chan, pos;3 for (i = 0, N = arguments.length; i < N; ++i) { chan = arguments[i];. pos = this._connections.indexOf(chan); if (pos >= 0) {- this._connections.splice(pos, 1); } } return this;}function fanoutStart() { var self = this; if (!self._started) { self._started = true;0 self.take(function receive(err, value) {! if (value !== null) {K for (var i = 0, N = self._connections.length; i < N; ++i) {4 self._connections[i].put(value); }# self.take(receive); } }); } return self;}module.exports = Channel;5