on$: fix unsubscribe and add cleanup

Add an example of concurrent subscription to gun in the
view with a button to subscribe and unsubscribe.

examples/angular/src/app/app.component.html

@@ -8,4 +8,6 @@
   <ul>
     <li *ngFor="let todo of todos$ | async | pairs" (click)="delete(todo[0])">{{todo[1]}}</li>
   </ul>
+  <button (click)="sub()" *ngIf="!todosSub">Log in console</button>
+  <button (click)="unsub()" *ngIf="!!todosSub">Stop logging</button>
 </div>

examples/angular/src/app/app.component.ts

@@ -1,7 +1,7 @@
 import { Component, OnInit } from '@angular/core';
-import Gun from 'gun/gun';
 import { Observable } from 'rxjs/Observable';
-import { omit } from 'underscore';
+import { Subscription } from 'rxjs/Subscription';
+import Gun from 'gun/gun';
 
 import { GunDb } from 'app/gun.service';
 import { on$ } from 'app/gun.helper';
@@ -15,8 +15,9 @@ export class AppComponent implements OnInit {
   newTodo = '';
 
   todos = this.db.gun.get('todos');
-  todos$: Observable<string[]> = on$(this.todos)
-    .map(o => omit(o, '_'));
+  todos$: Observable<string[]> = on$(this.todos);
+
+  todosSub: Subscription;
 
   constructor(private db: GunDb) { }
 
@@ -32,4 +33,12 @@ export class AppComponent implements OnInit {
   delete(key: string) {
     this.todos.get(key).put(null);
   }
+
+  sub() {
+    this.todosSub = this.todos$.subscribe(v => console.log(v));
+  }
+
+  unsub() {
+    this.todosSub.unsubscribe();
+  }
 }

examples/angular/src/app/gun.helper.ts

@@ -1,12 +1,24 @@
 import { Observable } from 'rxjs/Observable';
 import { Gun } from 'gun/gun';
+import { pick } from 'underscore';
 
-export function on$(node): Observable<any> {
+export function on$(node, cleanup = true): Observable<any> {
     return Observable.fromEventPattern(
-        // note: passing directly node.on doesn't seem to work...
-        h => node.on(v => h(v)),
-        // TODO this is incorrect
-        (_, s) => s.off()
+        h => {
+            // there is no way to off() an on() until at least one value is trigerred
+            // so that we can access the event listener to off() it
+            const signal = { stop: false };
+            node.on((data, key, at, ev) => {
+                if (signal.stop) {
+                    ev.off();
+                } else {
+                    // modifying data directly does not seem to work...
+                    h(cleanup ? pick(data, (v, k, o) => v !== null && k !== '_') : data);
+                }
+            });
+            return signal;
+        },
+        (h, signal) => { signal.stop = true; }
     );
 }