fix(operator): fix all operators redirection of error/complete

staltz · Feb 25, 2016 · 2caa2ca · 2caa2ca
1 parent e76acef
commit 2caa2ca
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 15 deletions.
diff --git a/src/operator/DebugMachine.ts b/src/operator/DebugMachine.ts
@@ -9,7 +9,7 @@ export class DebugMachine<T> implements Machine<T> {
               public inStream: Stream<T>) {
   }
 
-  start(outStream: Observer<T>): void {
+  start(outStream: Stream<T>): void {
     this.proxy = {
       next: (t: T) => {
         if (this.spy) {
@@ -19,8 +19,8 @@ export class DebugMachine<T> implements Machine<T> {
         }
         outStream.next(t);
       },
-      error: outStream.error,
-      complete: outStream.complete,
+      error: (err) => outStream.error(err),
+      complete: () => outStream.complete(),
     };
     this.inStream.subscribe(this.proxy);
   }

diff --git a/src/operator/FilterMachine.ts b/src/operator/FilterMachine.ts
@@ -9,13 +9,13 @@ export class FilterMachine<T> implements Machine<T> {
               public inStream: Stream<T>) {
   }
 
-  start(outStream: Observer<T>): void {
+  start(outStream: Stream<T>): void {
     this.proxy = {
       next: (t: T) => {
         if (this.predicate(t)) outStream.next(t);
       },
-      error: outStream.error,
-      complete: outStream.complete,
+      error: (err) => outStream.error(err),
+      complete: () => outStream.complete(),
     };
     this.inStream.subscribe(this.proxy);
   }

diff --git a/src/operator/MapMachine.ts b/src/operator/MapMachine.ts
@@ -9,11 +9,11 @@ export class MapMachine<T, U> implements Machine<U> {
               public inStream: Stream<T>) {
   }
 
-  start(outStream: Observer<U>): void {
+  start(outStream: Stream<U>): void {
     this.proxy = {
       next: (t: T) => outStream.next(this.projection(t)),
-      error: outStream.error,
-      complete: outStream.complete,
+      error: (err) => outStream.error(err),
+      complete: () => outStream.complete(),
     };
     this.inStream.subscribe(this.proxy);
   }

diff --git a/src/operator/SkipMachine.ts b/src/operator/SkipMachine.ts
@@ -11,13 +11,13 @@ export class SkipMachine<T> implements Machine<T> {
     this.skipped = 0;
   }
 
-  start(outStream: Observer<T>): void {
+  start(outStream: Stream<T>): void {
     this.proxy = {
       next: (t: T) => {
         if (this.skipped++ >= this.max) outStream.next(t);
       },
-      error: outStream.error,
-      complete: outStream.complete,
+      error: (err) => outStream.error(err),
+      complete: () => outStream.complete(),
     };
     this.inStream.subscribe(this.proxy);
   }

diff --git a/src/operator/TakeMachine.ts b/src/operator/TakeMachine.ts
@@ -11,7 +11,7 @@ export class TakeMachine<T> implements Machine<T> {
     this.taken = 0;
   }
 
-  start(outStream: Observer<T>): void {
+  start(outStream: Stream<T>): void {
     this.proxy = {
       next: (t: T) => {
         if (this.taken++ < this.max) {
@@ -21,8 +21,8 @@ export class TakeMachine<T> implements Machine<T> {
           this.stop();
         }
       },
-      error: outStream.error,
-      complete: outStream.complete,
+      error: (err) => outStream.error(err),
+      complete: () => outStream.complete(),
     };
     this.inStream.subscribe(this.proxy);
   }