perf(take): improve take() perf by using Proxy Observer class

src/operator/TakeProducer.ts

@@ -3,6 +3,30 @@ import {Producer} from '../Producer';
 import {Stream} from '../Stream';
 import {emptyObserver} from '../utils/emptyObserver';
 
+export class Proxy<T> implements Observer<T> {
+  constructor(public out: Stream<T>,
+              public prod: TakeProducer<T>) {
+  }
+
+  next(t: T) {
+    const {prod, out} = this;
+    if (prod.taken++ < prod.max) {
+      out.next(t);
+    } else {
+      out.end();
+      prod.stop();
+    }
+  }
+
+  error(err: any) {
+    this.out.error(err);
+  }
+
+  end() {
+    this.out.end();
+  }
+}
+
 export class TakeProducer<T> implements Producer<T> {
   public proxy: Observer<T> = emptyObserver;
   public taken: number = 0;
@@ -12,19 +36,7 @@ export class TakeProducer<T> implements Producer<T> {
   }
 
   start(out: Stream<T>): void {
-    this.proxy = {
-      next: (t: T) => {
-        if (this.taken++ < this.max) {
-          out.next(t);
-        } else {
-          out.end();
-          this.stop();
-        }
-      },
-      error: (err) => out.error(err),
-      end: () => out.end(),
-    };
-    this.ins.subscribe(this.proxy);
+    this.ins.subscribe(this.proxy = new Proxy(out, this));
   }
 
   stop(): void {