feat(Observer): rename complete() callback to end()

perf/runners.js

@@ -103,7 +103,7 @@ function runRx5(deferred, rxStream) {
 function runXStream(deferred, xstream) {
   xstream.subscribe({
     next: noop,
-    complete: function() {
+    end: function() {
       deferred.resolve();
     },
     error: function(e) {

src/Observer.ts

@@ -1,5 +1,5 @@
 export interface Observer<T> {
   next: (x: T) => void;
   error: (err: any) => void;
-  complete: () => void;
+  end: () => void;
 }

src/Stream.ts

@@ -37,13 +37,13 @@ export class Stream<T> implements Observer<T> {
     }
   }
 
-  complete(): void {
+  end(): void {
     const len = this.observers.length;
     if (len === 1) {
-      this.observers[0].complete();
+      this.observers[0].end();
     } else {
       for (let i = 0; i < len; i++) {
-        this.observers[i].complete();
+        this.observers[i].end();
       }
     }
   }

src/factory/from.ts

@@ -13,7 +13,7 @@ class FromProducer<T> implements Producer<T> {
     for (let i = 0; i < L; i++) {
       out.next(a[i]);
     }
-    out.complete();
+    out.end();
   }
 
   stop(): void {

src/operator/DebugProducer.ts

@@ -21,7 +21,7 @@ export class DebugProducer<T> implements Producer<T> {
         out.next(t);
       },
       error: (err) => out.error(err),
-      complete: () => out.complete(),
+      end: () => out.end(),
     };
     this.ins.subscribe(this.proxy);
   }

src/operator/FilterProducer.ts

@@ -16,8 +16,8 @@ export class Proxy<T> implements Observer<T> {
     this.out.error(err);
   }
 
-  complete() {
-    this.out.complete();
+  end() {
+    this.out.end();
   }
 }
 

src/operator/FoldProducer.ts

@@ -17,8 +17,8 @@ export class Proxy<T, R> implements Observer<T> {
     this.out.error(err);
   }
 
-  complete() {
-    this.out.complete();
+  end() {
+    this.out.end();
   }
 }
 

src/operator/LastProducer.ts

@@ -18,12 +18,12 @@ export class Proxy<T> implements Observer<T> {
     this.out.error(err);
   }
 
-  complete() {
+  end() {
     const p = this.p;
     const out = this.out;
     if (p.has) {
       out.next(p.val);
-      out.complete();
+      out.end();
     } else {
       out.error('TODO show proper error');
     }

src/operator/MapProducer.ts

@@ -16,8 +16,8 @@ export class Proxy<T, U> implements Observer<T> {
     this.out.error(err);
   }
 
-  complete() {
-    this.out.complete();
+  end() {
+    this.out.end();
   }
 }
 

src/operator/SkipProducer.ts

@@ -17,7 +17,7 @@ export class SkipProducer<T> implements Producer<T> {
         if (this.skipped++ >= this.max) out.next(t);
       },
       error: (err) => out.error(err),
-      complete: () => out.complete(),
+      end: () => out.end(),
     };
     this.ins.subscribe(this.proxy);
   }

src/operator/TakeProducer.ts

@@ -17,12 +17,12 @@ export class TakeProducer<T> implements Producer<T> {
         if (this.taken++ < this.max) {
           out.next(t);
         } else {
-          out.complete();
+          out.end();
           this.stop();
         }
       },
       error: (err) => out.error(err),
-      complete: () => out.complete(),
+      end: () => out.end(),
     };
     this.ins.subscribe(this.proxy);
   }

src/utils/emptyObserver.ts

@@ -2,5 +2,5 @@ import {Observer} from '../Observer';
 export var emptyObserver: Observer<any> = {
   next: null,
   error: null,
-  complete: null,
+  end: null,
 };

tests/factory/from.ts

@@ -11,7 +11,7 @@ describe('Stream.prototype.from', () => {
         assert.equal(x, expected.shift());
       },
       error: done.fail,
-      complete: () => {
+      end: () => {
         assert.equal(expected.length, 0);
         done();
       },

tests/operator/debug.ts

@@ -16,7 +16,7 @@ describe('Stream.prototype.debug', () => {
         }
       },
       error: done.fail,
-      complete: done.fail,
+      end: done.fail,
     };
     stream.subscribe(observer);
   });

tests/operator/filter.ts

@@ -14,7 +14,7 @@ describe('Stream.prototype.filter', () => {
         }
       },
       error: done.fail,
-      complete: done.fail,
+      end: done.fail,
     };
     stream.subscribe(observer);
   });

tests/operator/fold.ts

@@ -10,7 +10,7 @@ describe('Stream.prototype.fold', () => {
         assert.equal(x, expected.shift());
       },
       error: done.fail,
-      complete: () => {
+      end: () => {
         assert.equal(expected.length, 0);
         stream.unsubscribe(observer);
         done();

tests/operator/last.ts

@@ -10,7 +10,7 @@ describe('Stream.prototype.last', () => {
         assert.equal(x, expected.shift());
       },
       error: done.fail,
-      complete: () => {
+      end: () => {
         assert.equal(expected.length, 0);
         stream.unsubscribe(observer);
         done();

tests/operator/map.ts

@@ -14,7 +14,7 @@ describe('Stream.prototype.map', () => {
         }
       },
       error: done.fail,
-      complete: done.fail,
+      end: done.fail,
     };
     stream.subscribe(observer);
   });

tests/operator/skip.ts

@@ -14,7 +14,7 @@ describe('Stream.prototype.skip', () => {
         }
       },
       error: done.fail,
-      complete: done.fail,
+      end: done.fail,
     };
     stream.subscribe(observer);
   });

tests/operator/take.ts

@@ -10,7 +10,7 @@ describe('Stream.prototype.take', () => {
         assert.equal(x, expected.shift());
       },
       error: done.fail,
-      complete: () => {
+      end: () => {
         assert.equal(expected.length, 0);
         stream.unsubscribe(observer);
         done();

tests/stream.ts

@@ -14,7 +14,7 @@ describe('Stream', () => {
         }
       },
       error: done.fail,
-      complete: done.fail,
+      end: done.fail,
     };
     stream.subscribe(observer);
   });