feat(operator): implement take operator with TakeMachine

src/Stream.ts

@@ -2,6 +2,7 @@ import {Observer} from './Observer';
 import {Machine} from './Machine';
 import {MapMachine} from './operator/MapMachine';
 import {FilterMachine} from './operator/FilterMachine';
+import {TakeMachine} from './operator/TakeMachine';
 
 export class Stream<T> implements Observer<T> {
   public observers: Array<Observer<T>>;
@@ -53,4 +54,8 @@ export class Stream<T> implements Observer<T> {
   filter(predicate: (t: T) => boolean): Stream<T> {
     return new Stream<T>(new FilterMachine(predicate, this));
   }
+
+  take(amount: number): Stream<T> {
+    return new Stream<T>(new TakeMachine(amount, this));
+  }
 }

src/operator/TakeMachine.ts

@@ -0,0 +1,33 @@
+import {Observer} from '../Observer';
+import {Machine} from '../Machine';
+import {Stream} from '../Stream';
+
+export class TakeMachine<T> implements Machine<T> {
+  public proxy: Observer<T>;
+  public taken: number;
+
+  constructor(public max: number,
+              public inStream: Stream<T>) {
+    this.taken = 0;
+  }
+
+  start(outStream: Observer<T>): void {
+    this.proxy = {
+      next: (t: T) => {
+        if (this.taken++ < this.max) {
+          outStream.next(t);
+        } else {
+          outStream.complete();
+          this.stop();
+        }
+      },
+      error: outStream.error,
+      complete: outStream.complete,
+    };
+    this.inStream.subscribe(this.proxy);
+  }
+
+  stop(): void {
+    this.inStream.unsubscribe(this.proxy);
+  }
+}

tests/stream.ts

@@ -57,3 +57,22 @@ describe('Stream.prototype.filter', () => {
     stream.subscribe(observer);
   });
 });
+
+describe('Stream.prototype.take', () => {
+  it('should allow specifying max amount to take from input stream', (done) => {
+    const stream = xs.interval(50).take(4)
+    const expected = [0, 1, 2, 3];
+    let observer = {
+      next: (x: number) => {
+        assert.equal(x, expected.shift());
+      },
+      error: done.fail,
+      complete: () => {
+        assert.equal(expected.length, 0);
+        stream.unsubscribe(observer);
+        done();
+      },
+    };
+    stream.subscribe(observer);
+  });
+});