test(TakeOperator): test for recursive case of take operator

staltz#158

tests/operator/take.ts

@@ -1,7 +1,8 @@
 /// <reference path="../../typings/globals/mocha/index.d.ts" />
 /// <reference path="../../typings/globals/node/index.d.ts" />
-import xs, {Stream, MemoryStream} from '../../src/index';
+import xs, { Stream, MemoryStream } from '../../src/index';
 import * as assert from 'assert';
+import { listeners } from "cluster";
 
 describe('Stream.prototype.take', () => {
   it('should allow specifying max amount to take from input stream', (done: any) => {
@@ -79,4 +80,40 @@ describe('Stream.prototype.take', () => {
       },
     });
   });
+
+  it('should terminate properly when "next" function recursively calls itself', (done: any) => {
+    const producer = {
+      start: (listener: any) => {
+        this.listener = listener;
+        listener.next(1);
+      },
+      _n: (value: any) => {
+        const listener = this.listener;
+        listener && listener.next(value);
+      },
+      _e: (value: string) => {
+        const listener = this.listener;
+        listener && listener.error(value);
+      },
+      stop: () => this.listener = null,
+      listener: null
+    };
+    const stream = xs.create(producer);
+
+    let nextCount = 0;
+    stream.take(1).addListener({
+      next: (x: number) => {
+        nextCount++;
+        if (nextCount > 1) {
+          producer._e('next should not be called');
+        } else {
+          producer._n(x);
+        }
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        done();
+      }
+    });
+  });
 });