From 7279ad80fa597721e7945827eba5d33eabbe8c5c Mon Sep 17 00:00:00 2001
From: TylorS <tlsteinberger167@gmail.com>
Date: Mon, 29 Feb 2016 18:38:45 -0500
Subject: [PATCH] feat(remember): implement RememeberProducer

Adds initial implementation of .remember()
Simply creates a stream that will replay the last event.
---
 src/operator/RememberProducer.ts | 39 ++++++++++++++++++++++++++++++++
 1 file changed, 39 insertions(+)
 create mode 100644 src/operator/RememberProducer.ts

diff --git a/src/operator/RememberProducer.ts b/src/operator/RememberProducer.ts
new file mode 100644
index 0000000..dc4129e
--- /dev/null
+++ b/src/operator/RememberProducer.ts
@@ -0,0 +1,39 @@
+import {Observer} from '../Observer';
+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 p: RememberProducer<T>) {
+  }
+
+  next(t: T) {
+    this.out.next(t);
+    this.out.value = t;
+  }
+
+  error(err: any) {
+    this.out.error(err);
+  }
+
+  end() {
+    this.out.end();
+  }
+}
+
+export class RememberProducer<T> implements Producer<T> {
+  public proxy: Observer<T> = emptyObserver;
+  public value: any;
+
+  constructor(public ins: Stream<T>) {
+  }
+
+  start(out: Stream<T>): void {
+    this.ins.subscribe(this.proxy = new Proxy(out, this));
+  }
+
+  stop(): void {
+    this.ins.unsubscribe(this.proxy);
+  }
+}
\ No newline at end of file