xstream - npm Package Compare versions

Comparing version 5.0.3 to 5.0.4

CHANGELOG.md

@@ -0,1 +1,11 @@
+<a name="5.0.4"></a>
+## [5.0.4](https://github.com/staltz/xstream/compare/v5.0.3...v5.0.4) (2016-06-14)
+
+
+### Bug Fixes
+
+* **imitate:** fix cyclic execution leaks ([8658aa0](https://github.com/staltz/xstream/commit/8658aa0)), closes [#51](https://github.com/staltz/xstream/issues/51)
+
+
+
 <a name="5.0.3"></a>
@@ -2,0 +12,0 @@ ## [5.0.3](https://github.com/staltz/xstream/compare/v5.0.2...v5.0.3) (2016-06-13)

core.d.ts

@@ -6,3 +6,3 @@ export interface InternalListener<T> {
 }
-export declare const emptyListener: InternalListener<any>;
+export declare const emptyIL: InternalListener<any>;
 export interface InternalProducer<T> {
@@ -12,11 +12,15 @@ _start: (listener: InternalListener<T>) => void;
 }
-export interface Operator<T, R> extends InternalProducer<R>, InternalListener<T> {
+export interface OutSender<T> {
+    out: Stream<T>;
+}
+export interface Operator<T, R> extends InternalProducer<R>, InternalListener<T>, OutSender<R> {
     type: string;
     ins: Stream<T>;
     _start: (out: Stream<R>) => void;
-    _stop: () => void;
-    _n: (v: T) => void;
-    _e: (err: any) => void;
-    _c: () => void;
 }
+export interface Aggregator<T, U> extends InternalProducer<U>, OutSender<U> {
+    type: string;
+    insArr: Array<Stream<T>>;
+    _start: (out: Stream<U>) => void;
+}
 export interface Producer<T> {
@@ -31,9 +35,9 @@ start: (listener: Listener<T>) => void;
 }
-export declare class MergeProducer<T> implements InternalProducer<T>, InternalListener<T> {
-    streams: Array<Stream<T>>;
+export declare class MergeProducer<T> implements Aggregator<T, T>, InternalListener<T> {
+    insArr: Array<Stream<T>>;
     type: string;
-    private out;
+    out: Stream<T>;
     private ac;
-    constructor(streams: Array<Stream<T>>);
-    _start(out: InternalListener<T>): void;
+    constructor(insArr: Array<Stream<T>>);
+    _start(out: Stream<T>): void;
     _stop(): void;
@@ -54,6 +58,7 @@ _n(t: T): void;
 }
-export declare class CombineListener<T> implements InternalListener<T> {
+export declare class CombineListener<T> implements InternalListener<T>, OutSender<Array<T>> {
     private i;
+    out: Stream<Array<T>>;
     private p;
-    constructor(i: number, p: CombineProducer<T>);
+    constructor(i: number, out: Stream<Array<T>>, p: CombineProducer<T>);
     _n(t: T): void;
@@ -63,6 +68,6 @@ _e(err: any): void;
 }
-export declare class CombineProducer<R> implements InternalProducer<Array<R>> {
-    s: Array<Stream<any>>;
+export declare class CombineProducer<R> implements Aggregator<any, Array<R>> {
+    insArr: Array<Stream<any>>;
     type: string;
-    out: InternalListener<Array<R>>;
+    out: Stream<Array<R>>;
     ils: Array<CombineListener<any>>;
@@ -72,5 +77,5 @@ Nc: number;
     vals: Array<R>;
-    constructor(s: Array<Stream<any>>);
+    constructor(insArr: Array<Stream<any>>);
     up(t: any, i: number): boolean;
-    _start(out: InternalListener<Array<R>>): void;
+    _start(out: Stream<Array<R>>): void;
     _stop(): void;
@@ -105,3 +110,3 @@ }
     type: string;
-    private out;
+    out: Stream<T>;
     private s;
@@ -120,3 +125,3 @@ private l;
     type: string;
-    private out;
+    out: Stream<T>;
     private dropped;
@@ -134,3 +139,3 @@ constructor(max: number, ins: Stream<T>);
     type: string;
-    private out;
+    out: Stream<T>;
     private oil;
@@ -149,3 +154,3 @@ constructor(o: Stream<any>, ins: Stream<T>);
     type: string;
-    private out;
+    out: Stream<T>;
     constructor(passes: (t: T) => boolean, ins: Stream<T>);
@@ -164,3 +169,3 @@ _start(out: Stream<T>): void;
     private open;
-    private out;
+    out: Stream<T>;
     constructor(ins: Stream<Stream<T>>);
@@ -179,3 +184,3 @@ _start(out: Stream<T>): void;
     type: string;
-    private out;
+    out: Stream<R>;
     private acc;
@@ -192,3 +197,3 @@ constructor(f: (acc: R, t: T) => R, seed: R, ins: Stream<T>);
     type: string;
-    private out;
+    out: Stream<T>;
     private has;
@@ -210,3 +215,3 @@ private val;
     private open;
-    private out;
+    out: Stream<R>;
     constructor(mapOp: MapOperator<T, Stream<R>>);
@@ -224,3 +229,3 @@ _start(out: Stream<R>): void;
     type: string;
-    protected out: Stream<R>;
+    out: Stream<R>;
     constructor(project: (t: T) => R, ins: Stream<T>);
@@ -243,3 +248,3 @@ _start(out: Stream<R>): void;
     type: string;
-    private out;
+    out: Stream<T>;
     constructor(fn: (err: any) => Stream<T>, ins: Stream<T>);
@@ -265,3 +270,3 @@ _start(out: Stream<T>): void;
     type: string;
-    private out;
+    out: Stream<T>;
     private taken;
@@ -277,3 +282,2 @@ constructor(max: number, ins: Stream<T>);
     protected _ils: Array<InternalListener<T>>;
-    protected _hil: InternalListener<T>;
     protected _stopID: any;
@@ -301,3 +305,4 @@ protected _prod: InternalProducer<T>;
     _remove(il: InternalListener<T>): void;
-    _setHIL(il: InternalListener<T>): void;
+    _pruneCycles(): void;
+    _onlyReachesThis(x: InternalListener<any>): boolean;
     private ctor();
@@ -304,0 +309,0 @@ /**

core.js

@@ -17,3 +17,3 @@ "use strict";
 }
-exports.emptyListener = {
+exports.emptyIL = {
     _n: noop,
@@ -45,11 +45,11 @@ _e: noop,
 var MergeProducer = (function () {
-    function MergeProducer(streams) {
-        this.streams = streams;
+    function MergeProducer(insArr) {
+        this.insArr = insArr;
         this.type = 'merge';
-        this.out = exports.emptyListener;
-        this.ac = streams.length;
+        this.out = null;
+        this.ac = insArr.length;
     }
     MergeProducer.prototype._start = function (out) {
         this.out = out;
-        var s = this.streams;
+        var s = this.insArr;
         var L = s.length;
@@ -61,3 +61,3 @@ for (var i = 0; i < L; i++) {
     MergeProducer.prototype._stop = function () {
-        var s = this.streams;
+        var s = this.insArr;
         var L = s.length;
@@ -94,4 +94,5 @@ for (var i = 0; i < L; i++) {
 var CombineListener = (function () {
-    function CombineListener(i, p) {
+    function CombineListener(i, out, p) {
         this.i = i;
+        this.out = out;
         this.p = p;
@@ -101,3 +102,3 @@ p.ils.push(this);
     CombineListener.prototype._n = function (t) {
-        var p = this.p, out = p.out;
+        var p = this.p, out = this.out;
         if (!out)
@@ -110,3 +111,3 @@ return;
     CombineListener.prototype._e = function (err) {
-        var out = this.p.out;
+        var out = this.out;
         if (!out)
@@ -128,8 +129,8 @@ return;
 var CombineProducer = (function () {
-    function CombineProducer(s) {
-        this.s = s;
+    function CombineProducer(insArr) {
+        this.insArr = insArr;
         this.type = 'combine';
         this.out = null;
         this.ils = [];
-        var n = this.Nc = this.Nn = s.length;
+        var n = this.Nc = this.Nn = insArr.length;
         var vals = this.vals = new Array(n);
@@ -148,3 +149,3 @@ for (var i = 0; i < n; i++) {
         this.out = out;
-        var s = this.s;
+        var s = this.insArr;
         var n = s.length;
@@ -157,3 +158,3 @@ if (n === 0) {
             for (var i = 0; i < n; i++) {
-                s[i]._add(new CombineListener(i, this));
+                s[i]._add(new CombineListener(i, out, this));
             }
@@ -163,3 +164,3 @@ }
     CombineProducer.prototype._stop = function () {
-        var s = this.s;
+        var s = this.insArr;
         var n = this.Nc = this.Nn = s.length;
@@ -361,3 +362,3 @@ var vals = this.vals = new Array(n);
         this.out = null;
-        this.oil = exports.emptyListener; // oil = other InternalListener
+        this.oil = exports.emptyIL; // oil = other InternalListener
     }
@@ -772,3 +773,3 @@ EndWhenOperator.prototype._start = function (out) {
         this.type = 'startWith';
-        this.out = exports.emptyListener;
+        this.out = exports.emptyIL;
     }
@@ -836,3 +837,2 @@ StartWithOperator.prototype._start = function (out) {
         this._ils = [];
-        this._hil = null;
         this._target = null;
@@ -850,5 +850,2 @@ }
         }
-        var h = this._hil;
-        if (h)
-            h._n(t);
     };
@@ -865,5 +862,2 @@ Stream.prototype._e = function (err) {
         }
-        var h = this._hil;
-        if (h)
-            h._e(err);
         this._x();
@@ -881,5 +875,2 @@ };
         }
-        var h = this._hil;
-        if (h)
-            h._c();
         this._x();
@@ -921,5 +912,4 @@ };
         var ta = this._target;
-        if (ta && ta._ils.length === 0) {
+        if (ta)
             return ta._add(il);
-        }
         var a = this._ils;
@@ -938,2 +928,5 @@ a.push(il);
     Stream.prototype._remove = function (il) {
+        var ta = this._target;
+        if (ta)
+            return ta._remove(il);
         var a = this._ils;
@@ -947,9 +940,36 @@ var i = a.indexOf(il);
             }
+            else if (a.length === 1) {
+                this._pruneCycles();
+            }
         }
-        else if (this._target) {
-            this._target._remove(il);
+    };
+    // If all paths stemming from `this` stream eventually end at `this`
+    // stream, then we remove the single listener of `this` stream, to
+    // force it to end its execution and dispose resources. This method
+    // assumes as a precondition that this._ils has just one listener.
+    Stream.prototype._pruneCycles = function () {
+        if (this._onlyReachesThis(this)) {
+            this._remove(this._ils[0]);
         }
     };
-    Stream.prototype._setHIL = function (il) {
-        this._hil = il;
+    // Checks whether *all* paths starting from `x` will eventually end at
+    // `this` stream, on the stream graph, following edges A->B where B is
+    // a listener of A.
+    Stream.prototype._onlyReachesThis = function (x) {
+        if (x.out === this) {
+            return true;
+        }
+        else if (x.out) {
+            return this._onlyReachesThis(x.out);
+        }
+        else if (x._ils) {
+            for (var i = 0, N = x._ils.length; i < N; i++) {
+                if (!this._onlyReachesThis(x._ils[i]))
+                    return false;
+            }
+            return true;
+        }
+        else {
+            return false;
+        }
     };
@@ -1563,3 +1583,2 @@ Stream.prototype.ctor = function () {
         this._target = target;
-        target._setHIL(this);
     };
@@ -1566,0 +1585,0 @@ /**

dist/xstream.js

@@ -18,3 +18,3 @@ (function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.xstream = f()}})(function(){var define,module,exports;return (function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s})({1:[function(require,module,exports){
 }
-exports.emptyListener = {
+exports.emptyIL = {
     _n: noop,
@@ -46,11 +46,11 @@ _e: noop,
 var MergeProducer = (function () {
-    function MergeProducer(streams) {
-        this.streams = streams;
+    function MergeProducer(insArr) {
+        this.insArr = insArr;
         this.type = 'merge';
-        this.out = exports.emptyListener;
-        this.ac = streams.length;
+        this.out = null;
+        this.ac = insArr.length;
     }
     MergeProducer.prototype._start = function (out) {
         this.out = out;
-        var s = this.streams;
+        var s = this.insArr;
         var L = s.length;
@@ -62,3 +62,3 @@ for (var i = 0; i < L; i++) {
     MergeProducer.prototype._stop = function () {
-        var s = this.streams;
+        var s = this.insArr;
         var L = s.length;
@@ -95,4 +95,5 @@ for (var i = 0; i < L; i++) {
 var CombineListener = (function () {
-    function CombineListener(i, p) {
+    function CombineListener(i, out, p) {
         this.i = i;
+        this.out = out;
         this.p = p;
@@ -102,3 +103,3 @@ p.ils.push(this);
     CombineListener.prototype._n = function (t) {
-        var p = this.p, out = p.out;
+        var p = this.p, out = this.out;
         if (!out)
@@ -111,3 +112,3 @@ return;
     CombineListener.prototype._e = function (err) {
-        var out = this.p.out;
+        var out = this.out;
         if (!out)
@@ -129,8 +130,8 @@ return;
 var CombineProducer = (function () {
-    function CombineProducer(s) {
-        this.s = s;
+    function CombineProducer(insArr) {
+        this.insArr = insArr;
         this.type = 'combine';
         this.out = null;
         this.ils = [];
-        var n = this.Nc = this.Nn = s.length;
+        var n = this.Nc = this.Nn = insArr.length;
         var vals = this.vals = new Array(n);
@@ -149,3 +150,3 @@ for (var i = 0; i < n; i++) {
         this.out = out;
-        var s = this.s;
+        var s = this.insArr;
         var n = s.length;
@@ -158,3 +159,3 @@ if (n === 0) {
             for (var i = 0; i < n; i++) {
-                s[i]._add(new CombineListener(i, this));
+                s[i]._add(new CombineListener(i, out, this));
             }
@@ -164,3 +165,3 @@ }
     CombineProducer.prototype._stop = function () {
-        var s = this.s;
+        var s = this.insArr;
         var n = this.Nc = this.Nn = s.length;
@@ -362,3 +363,3 @@ var vals = this.vals = new Array(n);
         this.out = null;
-        this.oil = exports.emptyListener; 
+        this.oil = exports.emptyIL; 
     }
@@ -773,3 +774,3 @@ EndWhenOperator.prototype._start = function (out) {
         this.type = 'startWith';
-        this.out = exports.emptyListener;
+        this.out = exports.emptyIL;
     }
@@ -837,3 +838,2 @@ StartWithOperator.prototype._start = function (out) {
         this._ils = [];
-        this._hil = null;
         this._target = null;
@@ -851,5 +851,2 @@ }
         }
-        var h = this._hil;
-        if (h)
-            h._n(t);
     };
@@ -866,5 +863,2 @@ Stream.prototype._e = function (err) {
         }
-        var h = this._hil;
-        if (h)
-            h._e(err);
         this._x();
@@ -882,5 +876,2 @@ };
         }
-        var h = this._hil;
-        if (h)
-            h._c();
         this._x();
@@ -914,5 +905,4 @@ };
         var ta = this._target;
-        if (ta && ta._ils.length === 0) {
+        if (ta)
             return ta._add(il);
-        }
         var a = this._ils;
@@ -931,2 +921,5 @@ a.push(il);
     Stream.prototype._remove = function (il) {
+        var ta = this._target;
+        if (ta)
+            return ta._remove(il);
         var a = this._ils;
@@ -940,9 +933,36 @@ var i = a.indexOf(il);
             }
+            else if (a.length === 1) {
+                this._pruneCycles();
+            }
         }
-        else if (this._target) {
-            this._target._remove(il);
+    };
+    
+    
+    
+    
+    Stream.prototype._pruneCycles = function () {
+        if (this._onlyReachesThis(this)) {
+            this._remove(this._ils[0]);
         }
     };
-    Stream.prototype._setHIL = function (il) {
-        this._hil = il;
+    
+    
+    
+    Stream.prototype._onlyReachesThis = function (x) {
+        if (x.out === this) {
+            return true;
+        }
+        else if (x.out) {
+            return this._onlyReachesThis(x.out);
+        }
+        else if (x._ils) {
+            for (var i = 0, N = x._ils.length; i < N; i++) {
+                if (!this._onlyReachesThis(x._ils[i]))
+                    return false;
+            }
+            return true;
+        }
+        else {
+            return false;
+        }
     };
@@ -1116,3 +1136,2 @@ Stream.prototype.ctor = function () {
         this._target = target;
-        target._setHIL(this);
     };
@@ -1119,0 +1138,0 @@

dist/xstream.min.js

@@ -1,1 +0,1 @@
-(function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.xstream=f()}})(function(){var define,module,exports;return function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s}({1:[function(require,module,exports){"use strict";var __extends=this&&this.__extends||function(d,b){for(var p in b)if(b.hasOwnProperty(p))d[p]=b[p];function __(){this.constructor=d}d.prototype=b===null?Object.create(b):(__.prototype=b.prototype,new __)};var empty={};function noop(){}function copy(a){var l=a.length;var b=Array(l);for(var i=0;i<l;++i){b[i]=a[i]}return b}exports.emptyListener={_n:noop,_e:noop,_c:noop};function internalizeProducer(producer){producer._start=function _start(il){il.next=il._n;il.error=il._e;il.complete=il._c;this.start(il)};producer._stop=producer.stop}function compose2(f1,f2){return function composedFn(arg){return f1(f2(arg))}}function and(f1,f2){return function andFn(t){return f1(t)&&f2(t)}}var MergeProducer=function(){function MergeProducer(streams){this.streams=streams;this.type="merge";this.out=exports.emptyListener;this.ac=streams.length}MergeProducer.prototype._start=function(out){this.out=out;var s=this.streams;var L=s.length;for(var i=0;i<L;i++){s[i]._add(this)}};MergeProducer.prototype._stop=function(){var s=this.streams;var L=s.length;for(var i=0;i<L;i++){s[i]._remove(this)}this.out=null;this.ac=L};MergeProducer.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};MergeProducer.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MergeProducer.prototype._c=function(){if(--this.ac===0){var u=this.out;if(!u)return;u._c()}};return MergeProducer}();exports.MergeProducer=MergeProducer;var CombineListener=function(){function CombineListener(i,p){this.i=i;this.p=p;p.ils.push(this)}CombineListener.prototype._n=function(t){var p=this.p,out=p.out;if(!out)return;if(p.up(t,this.i)){out._n(p.vals)}};CombineListener.prototype._e=function(err){var out=this.p.out;if(!out)return;out._e(err)};CombineListener.prototype._c=function(){var p=this.p;if(!p.out)return;if(--p.Nc===0){p.out._c()}};return CombineListener}();exports.CombineListener=CombineListener;var CombineProducer=function(){function CombineProducer(s){this.s=s;this.type="combine";this.out=null;this.ils=[];var n=this.Nc=this.Nn=s.length;var vals=this.vals=new Array(n);for(var i=0;i<n;i++){vals[i]=empty}}CombineProducer.prototype.up=function(t,i){var v=this.vals[i];var Nn=!this.Nn?0:v===empty?--this.Nn:this.Nn;this.vals[i]=t;return Nn===0};CombineProducer.prototype._start=function(out){this.out=out;var s=this.s;var n=s.length;if(n===0){out._n([]);out._c()}else{for(var i=0;i<n;i++){s[i]._add(new CombineListener(i,this))}}};CombineProducer.prototype._stop=function(){var s=this.s;var n=this.Nc=this.Nn=s.length;var vals=this.vals=new Array(n);for(var i=0;i<n;i++){s[i]._remove(this.ils[i]);vals[i]=empty}this.out=null;this.ils=[]};return CombineProducer}();exports.CombineProducer=CombineProducer;var FromArrayProducer=function(){function FromArrayProducer(a){this.a=a;this.type="fromArray"}FromArrayProducer.prototype._start=function(out){var a=this.a;for(var i=0,l=a.length;i<l;i++){out._n(a[i])}out._c()};FromArrayProducer.prototype._stop=function(){};return FromArrayProducer}();exports.FromArrayProducer=FromArrayProducer;var FromPromiseProducer=function(){function FromPromiseProducer(p){this.p=p;this.type="fromPromise";this.on=false}FromPromiseProducer.prototype._start=function(out){var prod=this;this.on=true;this.p.then(function(v){if(prod.on){out._n(v);out._c()}},function(e){out._e(e)}).then(null,function(err){setTimeout(function(){throw err})})};FromPromiseProducer.prototype._stop=function(){this.on=false};return FromPromiseProducer}();exports.FromPromiseProducer=FromPromiseProducer;var PeriodicProducer=function(){function PeriodicProducer(period){this.period=period;this.type="periodic";this.intervalID=-1;this.i=0}PeriodicProducer.prototype._start=function(stream){var self=this;function intervalHandler(){stream._n(self.i++)}this.intervalID=setInterval(intervalHandler,this.period)};PeriodicProducer.prototype._stop=function(){if(this.intervalID!==-1)clearInterval(this.intervalID);this.intervalID=-1;this.i=0};return PeriodicProducer}();exports.PeriodicProducer=PeriodicProducer;var DebugOperator=function(){function DebugOperator(arg,ins){this.ins=ins;this.type="debug";this.out=null;this.s=null;this.l=null;if(typeof arg==="string"){this.l=arg}else{this.s=arg}}DebugOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};DebugOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};DebugOperator.prototype._n=function(t){var u=this.out;if(!u)return;var s=this.s,l=this.l;if(s){try{s(t)}catch(e){u._e(e)}}else if(l){console.log(l+":",t)}else{console.log(t)}u._n(t)};DebugOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};DebugOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return DebugOperator}();exports.DebugOperator=DebugOperator;var DropOperator=function(){function DropOperator(max,ins){this.max=max;this.ins=ins;this.type="drop";this.out=null;this.dropped=0}DropOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};DropOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.dropped=0};DropOperator.prototype._n=function(t){var u=this.out;if(!u)return;if(this.dropped++>=this.max)u._n(t)};DropOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};DropOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return DropOperator}();exports.DropOperator=DropOperator;var OtherIL=function(){function OtherIL(out,op){this.out=out;this.op=op}OtherIL.prototype._n=function(t){this.op.end()};OtherIL.prototype._e=function(err){this.out._e(err)};OtherIL.prototype._c=function(){this.op.end()};return OtherIL}();var EndWhenOperator=function(){function EndWhenOperator(o,ins){this.o=o;this.ins=ins;this.type="endWhen";this.out=null;this.oil=exports.emptyListener}EndWhenOperator.prototype._start=function(out){this.out=out;this.o._add(this.oil=new OtherIL(out,this));this.ins._add(this)};EndWhenOperator.prototype._stop=function(){this.ins._remove(this);this.o._remove(this.oil);this.out=null;this.oil=null};EndWhenOperator.prototype.end=function(){var u=this.out;if(!u)return;u._c()};EndWhenOperator.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};EndWhenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};EndWhenOperator.prototype._c=function(){this.end()};return EndWhenOperator}();exports.EndWhenOperator=EndWhenOperator;var FilterOperator=function(){function FilterOperator(passes,ins){this.passes=passes;this.ins=ins;this.type="filter";this.out=null}FilterOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};FilterOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};FilterOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{if(this.passes(t))u._n(t)}catch(e){u._e(e)}};FilterOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FilterOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return FilterOperator}();exports.FilterOperator=FilterOperator;var FlattenListener=function(){function FlattenListener(out,op){this.out=out;this.op=op}FlattenListener.prototype._n=function(t){this.out._n(t)};FlattenListener.prototype._e=function(err){this.out._e(err)};FlattenListener.prototype._c=function(){this.op.inner=null;this.op.less()};return FlattenListener}();var FlattenOperator=function(){function FlattenOperator(ins){this.ins=ins;this.type="flatten";this.inner=null;this.il=null;this.open=true;this.out=null}FlattenOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};FlattenOperator.prototype._stop=function(){this.ins._remove(this);this.inner=null;this.il=null;this.open=true;this.out=null};FlattenOperator.prototype.less=function(){var u=this.out;if(!u)return;if(!this.open&&!this.inner)u._c()};FlattenOperator.prototype._n=function(s){var u=this.out;if(!u)return;var _a=this,inner=_a.inner,il=_a.il;if(inner&&il)inner._remove(il);(this.inner=s)._add(this.il=new FlattenListener(u,this))};FlattenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FlattenOperator.prototype._c=function(){this.open=false;this.less()};return FlattenOperator}();exports.FlattenOperator=FlattenOperator;var FoldOperator=function(){function FoldOperator(f,seed,ins){this.f=f;this.seed=seed;this.ins=ins;this.type="fold";this.out=null;this.acc=seed}FoldOperator.prototype._start=function(out){this.out=out;out._n(this.acc);this.ins._add(this)};FoldOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.acc=this.seed};FoldOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{u._n(this.acc=this.f(this.acc,t))}catch(e){u._e(e)}};FoldOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FoldOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return FoldOperator}();exports.FoldOperator=FoldOperator;var LastOperator=function(){function LastOperator(ins){this.ins=ins;this.type="last";this.out=null;this.has=false;this.val=empty}LastOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};LastOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.has=false;this.val=empty};LastOperator.prototype._n=function(t){this.has=true;this.val=t};LastOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};LastOperator.prototype._c=function(){var u=this.out;if(!u)return;if(this.has){u._n(this.val);u._c()}else{u._e("TODO show proper error")}};return LastOperator}();exports.LastOperator=LastOperator;var MapFlattenInner=function(){function MapFlattenInner(out,op){this.out=out;this.op=op}MapFlattenInner.prototype._n=function(r){this.out._n(r)};MapFlattenInner.prototype._e=function(err){this.out._e(err)};MapFlattenInner.prototype._c=function(){this.op.inner=null;this.op.less()};return MapFlattenInner}();var MapFlattenOperator=function(){function MapFlattenOperator(mapOp){this.mapOp=mapOp;this.inner=null;this.il=null;this.open=true;this.out=null;this.type=mapOp.type+"+flatten";this.ins=mapOp.ins}MapFlattenOperator.prototype._start=function(out){this.out=out;this.mapOp.ins._add(this)};MapFlattenOperator.prototype._stop=function(){this.mapOp.ins._remove(this);this.inner=null;this.il=null;this.open=true;this.out=null};MapFlattenOperator.prototype.less=function(){if(!this.open&&!this.inner){var u=this.out;if(!u)return;u._c()}};MapFlattenOperator.prototype._n=function(v){var u=this.out;if(!u)return;var _a=this,inner=_a.inner,il=_a.il;if(inner&&il)inner._remove(il);try{(this.inner=this.mapOp.project(v))._add(this.il=new MapFlattenInner(u,this))}catch(e){u._e(e)}};MapFlattenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MapFlattenOperator.prototype._c=function(){this.open=false;this.less()};return MapFlattenOperator}();exports.MapFlattenOperator=MapFlattenOperator;var MapOperator=function(){function MapOperator(project,ins){this.project=project;this.ins=ins;this.type="map";this.out=null}MapOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};MapOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};MapOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{u._n(this.project(t))}catch(e){u._e(e)}};MapOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MapOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return MapOperator}();exports.MapOperator=MapOperator;var FilterMapOperator=function(_super){__extends(FilterMapOperator,_super);function FilterMapOperator(passes,project,ins){_super.call(this,project,ins);this.passes=passes;this.type="filter+map"}FilterMapOperator.prototype._n=function(v){if(this.passes(v)){_super.prototype._n.call(this,v)}};return FilterMapOperator}(MapOperator);exports.FilterMapOperator=FilterMapOperator;var ReplaceErrorOperator=function(){function ReplaceErrorOperator(fn,ins){this.fn=fn;this.ins=ins;this.type="replaceError";this.out=empty}ReplaceErrorOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};ReplaceErrorOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};ReplaceErrorOperator.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};ReplaceErrorOperator.prototype._e=function(err){var u=this.out;if(!u)return;try{this.ins._remove(this);(this.ins=this.fn(err))._add(this)}catch(e){u._e(e)}};ReplaceErrorOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return ReplaceErrorOperator}();exports.ReplaceErrorOperator=ReplaceErrorOperator;var StartWithOperator=function(){function StartWithOperator(ins,value){this.ins=ins;this.value=value;this.type="startWith";this.out=exports.emptyListener}StartWithOperator.prototype._start=function(out){this.out=out;this.out._n(this.value);this.ins._add(out)};StartWithOperator.prototype._stop=function(){this.ins._remove(this.out);this.out=null};return StartWithOperator}();exports.StartWithOperator=StartWithOperator;var TakeOperator=function(){function TakeOperator(max,ins){this.max=max;this.ins=ins;this.type="take";this.out=null;this.taken=0}TakeOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};TakeOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.taken=0};TakeOperator.prototype._n=function(t){var u=this.out;if(!u)return;if(this.taken++<this.max-1){u._n(t)}else{u._n(t);u._c()}};TakeOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};TakeOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return TakeOperator}();exports.TakeOperator=TakeOperator;var Stream=function(){function Stream(producer){this._stopID=empty;this._prod=producer;this._ils=[];this._hil=null;this._target=null}Stream.prototype._n=function(t){var a=this._ils;var L=a.length;if(L==1)a[0]._n(t);else{var b=copy(a);for(var i=0;i<L;i++)b[i]._n(t)}var h=this._hil;if(h)h._n(t)};Stream.prototype._e=function(err){var a=this._ils;var L=a.length;if(L==1)a[0]._e(err);else{var b=copy(a);for(var i=0;i<L;i++)b[i]._e(err)}var h=this._hil;if(h)h._e(err);this._x()};Stream.prototype._c=function(){var a=this._ils;var L=a.length;if(L==1)a[0]._c();else{var b=copy(a);for(var i=0;i<L;i++)b[i]._c()}var h=this._hil;if(h)h._c();this._x()};Stream.prototype._x=function(){if(this._ils.length===0)return;if(this._prod)this._prod._stop();this._ils=[]};Stream.prototype.addListener=function(listener){if(typeof listener.next!=="function"||typeof listener.error!=="function"||typeof listener.complete!=="function"){throw new Error("stream.addListener() requires all three next, error, "+"and complete functions.")}listener._n=listener.next;listener._e=listener.error;listener._c=listener.complete;this._add(listener)};Stream.prototype.removeListener=function(listener){this._remove(listener)};Stream.prototype._add=function(il){var ta=this._target;if(ta&&ta._ils.length===0){return ta._add(il)}var a=this._ils;a.push(il);if(a.length===1){if(this._stopID!==empty){clearTimeout(this._stopID);this._stopID=empty}var p=this._prod;if(p)p._start(this)}};Stream.prototype._remove=function(il){var a=this._ils;var i=a.indexOf(il);if(i>-1){a.splice(i,1);var p_1=this._prod;if(p_1&&a.length<=0){this._stopID=setTimeout(function(){return p_1._stop()})}}else if(this._target){this._target._remove(il)}};Stream.prototype._setHIL=function(il){this._hil=il};Stream.prototype.ctor=function(){return this instanceof MemoryStream?MemoryStream:Stream};Stream.create=function(producer){if(producer){if(typeof producer.start!=="function"||typeof producer.stop!=="function"){throw new Error("producer requires both start and stop functions")}internalizeProducer(producer)}return new Stream(producer)};Stream.createWithMemory=function(producer){if(producer){internalizeProducer(producer)}return new MemoryStream(producer)};Stream.never=function(){return new Stream({_start:noop,_stop:noop})};Stream.empty=function(){return new Stream({_start:function(il){il._c()},_stop:noop})};Stream.throw=function(error){return new Stream({_start:function(il){il._e(error)},_stop:noop})};Stream.of=function(){var items=[];for(var _i=0;_i<arguments.length;_i++){items[_i-0]=arguments[_i]}return Stream.fromArray(items)};Stream.fromArray=function(array){return new Stream(new FromArrayProducer(array))};Stream.fromPromise=function(promise){return new Stream(new FromPromiseProducer(promise))};Stream.periodic=function(period){return new Stream(new PeriodicProducer(period))};Stream.merge=function(){var streams=[];for(var _i=0;_i<arguments.length;_i++){streams[_i-0]=arguments[_i]}return new Stream(new MergeProducer(streams))};Stream.prototype._map=function(project){var p=this._prod;var ctor=this.ctor();if(p instanceof FilterOperator){return new ctor(new FilterMapOperator(p.passes,project,p.ins))}if(p instanceof FilterMapOperator){return new ctor(new FilterMapOperator(p.passes,compose2(project,p.project),p.ins))}if(p instanceof MapOperator){return new ctor(new MapOperator(compose2(project,p.project),p.ins))}return new ctor(new MapOperator(project,this))};Stream.prototype.map=function(project){return this._map(project)};Stream.prototype.mapTo=function(projectedValue){var s=this.map(function(){return projectedValue});var op=s._prod;op.type=op.type.replace("map","mapTo");return s};Stream.prototype.filter=function(passes){var p=this._prod;if(p instanceof FilterOperator){return new Stream(new FilterOperator(and(passes,p.passes),p.ins))}return new Stream(new FilterOperator(passes,this))};Stream.prototype.take=function(amount){return new(this.ctor())(new TakeOperator(amount,this))};Stream.prototype.drop=function(amount){return new Stream(new DropOperator(amount,this))};Stream.prototype.last=function(){return new Stream(new LastOperator(this))};Stream.prototype.startWith=function(initial){return new MemoryStream(new StartWithOperator(this,initial))};Stream.prototype.endWhen=function(other){return new(this.ctor())(new EndWhenOperator(other,this))};Stream.prototype.fold=function(accumulate,seed){return new MemoryStream(new FoldOperator(accumulate,seed,this))};Stream.prototype.replaceError=function(replace){return new(this.ctor())(new ReplaceErrorOperator(replace,this))};Stream.prototype.flatten=function(){var p=this._prod;return new Stream(p instanceof MapOperator&&!(p instanceof FilterMapOperator)?new MapFlattenOperator(p):new FlattenOperator(this))};Stream.prototype.compose=function(operator){return operator(this)};Stream.prototype.remember=function(){var _this=this;return new MemoryStream({_start:function(il){var p=_this._prod;if(p)p._start(il)},_stop:function(){var p=_this._prod;if(p)p._stop()}})};Stream.prototype.debug=function(labelOrSpy){return new(this.ctor())(new DebugOperator(labelOrSpy,this))};Stream.prototype.imitate=function(target){if(target instanceof MemoryStream){throw new Error("A MemoryStream was given to imitate(), but it only "+"supports a Stream. Read more about this restriction here: "+"https://github.com/staltz/xstream#faq")}this._target=target;target._setHIL(this)};Stream.prototype.shamefullySendNext=function(value){this._n(value)};Stream.prototype.shamefullySendError=function(error){this._e(error)};Stream.prototype.shamefullySendComplete=function(){this._c()};Stream.combine=function combine(){var streams=[];for(var _i=0;_i<arguments.length;_i++){streams[_i-0]=arguments[_i]}return new Stream(new CombineProducer(streams))};return Stream}();exports.Stream=Stream;var MemoryStream=function(_super){__extends(MemoryStream,_super);function MemoryStream(producer){_super.call(this,producer);this._has=false}MemoryStream.prototype._n=function(x){this._v=x;this._has=true;_super.prototype._n.call(this,x)};MemoryStream.prototype._add=function(il){if(this._has){il._n(this._v)}_super.prototype._add.call(this,il)};MemoryStream.prototype._x=function(){this._has=false;_super.prototype._x.call(this)};MemoryStream.prototype.map=function(project){return this._map(project)};MemoryStream.prototype.mapTo=function(projectedValue){return _super.prototype.mapTo.call(this,projectedValue)};MemoryStream.prototype.take=function(amount){return _super.prototype.take.call(this,amount)};MemoryStream.prototype.endWhen=function(other){return _super.prototype.endWhen.call(this,other)};MemoryStream.prototype.replaceError=function(replace){return _super.prototype.replaceError.call(this,replace)};MemoryStream.prototype.debug=function(labelOrSpy){return _super.prototype.debug.call(this,labelOrSpy)};return MemoryStream}(Stream);exports.MemoryStream=MemoryStream;Object.defineProperty(exports,"__esModule",{value:true});exports.default=Stream},{}],2:[function(require,module,exports){"use strict";var core_1=require("./core");exports.Stream=core_1.Stream;exports.MemoryStream=core_1.MemoryStream;Object.defineProperty(exports,"__esModule",{value:true});exports.default=core_1.Stream},{"./core":1}]},{},[2])(2)});
+(function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.xstream=f()}})(function(){var define,module,exports;return function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s}({1:[function(require,module,exports){"use strict";var __extends=this&&this.__extends||function(d,b){for(var p in b)if(b.hasOwnProperty(p))d[p]=b[p];function __(){this.constructor=d}d.prototype=b===null?Object.create(b):(__.prototype=b.prototype,new __)};var empty={};function noop(){}function copy(a){var l=a.length;var b=Array(l);for(var i=0;i<l;++i){b[i]=a[i]}return b}exports.emptyIL={_n:noop,_e:noop,_c:noop};function internalizeProducer(producer){producer._start=function _start(il){il.next=il._n;il.error=il._e;il.complete=il._c;this.start(il)};producer._stop=producer.stop}function compose2(f1,f2){return function composedFn(arg){return f1(f2(arg))}}function and(f1,f2){return function andFn(t){return f1(t)&&f2(t)}}var MergeProducer=function(){function MergeProducer(insArr){this.insArr=insArr;this.type="merge";this.out=null;this.ac=insArr.length}MergeProducer.prototype._start=function(out){this.out=out;var s=this.insArr;var L=s.length;for(var i=0;i<L;i++){s[i]._add(this)}};MergeProducer.prototype._stop=function(){var s=this.insArr;var L=s.length;for(var i=0;i<L;i++){s[i]._remove(this)}this.out=null;this.ac=L};MergeProducer.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};MergeProducer.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MergeProducer.prototype._c=function(){if(--this.ac===0){var u=this.out;if(!u)return;u._c()}};return MergeProducer}();exports.MergeProducer=MergeProducer;var CombineListener=function(){function CombineListener(i,out,p){this.i=i;this.out=out;this.p=p;p.ils.push(this)}CombineListener.prototype._n=function(t){var p=this.p,out=this.out;if(!out)return;if(p.up(t,this.i)){out._n(p.vals)}};CombineListener.prototype._e=function(err){var out=this.out;if(!out)return;out._e(err)};CombineListener.prototype._c=function(){var p=this.p;if(!p.out)return;if(--p.Nc===0){p.out._c()}};return CombineListener}();exports.CombineListener=CombineListener;var CombineProducer=function(){function CombineProducer(insArr){this.insArr=insArr;this.type="combine";this.out=null;this.ils=[];var n=this.Nc=this.Nn=insArr.length;var vals=this.vals=new Array(n);for(var i=0;i<n;i++){vals[i]=empty}}CombineProducer.prototype.up=function(t,i){var v=this.vals[i];var Nn=!this.Nn?0:v===empty?--this.Nn:this.Nn;this.vals[i]=t;return Nn===0};CombineProducer.prototype._start=function(out){this.out=out;var s=this.insArr;var n=s.length;if(n===0){out._n([]);out._c()}else{for(var i=0;i<n;i++){s[i]._add(new CombineListener(i,out,this))}}};CombineProducer.prototype._stop=function(){var s=this.insArr;var n=this.Nc=this.Nn=s.length;var vals=this.vals=new Array(n);for(var i=0;i<n;i++){s[i]._remove(this.ils[i]);vals[i]=empty}this.out=null;this.ils=[]};return CombineProducer}();exports.CombineProducer=CombineProducer;var FromArrayProducer=function(){function FromArrayProducer(a){this.a=a;this.type="fromArray"}FromArrayProducer.prototype._start=function(out){var a=this.a;for(var i=0,l=a.length;i<l;i++){out._n(a[i])}out._c()};FromArrayProducer.prototype._stop=function(){};return FromArrayProducer}();exports.FromArrayProducer=FromArrayProducer;var FromPromiseProducer=function(){function FromPromiseProducer(p){this.p=p;this.type="fromPromise";this.on=false}FromPromiseProducer.prototype._start=function(out){var prod=this;this.on=true;this.p.then(function(v){if(prod.on){out._n(v);out._c()}},function(e){out._e(e)}).then(null,function(err){setTimeout(function(){throw err})})};FromPromiseProducer.prototype._stop=function(){this.on=false};return FromPromiseProducer}();exports.FromPromiseProducer=FromPromiseProducer;var PeriodicProducer=function(){function PeriodicProducer(period){this.period=period;this.type="periodic";this.intervalID=-1;this.i=0}PeriodicProducer.prototype._start=function(stream){var self=this;function intervalHandler(){stream._n(self.i++)}this.intervalID=setInterval(intervalHandler,this.period)};PeriodicProducer.prototype._stop=function(){if(this.intervalID!==-1)clearInterval(this.intervalID);this.intervalID=-1;this.i=0};return PeriodicProducer}();exports.PeriodicProducer=PeriodicProducer;var DebugOperator=function(){function DebugOperator(arg,ins){this.ins=ins;this.type="debug";this.out=null;this.s=null;this.l=null;if(typeof arg==="string"){this.l=arg}else{this.s=arg}}DebugOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};DebugOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};DebugOperator.prototype._n=function(t){var u=this.out;if(!u)return;var s=this.s,l=this.l;if(s){try{s(t)}catch(e){u._e(e)}}else if(l){console.log(l+":",t)}else{console.log(t)}u._n(t)};DebugOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};DebugOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return DebugOperator}();exports.DebugOperator=DebugOperator;var DropOperator=function(){function DropOperator(max,ins){this.max=max;this.ins=ins;this.type="drop";this.out=null;this.dropped=0}DropOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};DropOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.dropped=0};DropOperator.prototype._n=function(t){var u=this.out;if(!u)return;if(this.dropped++>=this.max)u._n(t)};DropOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};DropOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return DropOperator}();exports.DropOperator=DropOperator;var OtherIL=function(){function OtherIL(out,op){this.out=out;this.op=op}OtherIL.prototype._n=function(t){this.op.end()};OtherIL.prototype._e=function(err){this.out._e(err)};OtherIL.prototype._c=function(){this.op.end()};return OtherIL}();var EndWhenOperator=function(){function EndWhenOperator(o,ins){this.o=o;this.ins=ins;this.type="endWhen";this.out=null;this.oil=exports.emptyIL}EndWhenOperator.prototype._start=function(out){this.out=out;this.o._add(this.oil=new OtherIL(out,this));this.ins._add(this)};EndWhenOperator.prototype._stop=function(){this.ins._remove(this);this.o._remove(this.oil);this.out=null;this.oil=null};EndWhenOperator.prototype.end=function(){var u=this.out;if(!u)return;u._c()};EndWhenOperator.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};EndWhenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};EndWhenOperator.prototype._c=function(){this.end()};return EndWhenOperator}();exports.EndWhenOperator=EndWhenOperator;var FilterOperator=function(){function FilterOperator(passes,ins){this.passes=passes;this.ins=ins;this.type="filter";this.out=null}FilterOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};FilterOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};FilterOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{if(this.passes(t))u._n(t)}catch(e){u._e(e)}};FilterOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FilterOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return FilterOperator}();exports.FilterOperator=FilterOperator;var FlattenListener=function(){function FlattenListener(out,op){this.out=out;this.op=op}FlattenListener.prototype._n=function(t){this.out._n(t)};FlattenListener.prototype._e=function(err){this.out._e(err)};FlattenListener.prototype._c=function(){this.op.inner=null;this.op.less()};return FlattenListener}();var FlattenOperator=function(){function FlattenOperator(ins){this.ins=ins;this.type="flatten";this.inner=null;this.il=null;this.open=true;this.out=null}FlattenOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};FlattenOperator.prototype._stop=function(){this.ins._remove(this);this.inner=null;this.il=null;this.open=true;this.out=null};FlattenOperator.prototype.less=function(){var u=this.out;if(!u)return;if(!this.open&&!this.inner)u._c()};FlattenOperator.prototype._n=function(s){var u=this.out;if(!u)return;var _a=this,inner=_a.inner,il=_a.il;if(inner&&il)inner._remove(il);(this.inner=s)._add(this.il=new FlattenListener(u,this))};FlattenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FlattenOperator.prototype._c=function(){this.open=false;this.less()};return FlattenOperator}();exports.FlattenOperator=FlattenOperator;var FoldOperator=function(){function FoldOperator(f,seed,ins){this.f=f;this.seed=seed;this.ins=ins;this.type="fold";this.out=null;this.acc=seed}FoldOperator.prototype._start=function(out){this.out=out;out._n(this.acc);this.ins._add(this)};FoldOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.acc=this.seed};FoldOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{u._n(this.acc=this.f(this.acc,t))}catch(e){u._e(e)}};FoldOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};FoldOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return FoldOperator}();exports.FoldOperator=FoldOperator;var LastOperator=function(){function LastOperator(ins){this.ins=ins;this.type="last";this.out=null;this.has=false;this.val=empty}LastOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};LastOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.has=false;this.val=empty};LastOperator.prototype._n=function(t){this.has=true;this.val=t};LastOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};LastOperator.prototype._c=function(){var u=this.out;if(!u)return;if(this.has){u._n(this.val);u._c()}else{u._e("TODO show proper error")}};return LastOperator}();exports.LastOperator=LastOperator;var MapFlattenInner=function(){function MapFlattenInner(out,op){this.out=out;this.op=op}MapFlattenInner.prototype._n=function(r){this.out._n(r)};MapFlattenInner.prototype._e=function(err){this.out._e(err)};MapFlattenInner.prototype._c=function(){this.op.inner=null;this.op.less()};return MapFlattenInner}();var MapFlattenOperator=function(){function MapFlattenOperator(mapOp){this.mapOp=mapOp;this.inner=null;this.il=null;this.open=true;this.out=null;this.type=mapOp.type+"+flatten";this.ins=mapOp.ins}MapFlattenOperator.prototype._start=function(out){this.out=out;this.mapOp.ins._add(this)};MapFlattenOperator.prototype._stop=function(){this.mapOp.ins._remove(this);this.inner=null;this.il=null;this.open=true;this.out=null};MapFlattenOperator.prototype.less=function(){if(!this.open&&!this.inner){var u=this.out;if(!u)return;u._c()}};MapFlattenOperator.prototype._n=function(v){var u=this.out;if(!u)return;var _a=this,inner=_a.inner,il=_a.il;if(inner&&il)inner._remove(il);try{(this.inner=this.mapOp.project(v))._add(this.il=new MapFlattenInner(u,this))}catch(e){u._e(e)}};MapFlattenOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MapFlattenOperator.prototype._c=function(){this.open=false;this.less()};return MapFlattenOperator}();exports.MapFlattenOperator=MapFlattenOperator;var MapOperator=function(){function MapOperator(project,ins){this.project=project;this.ins=ins;this.type="map";this.out=null}MapOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};MapOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};MapOperator.prototype._n=function(t){var u=this.out;if(!u)return;try{u._n(this.project(t))}catch(e){u._e(e)}};MapOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};MapOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return MapOperator}();exports.MapOperator=MapOperator;var FilterMapOperator=function(_super){__extends(FilterMapOperator,_super);function FilterMapOperator(passes,project,ins){_super.call(this,project,ins);this.passes=passes;this.type="filter+map"}FilterMapOperator.prototype._n=function(v){if(this.passes(v)){_super.prototype._n.call(this,v)}};return FilterMapOperator}(MapOperator);exports.FilterMapOperator=FilterMapOperator;var ReplaceErrorOperator=function(){function ReplaceErrorOperator(fn,ins){this.fn=fn;this.ins=ins;this.type="replaceError";this.out=empty}ReplaceErrorOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};ReplaceErrorOperator.prototype._stop=function(){this.ins._remove(this);this.out=null};ReplaceErrorOperator.prototype._n=function(t){var u=this.out;if(!u)return;u._n(t)};ReplaceErrorOperator.prototype._e=function(err){var u=this.out;if(!u)return;try{this.ins._remove(this);(this.ins=this.fn(err))._add(this)}catch(e){u._e(e)}};ReplaceErrorOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return ReplaceErrorOperator}();exports.ReplaceErrorOperator=ReplaceErrorOperator;var StartWithOperator=function(){function StartWithOperator(ins,value){this.ins=ins;this.value=value;this.type="startWith";this.out=exports.emptyIL}StartWithOperator.prototype._start=function(out){this.out=out;this.out._n(this.value);this.ins._add(out)};StartWithOperator.prototype._stop=function(){this.ins._remove(this.out);this.out=null};return StartWithOperator}();exports.StartWithOperator=StartWithOperator;var TakeOperator=function(){function TakeOperator(max,ins){this.max=max;this.ins=ins;this.type="take";this.out=null;this.taken=0}TakeOperator.prototype._start=function(out){this.out=out;this.ins._add(this)};TakeOperator.prototype._stop=function(){this.ins._remove(this);this.out=null;this.taken=0};TakeOperator.prototype._n=function(t){var u=this.out;if(!u)return;if(this.taken++<this.max-1){u._n(t)}else{u._n(t);u._c()}};TakeOperator.prototype._e=function(err){var u=this.out;if(!u)return;u._e(err)};TakeOperator.prototype._c=function(){var u=this.out;if(!u)return;u._c()};return TakeOperator}();exports.TakeOperator=TakeOperator;var Stream=function(){function Stream(producer){this._stopID=empty;this._prod=producer;this._ils=[];this._target=null}Stream.prototype._n=function(t){var a=this._ils;var L=a.length;if(L==1)a[0]._n(t);else{var b=copy(a);for(var i=0;i<L;i++)b[i]._n(t)}};Stream.prototype._e=function(err){var a=this._ils;var L=a.length;if(L==1)a[0]._e(err);else{var b=copy(a);for(var i=0;i<L;i++)b[i]._e(err)}this._x()};Stream.prototype._c=function(){var a=this._ils;var L=a.length;if(L==1)a[0]._c();else{var b=copy(a);for(var i=0;i<L;i++)b[i]._c()}this._x()};Stream.prototype._x=function(){if(this._ils.length===0)return;if(this._prod)this._prod._stop();this._ils=[]};Stream.prototype.addListener=function(listener){if(typeof listener.next!=="function"||typeof listener.error!=="function"||typeof listener.complete!=="function"){throw new Error("stream.addListener() requires all three next, error, "+"and complete functions.")}listener._n=listener.next;listener._e=listener.error;listener._c=listener.complete;this._add(listener)};Stream.prototype.removeListener=function(listener){this._remove(listener)};Stream.prototype._add=function(il){var ta=this._target;if(ta)return ta._add(il);var a=this._ils;a.push(il);if(a.length===1){if(this._stopID!==empty){clearTimeout(this._stopID);this._stopID=empty}var p=this._prod;if(p)p._start(this)}};Stream.prototype._remove=function(il){var ta=this._target;if(ta)return ta._remove(il);var a=this._ils;var i=a.indexOf(il);if(i>-1){a.splice(i,1);var p_1=this._prod;if(p_1&&a.length<=0){this._stopID=setTimeout(function(){return p_1._stop()})}else if(a.length===1){this._pruneCycles()}}};Stream.prototype._pruneCycles=function(){if(this._onlyReachesThis(this)){this._remove(this._ils[0])}};Stream.prototype._onlyReachesThis=function(x){if(x.out===this){return true}else if(x.out){return this._onlyReachesThis(x.out)}else if(x._ils){for(var i=0,N=x._ils.length;i<N;i++){if(!this._onlyReachesThis(x._ils[i]))return false}return true}else{return false}};Stream.prototype.ctor=function(){return this instanceof MemoryStream?MemoryStream:Stream};Stream.create=function(producer){if(producer){if(typeof producer.start!=="function"||typeof producer.stop!=="function"){throw new Error("producer requires both start and stop functions")}internalizeProducer(producer)}return new Stream(producer)};Stream.createWithMemory=function(producer){if(producer){internalizeProducer(producer)}return new MemoryStream(producer)};Stream.never=function(){return new Stream({_start:noop,_stop:noop})};Stream.empty=function(){return new Stream({_start:function(il){il._c()},_stop:noop})};Stream.throw=function(error){return new Stream({_start:function(il){il._e(error)},_stop:noop})};Stream.of=function(){var items=[];for(var _i=0;_i<arguments.length;_i++){items[_i-0]=arguments[_i]}return Stream.fromArray(items)};Stream.fromArray=function(array){return new Stream(new FromArrayProducer(array))};Stream.fromPromise=function(promise){return new Stream(new FromPromiseProducer(promise))};Stream.periodic=function(period){return new Stream(new PeriodicProducer(period))};Stream.merge=function(){var streams=[];for(var _i=0;_i<arguments.length;_i++){streams[_i-0]=arguments[_i]}return new Stream(new MergeProducer(streams))};Stream.prototype._map=function(project){var p=this._prod;var ctor=this.ctor();if(p instanceof FilterOperator){return new ctor(new FilterMapOperator(p.passes,project,p.ins))}if(p instanceof FilterMapOperator){return new ctor(new FilterMapOperator(p.passes,compose2(project,p.project),p.ins))}if(p instanceof MapOperator){return new ctor(new MapOperator(compose2(project,p.project),p.ins))}return new ctor(new MapOperator(project,this))};Stream.prototype.map=function(project){return this._map(project)};Stream.prototype.mapTo=function(projectedValue){var s=this.map(function(){return projectedValue});var op=s._prod;op.type=op.type.replace("map","mapTo");return s};Stream.prototype.filter=function(passes){var p=this._prod;if(p instanceof FilterOperator){return new Stream(new FilterOperator(and(passes,p.passes),p.ins))}return new Stream(new FilterOperator(passes,this))};Stream.prototype.take=function(amount){return new(this.ctor())(new TakeOperator(amount,this))};Stream.prototype.drop=function(amount){return new Stream(new DropOperator(amount,this))};Stream.prototype.last=function(){return new Stream(new LastOperator(this))};Stream.prototype.startWith=function(initial){return new MemoryStream(new StartWithOperator(this,initial))};Stream.prototype.endWhen=function(other){return new(this.ctor())(new EndWhenOperator(other,this))};Stream.prototype.fold=function(accumulate,seed){return new MemoryStream(new FoldOperator(accumulate,seed,this))};Stream.prototype.replaceError=function(replace){return new(this.ctor())(new ReplaceErrorOperator(replace,this))};Stream.prototype.flatten=function(){var p=this._prod;return new Stream(p instanceof MapOperator&&!(p instanceof FilterMapOperator)?new MapFlattenOperator(p):new FlattenOperator(this))};Stream.prototype.compose=function(operator){return operator(this)};Stream.prototype.remember=function(){var _this=this;return new MemoryStream({_start:function(il){var p=_this._prod;if(p)p._start(il)},_stop:function(){var p=_this._prod;if(p)p._stop()}})};Stream.prototype.debug=function(labelOrSpy){return new(this.ctor())(new DebugOperator(labelOrSpy,this))};Stream.prototype.imitate=function(target){if(target instanceof MemoryStream){throw new Error("A MemoryStream was given to imitate(), but it only "+"supports a Stream. Read more about this restriction here: "+"https://github.com/staltz/xstream#faq")}this._target=target};Stream.prototype.shamefullySendNext=function(value){this._n(value)};Stream.prototype.shamefullySendError=function(error){this._e(error)};Stream.prototype.shamefullySendComplete=function(){this._c()};Stream.combine=function combine(){var streams=[];for(var _i=0;_i<arguments.length;_i++){streams[_i-0]=arguments[_i]}return new Stream(new CombineProducer(streams))};return Stream}();exports.Stream=Stream;var MemoryStream=function(_super){__extends(MemoryStream,_super);function MemoryStream(producer){_super.call(this,producer);this._has=false}MemoryStream.prototype._n=function(x){this._v=x;this._has=true;_super.prototype._n.call(this,x)};MemoryStream.prototype._add=function(il){if(this._has){il._n(this._v)}_super.prototype._add.call(this,il)};MemoryStream.prototype._x=function(){this._has=false;_super.prototype._x.call(this)};MemoryStream.prototype.map=function(project){return this._map(project)};MemoryStream.prototype.mapTo=function(projectedValue){return _super.prototype.mapTo.call(this,projectedValue)};MemoryStream.prototype.take=function(amount){return _super.prototype.take.call(this,amount)};MemoryStream.prototype.endWhen=function(other){return _super.prototype.endWhen.call(this,other)};MemoryStream.prototype.replaceError=function(replace){return _super.prototype.replaceError.call(this,replace)};MemoryStream.prototype.debug=function(labelOrSpy){return _super.prototype.debug.call(this,labelOrSpy)};return MemoryStream}(Stream);exports.MemoryStream=MemoryStream;Object.defineProperty(exports,"__esModule",{value:true});exports.default=Stream},{}],2:[function(require,module,exports){"use strict";var core_1=require("./core");exports.Stream=core_1.Stream;exports.MemoryStream=core_1.MemoryStream;Object.defineProperty(exports,"__esModule",{value:true});exports.default=core_1.Stream},{"./core":1}]},{},[2])(2)});

extra/dropRepeats.d.ts

@@ -6,3 +6,3 @@ import { Operator, Stream } from '../core';
     type: string;
-    private out;
+    out: Stream<T>;
     private v;
@@ -9,0 +9,0 @@ constructor(fn: (x: T, y: T) => boolean, ins: Stream<T>);

extra/dropUntil.d.ts

@@ -6,3 +6,3 @@ import { Operator, Stream } from '../core';
     type: string;
-    private out;
+    out: Stream<T>;
     private oil;
@@ -9,0 +9,0 @@ private on;

extra/dropUntil.js

@@ -26,3 +26,3 @@ "use strict";
         this.out = null;
-        this.oil = core_1.emptyListener; // oil = other InternalListener
+        this.oil = core_1.emptyIL; // oil = other InternalListener
         this.on = false;
@@ -29,0 +29,0 @@ }

extra/flattenConcurrently.d.ts

@@ -6,3 +6,3 @@ import { Operator, Stream } from '../core';
     private active;
-    private out;
+    out: Stream<T>;
     constructor(ins: Stream<Stream<T>>);
@@ -9,0 +9,0 @@ _start(out: Stream<T>): void;

extra/flattenSequentially.d.ts

@@ -8,3 +8,3 @@ import { Operator, Stream } from '../core';
     private seq;
-    private out;
+    out: Stream<T>;
     constructor(ins: Stream<Stream<T>>);
@@ -11,0 +11,0 @@ _start(out: Stream<T>): void;

extra/split.d.ts

@@ -7,3 +7,3 @@ import { Operator, Stream } from '../core';
     curr: Stream<T>;
-    private out;
+    out: Stream<Stream<T>>;
     private sil;
@@ -10,0 +10,0 @@ constructor(s: Stream<any>, ins: Stream<T>);

extra/split.js

@@ -28,3 +28,3 @@ "use strict";
         this.out = null;
-        this.sil = core_1.emptyListener; // sil = separator InternalListener
+        this.sil = core_1.emptyIL; // sil = separator InternalListener
     }
@@ -42,3 +42,3 @@ SplitOperator.prototype._start = function (out) {
         this.out = null;
-        this.sil = core_1.emptyListener;
+        this.sil = core_1.emptyIL;
     };
@@ -45,0 +45,0 @@ SplitOperator.prototype.up = function () {

package.json

 {
   "name": "xstream",
-  "version": "5.0.3",
+  "version": "5.0.4",
   "description": "An extremely intuitive, small, and fast functional reactive stream library for JavaScript",
@@ -5,0 +5,0 @@ "main": "index.js",

src/core.ts

@@ -21,3 +21,3 @@ import {Promise} from 'es6-promise';
 
-export const emptyListener: InternalListener<any> = {
+export const emptyIL: InternalListener<any> = {
   _n: noop,
@@ -33,12 +33,18 @@ _e: noop,
 
-export interface Operator<T, R> extends InternalProducer<R>, InternalListener<T> {
+export interface OutSender<T> {
+  out: Stream<T>;
+}
+
+export interface Operator<T, R> extends InternalProducer<R>, InternalListener<T>, OutSender<R> {
   type: string;
   ins: Stream<T>;
   _start: (out: Stream<R>) => void;
-  _stop: () => void;
-  _n: (v: T) => void;
-  _e: (err: any) => void;
-  _c: () => void;
 }
 
+export interface Aggregator<T, U> extends InternalProducer<U>, OutSender<U> {
+  type: string;
+  insArr: Array<Stream<T>>;
+  _start: (out: Stream<U>) => void;
+}
+
 export interface Producer<T> {
@@ -79,14 +85,14 @@ start: (listener: Listener<T>) => void;
 
-export class MergeProducer<T> implements InternalProducer<T>, InternalListener<T> {
+export class MergeProducer<T> implements Aggregator<T, T>, InternalListener<T> {
   public type = 'merge';
-  private out: InternalListener<T> = emptyListener;
+  public out: Stream<T> = null;
   private ac: number; // ac is activeCount, starts initialized
 
-  constructor(public streams: Array<Stream<T>>) {
-    this.ac = streams.length;
+  constructor(public insArr: Array<Stream<T>>) {
+    this.ac = insArr.length;
   }
 
-  _start(out: InternalListener<T>): void {
+  _start(out: Stream<T>): void {
     this.out = out;
-    const s = this.streams;
+    const s = this.insArr;
     const L = s.length;
@@ -99,3 +105,3 @@ for (let i = 0; i < L; i++) {
   _stop(): void {
-    const s = this.streams;
+    const s = this.insArr;
     const L = s.length;
@@ -161,4 +167,5 @@ for (let i = 0; i < L; i++) {
 
-export class CombineListener<T> implements InternalListener<T> {
+export class CombineListener<T> implements InternalListener<T>, OutSender<Array<T>> {
   constructor(private i: number,
+              public out: Stream<Array<T>>,
               private p: CombineProducer<T>) {
@@ -169,3 +176,3 @@ p.ils.push(this);
   _n(t: T): void {
-    const p = this.p, out = p.out;
+    const p = this.p, out = this.out;
     if (!out) return;
@@ -178,3 +185,3 @@ if (p.up(t, this.i)) {
   _e(err: any): void {
-    const out = this.p.out;
+    const out = this.out;
     if (!out) return;
@@ -193,5 +200,5 @@ out._e(err);
 
-export class CombineProducer<R> implements InternalProducer<Array<R>> {
+export class CombineProducer<R> implements Aggregator<any, Array<R>> {
   public type = 'combine';
-  public out: InternalListener<Array<R>> = null;
+  public out: Stream<Array<R>> = null;
   public ils: Array<CombineListener<any>> = [];
@@ -202,4 +209,4 @@ public Nc: number; // *N*umber of streams still to send *c*omplete
 
-  constructor(public s: Array<Stream<any>>) {
-    const n = this.Nc = this.Nn = s.length;
+  constructor(public insArr: Array<Stream<any>>) {
+    const n = this.Nc = this.Nn = insArr.length;
     const vals = this.vals = new Array(n);
@@ -218,5 +225,5 @@ for (let i = 0; i < n; i++) {
 
-  _start(out: InternalListener<Array<R>>): void {
+  _start(out: Stream<Array<R>>): void {
     this.out = out;
-    const s = this.s;
+    const s = this.insArr;
     const n = s.length;
@@ -228,3 +235,3 @@ if (n === 0) {
       for (let i = 0; i < n; i++) {
-        s[i]._add(new CombineListener(i, this));
+        s[i]._add(new CombineListener(i, out, this));
       }
@@ -235,3 +242,3 @@ }
   _stop(): void {
-    const s = this.s;
+    const s = this.insArr;
     const n = this.Nc = this.Nn = s.length;
@@ -318,3 +325,3 @@ const vals = this.vals = new Array(n);
   public type = 'debug';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private s: (t: T) => any = null; // spy
@@ -375,3 +382,3 @@ private l: string = null; // label
   public type = 'drop';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private dropped: number = 0;
@@ -433,4 +440,4 @@
   public type = 'endWhen';
-  private out: Stream<T> = null;
-  private oil: InternalListener<any> = emptyListener; // oil = other InternalListener
+  public out: Stream<T> = null;
+  private oil: InternalListener<any> = emptyIL; // oil = other InternalListener
 
@@ -479,3 +486,3 @@ constructor(public o: Stream<any>, // o = other
   public type = 'filter';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
 
@@ -543,3 +550,3 @@ constructor(public passes: (t: T) => boolean,
   private open: boolean = true;
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
 
@@ -590,3 +597,3 @@ constructor(public ins: Stream<Stream<T>>) {
   public type = 'fold';
-  private out: Stream<R> = null;
+  public out: Stream<R> = null;
   private acc: R; // initialized as seed
@@ -637,3 +644,3 @@
   public type = 'last';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private has: boolean = false;
@@ -705,3 +712,3 @@ private val: T = <T> empty;
   private open: boolean = true;
-  private out: Stream<R> = null;
+  public out: Stream<R> = null;
 
@@ -762,3 +769,3 @@ constructor(public mapOp: MapOperator<T, Stream<R>>) {
   public type = 'map';
-  protected out: Stream<R> = null;
+  public out: Stream<R> = null;
 
@@ -819,3 +826,3 @@ constructor(public project: (t: T) => R,
   public type = 'replaceError';
-  private out: Stream<T> = <Stream<T>> empty;
+  public out: Stream<T> = <Stream<T>> empty;
 
@@ -862,3 +869,3 @@ constructor(public fn: (err: any) => Stream<T>,
   public type = 'startWith';
-  private out: InternalListener<T> = emptyListener;
+  private out: InternalListener<T> = emptyIL;
 
@@ -883,3 +890,3 @@ constructor(public ins: Stream<T>,
   public type = 'take';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private taken: number = 0;
@@ -928,3 +935,2 @@
   protected _ils: Array<InternalListener<T>>; // 'ils' = Internal listeners
-  protected _hil: InternalListener<T>; // 'hil' = Hidden Internal Listener
   protected _stopID: any = empty;
@@ -937,3 +943,2 @@ protected _prod: InternalProducer<T>;
     this._ils = [];
-    this._hil = null;
     this._target = null;
@@ -949,4 +954,2 @@ }
     }
-    const h = this._hil;
-    if (h) h._n(t);
   }
@@ -961,4 +964,2 @@
     }
-    const h = this._hil;
-    if (h) h._e(err);
     this._x();
@@ -974,4 +975,2 @@ }
     }
-    const h = this._hil;
-    if (h) h._c();
     this._x();
@@ -1015,5 +1014,3 @@ }
     const ta = this._target;
-    if (ta && ta._ils.length === 0) {
-      return ta._add(il);
-    }
+    if (ta) return ta._add(il);
     const a = this._ils;
@@ -1032,2 +1029,4 @@ a.push(il);
   _remove(il: InternalListener<T>): void {
+    const ta = this._target;
+    if (ta) return ta._remove(il);
     const a = this._ils;
@@ -1040,12 +1039,36 @@ const i = a.indexOf(il);
         this._stopID = setTimeout(() => p._stop());
+      } else if (a.length === 1) {
+        this._pruneCycles();
       }
-    } else if (this._target) {
-      this._target._remove(il);
     }
   }
 
-  _setHIL(il: InternalListener<T>): void {
-    this._hil = il;
+  // If all paths stemming from `this` stream eventually end at `this`
+  // stream, then we remove the single listener of `this` stream, to
+  // force it to end its execution and dispose resources. This method
+  // assumes as a precondition that this._ils has just one listener.
+  _pruneCycles() {
+    if (this._onlyReachesThis(this)) {
+      this._remove(this._ils[0]);
+    }
   }
 
+  // Checks whether *all* paths starting from `x` will eventually end at
+  // `this` stream, on the stream graph, following edges A->B where B is
+  // a listener of A.
+  _onlyReachesThis(x: InternalListener<any>): boolean {
+    if ((<OutSender<any>><any>x).out === this) {
+      return true;
+    } else if ((<OutSender<any>><any>x).out) {
+      return this._onlyReachesThis((<OutSender<any>><any>x).out);
+    } else if ((<Stream<any>>x)._ils) {
+      for (let i = 0, N = (<Stream<any>>x)._ils.length; i < N; i++) {
+        if (!this._onlyReachesThis((<Stream<any>>x)._ils[i])) return false;
+      }
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   private ctor(): typeof Stream {
@@ -1719,3 +1742,2 @@ return this instanceof MemoryStream ? MemoryStream : Stream;
     this._target = target;
-    target._setHIL(this);
   }
@@ -1722,0 +1744,0 @@

src/extra/concat.ts

@@ -1,6 +0,6 @@
-import {Stream, InternalProducer, InternalListener} from '../core';
+import {Stream, InternalProducer, InternalListener, OutSender} from '../core';
 
-class ConcatProducer<T> implements InternalProducer<T>, InternalListener<T> {
+class ConcatProducer<T> implements InternalProducer<T>, InternalListener<T>, OutSender<T> {
   public type = 'concat';
-  private out: InternalListener<T> = null;
+  public out: Stream<T> = null;
   private i: number = 0;
@@ -11,3 +11,3 @@
 
-  _start(out: InternalListener<T>): void {
+  _start(out: Stream<T>): void {
     this.out = out;
@@ -14,0 +14,0 @@ this.streams[this.i]._add(this);

src/extra/debounce.ts

@@ -5,3 +5,3 @@ import {Operator, Stream} from '../core';
   public type = 'debounce';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private value: T = null;
@@ -8,0 +8,0 @@ private id: any = null;

src/extra/delay.ts

@@ -5,3 +5,3 @@ import {Operator, Stream} from '../core';
   public type = 'delay';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
 
@@ -8,0 +8,0 @@ constructor(public dt: number,

src/extra/dropRepeats.ts

@@ -6,3 +6,3 @@ import {Operator, Stream} from '../core';
   public type = 'dropRepeats';
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
   private v: T = <any> empty;
@@ -9,0 +9,0 @@

src/extra/dropUntil.ts

@@ -1,5 +0,5 @@
-import {Operator, InternalListener, Stream, emptyListener} from '../core';
+import {Operator, InternalListener, Stream, OutSender, emptyIL} from '../core';
 
-class OtherIL<T> implements InternalListener<any> {
-  constructor(private out: Stream<T>,
+class OtherIL<T> implements InternalListener<any>, OutSender<T> {
+  constructor(public out: Stream<T>,
               private op: DropUntilOperator<T>) {
@@ -23,4 +23,4 @@ }
   public type = 'dropUntil';
-  private out: Stream<T> = null;
-  private oil: InternalListener<any> = emptyListener; // oil = other InternalListener
+  public out: Stream<T> = null;
+  private oil: InternalListener<any> = emptyIL; // oil = other InternalListener
   private on: boolean = false;
@@ -27,0 +27,0 @@

src/extra/flattenConcurrently.ts

@@ -1,5 +0,5 @@
-import {Operator, Stream, InternalListener} from '../core';
+import {Operator, Stream, OutSender, InternalListener} from '../core';
 
-class FCIL<T> implements InternalListener<T> {
-  constructor(private out: Stream<T>,
+class FCIL<T> implements InternalListener<T>, OutSender<T> {
+  constructor(public out: Stream<T>,
               private op: FlattenConcOperator<T>) {
@@ -24,3 +24,3 @@ }
   private active: number = 1; // number of outers and inners that have not yet ended
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
 
@@ -27,0 +27,0 @@ constructor(public ins: Stream<Stream<T>>) {

src/extra/flattenSequentially.ts

@@ -26,3 +26,3 @@ import {Operator, Stream, InternalListener} from '../core';
   private seq: Array<Stream<T>> = [];
-  private out: Stream<T> = null;
+  public out: Stream<T> = null;
 
@@ -94,3 +94,3 @@ constructor(public ins: Stream<Stream<T>>) {
  * start imitating it once the previous nested stream completes.
- * 
+ *
  * In essence, `flattenSequentially` concatenates all nested streams.
@@ -97,0 +97,0 @@ *

src/extra/pairwise.ts

@@ -7,3 +7,3 @@ import {Operator, Stream} from '../core';
   private has: boolean = false;
-  private out: Stream<[T, T]> = null;
+  public out: Stream<[T, T]> = null;
 
@@ -10,0 +10,0 @@ constructor(public ins: Stream<T>) {

src/extra/split.ts

@@ -1,5 +0,5 @@
-import {Operator, InternalListener, Stream, emptyListener} from '../core';
+import {Operator, InternalListener, Stream, OutSender, emptyIL} from '../core';
 
-class SeparatorIL<T> implements InternalListener<any> {
-  constructor(private out: Stream<Stream<T>>,
+class SeparatorIL<T> implements InternalListener<any>, OutSender<Stream<T>> {
+  constructor(public out: Stream<Stream<T>>,
               private op: SplitOperator<T>) {
@@ -25,4 +25,4 @@ }
   public curr: Stream<T> = new Stream<T>();
-  private out: Stream<Stream<T>> = null;
-  private sil: InternalListener<any> = emptyListener; // sil = separator InternalListener
+  public out: Stream<Stream<T>> = null;
+  private sil: InternalListener<any> = emptyIL; // sil = separator InternalListener
 
@@ -45,3 +45,3 @@ constructor(public s: Stream<any>, // s = separator
     this.out = null;
-    this.sil = emptyListener;
+    this.sil = emptyIL;
   }
@@ -48,0 +48,0 @@

tests/operator/imitate.ts

@@ -93,3 +93,3 @@ /// <reference path="../../typings/globals/mocha/index.d.ts" />
 
-  it('should not cause leaked cyclic executions', (done) => {
+  it('should not cause leaked cyclic executions (1)', (done) => {
     const expectedProxy = [2, 4, 8, 16, 32 /* inertia due to stopping on next tick */];
@@ -110,3 +110,3 @@ const expectedResult = [2, 4, 8, 16];
     const result$ = source$.compose(delay(100)).compose(s => <Stream<number>> s);
-    proxy$.imitate(result$)
+    proxy$.imitate(result$);
 
@@ -128,2 +128,35 @@ result$.take(4).addListener({
 
+  it('should not cause leaked cyclic executions (2)', (done) => {
+    const expectedProxy = [2, 4, 8, 16, 32 /* inertia due to stopping on next tick */];
+    const expectedResult = [2, 4, 8, 16];
+
+    const proxy$ = xs.create<number>();
+    const source$ = proxy$.startWith(1).map(x => x * 2)
+      .debug(x => {
+        try {
+          assert.equal(expectedProxy.length > 0, true,
+            'should be expecting the next value ' + x);
+          assert.equal(x, expectedProxy.shift());
+        } catch (err) {
+          done(err);
+        }
+      });
+    const result$ = source$.compose(delay(100)).compose(s => <Stream<number>> s);
+    proxy$.imitate(result$)
+
+    source$.take(4).addListener({
+      next: (x: number) => {
+        assert.equal(x, expectedResult.shift());
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.equal(expectedProxy.length, 1); // still waiting for 32
+        assert.equal(expectedResult.length, 0);
+        setTimeout(() => {
+          done();
+        }, 1000);
+      },
+    });
+  });
+
   it('should not by itself start the target stream execution', (done) => {
@@ -130,0 +163,0 @@ let nextDelivered = false;

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

730737

increased by0.62%

Lines of code

13098

increased by0.65%

No dependency changes