feat(database): allow to unsubscribe from observables

* Subscription management for database and reset support

Subscriptions to firebase are tracked in an array.
Reset subscriptions and caches to be rebuild from the remote db.

* enable to clear object and list subscriptions

related: #30

src/database/database.spec.ts

@@ -423,6 +423,46 @@ describe('Service: AngularFireOfflineDatabase', () => {
     })));
   });
 
+  describe('Unsubscribe', () => {
+    it('should unsubscribe from an object', done => {
+      inject([AngularFireOfflineDatabase], (service: AngularFireOfflineDatabase) => {
+        let newValue = { val: () => { return 'abc23-7'; } };
+        let afoObject = service.object('/slug-2');
+
+        afoObject.subscribe(object => {
+          expect(object.$value).toBe('abc23-7');
+        });
+        mockAngularFireDatabase.update('object', newValue);
+
+        afoObject.complete();
+
+        expect((<any>afoObject).ref.observers.length).toBe(0);
+        setTimeout(done);
+      })();
+    });
+
+    it('should unsubscribe from a list', done => {
+      inject([AngularFireOfflineDatabase], (service: AngularFireOfflineDatabase) => {
+        const key = '/slug-2';
+        let newValue = [
+          { val: () => { return 'xyz'; } }
+        ];
+
+        service.processing.current = false;
+        const afoList = service.list(key);
+        afoList.subscribe(list => {
+          expect(list[0].$value).toBe('xyz');
+        });
+        mockAngularFireDatabase.update('list', newValue);
+
+        afoList.complete();
+
+        expect((<any>afoList).ref.observers.length).toBe(0);
+        setTimeout(done);
+      })();
+    });
+  });
+
   it('should return an unwrapped null value', async(inject([AngularFireOfflineDatabase], (service: AngularFireOfflineDatabase) => {
     let newValue = { val: () => { return null; } };
     service.object('/slug-2').subscribe(object => {

src/database/database.ts

@@ -133,6 +133,25 @@ export class AngularFireOfflineDatabase {
     if (!(key in this.objectCache)) { this.setupObject(key, options); }
     return this.objectCache[key].sub;
   }
+
+  /**
+   * Unsubscribes from all firebase subscriptions and clears the cache
+   *
+   * - run before e.g. logout to make sure there are no permission errors.
+   * @TODO: might cause data loss of unwritten data?
+   */
+  reset() {
+    Object.keys(this.objectCache).forEach(key => {
+      this.objectCache[key].sub.complete();
+    });
+    Object.keys(this.listCache).forEach(key => {
+      this.listCache[key].sub.complete();
+    });
+    this.objectCache = {};
+    this.listCache = {};
+    this.localForage.clear();
+  };
+
   /**
    * Retrives a list if locally stored on the device
    * - Lists are stored as individual objects, to allow for better offline reuse.
@@ -189,8 +208,9 @@ export class AngularFireOfflineDatabase {
       offlineInit: false,
       sub: new AfoObjectObservable(ref, this.localUpdateService)
     };
+
     // Firebase
-    ref.subscribe(snap => {
+    const subscription = ref.subscribe(snap => {
       this.objectCache[key].loaded = true;
       const cacheValue = unwrap(snap.key, snap.val(), () => !isNil(snap.val()));
       if (this.processing.current) {
@@ -200,6 +220,11 @@ export class AngularFireOfflineDatabase {
       }
       this.localForage.setItem(`read/object${key}`, snap.val());
     });
+
+    this.objectCache[key].sub.subscribe({
+      complete: () => subscription.unsubscribe()
+    });
+
     // Local
     this.localForage.getItem(`read/object${key}`).then(value => {
       if (!this.objectCache[key].loaded && value !== null) {
@@ -272,8 +297,9 @@ export class AngularFireOfflineDatabase {
       offlineInit: false,
       sub: new AfoListObservable(ref, this.localUpdateService)
     };
+
     // Firebase
-    ref.subscribe(value => {
+    const subscription = ref.subscribe(value => {
       this.listCache[key].loaded = true;
       const cacheValue = value.map(snap => unwrap(snap.key, snap.val(), () => !isNil(snap.val())));
       if (this.processing.current) {
@@ -283,6 +309,11 @@ export class AngularFireOfflineDatabase {
       }
       this.setList(key, value);
     });
+
+    this.listCache[key].sub.subscribe({
+      complete: () => subscription.unsubscribe()
+    });
+
     // Local
     this.getList(key);
   }