feat(operator): implement skip operator with SkipMachine

src/Stream.ts

@@ -3,6 +3,7 @@ import {Machine} from './Machine';
 import {MapMachine} from './operator/MapMachine';
 import {FilterMachine} from './operator/FilterMachine';
 import {TakeMachine} from './operator/TakeMachine';
+import {SkipMachine} from './operator/SkipMachine';
 
 export class Stream<T> implements Observer<T> {
   public observers: Array<Observer<T>>;
@@ -58,4 +59,8 @@ export class Stream<T> implements Observer<T> {
   take(amount: number): Stream<T> {
     return new Stream<T>(new TakeMachine(amount, this));
   }
+
+  skip(amount: number): Stream<T> {
+    return new Stream<T>(new SkipMachine(amount, this));
+  }
 }

src/operator/SkipMachine.ts

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

tests/stream.ts

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