fix(Database): use Zone scheduler for object and list factories

Since Firebase is adding WebSocket listeners before the Angular Zone exists,
all callbacks when data is received are being executed outside of Angular's
zone and are not triggering change detection. This fix uses the existing
ZoneScheduler, which is already used in auth.onAuth, to cause the observables
created by af.list() and af.object() to be emitted inside the current Zone at
the time of calling .list() or .object().

See angular/angular#4603 for progress on making RxJS more zone-aware.

Fixes #637

Author: jeffbcross

src/database/firebase_list_factory.spec.ts

@@ -541,6 +541,26 @@ describe('FirebaseListFactory', () => {
       });
     });
 
+    it('should emit values in the observable creation zone', (done: any) => {
+      Zone.current.fork({
+        name: 'newZone'
+      })
+      .run(() => {
+        // Creating a new observable so that the current zone is captured.
+        subscription = FirebaseListFactory(`${rootDatabaseUrl}/questions`)
+          .filter(d => d
+            .map(v => v.$value)
+            .indexOf('in-the-zone') > -1)
+          .subscribe(data => {
+            expect(Zone.current.name).toBe('newZone');
+            done();
+          });
+      });
+
+      ref.remove(() => {
+        ref.push('in-the-zone');
+      });
+    });
   });
 });
 

src/database/firebase_list_factory.ts

@@ -2,6 +2,7 @@ import * as firebase from 'firebase';
 import { AFUnwrappedDataSnapshot } from '../interfaces';
 import { FirebaseListObservable } from './firebase_list_observable';
 import { Observer } from 'rxjs/Observer';
+import { observeOn } from 'rxjs/operator/observeOn';
 import { observeQuery } from './query_observable';
 import { Query, FirebaseListFactoryOpts } from '../interfaces';
 import * as utils from '../utils';
@@ -152,7 +153,9 @@ function firebaseListObservable(ref: firebase.database.Reference | firebase.data
       ref.off('child_changed', chgFn);
     }
   });
-  return listObs;
+
+  // TODO: should be in the subscription zone instead
+  return observeOn.call(listObs, new utils.ZoneScheduler(Zone.current));
 }
 
 export function onChildAdded(arr:any[], child:any, prevKey:string): any[] {

src/database/firebase_object_factory.spec.ts

@@ -136,5 +136,24 @@ describe('FirebaseObjectFactory', () => {
       subscription.unsubscribe();
       expect(firebaseSpy).toHaveBeenCalled();
     });
+
+    it('should emit values in the observable creation zone', (done: any) => {
+      Zone.current.fork({
+        name: 'newZone'
+      })
+      .run(() => {
+        // Creating a new observable so that the current zone is captured.
+        subscription = FirebaseObjectFactory(`${rootDatabaseUrl}/questions/${i}`)
+          .filter(d => d.$value === 'in-the-zone')
+          .subscribe(data => {
+            expect(Zone.current.name).toBe('newZone');
+            done();
+          });
+      });
+
+      ref.remove(() => {
+        ref.set('in-the-zone');
+      });
+    });
   });
 });

src/database/firebase_object_factory.ts

@@ -1,5 +1,6 @@
 import { FirebaseObjectObservable } from './index';
 import { Observer } from 'rxjs/Observer';
+import { observeOn } from 'rxjs/operator/observeOn';
 import * as firebase from 'firebase';
 import * as utils from '../utils';
 import { Query } from '../interfaces';
@@ -17,7 +18,7 @@ export function FirebaseObjectFactory (
     isRef: () => ref = <firebase.database.Reference>absoluteUrlOrDbRef
   });
 
-  return new FirebaseObjectObservable((obs: Observer<any>) => {
+  const objectObservable = new FirebaseObjectObservable((obs: Observer<any>) => {
     let fn = ref.on('value', (snapshot: firebase.database.DataSnapshot) => {
       obs.next(preserveSnapshot ? snapshot : utils.unwrapMapFn(snapshot))
     }, err => {
@@ -26,4 +27,7 @@ export function FirebaseObjectFactory (
 
     return () => ref.off('value', fn);
   }, ref);
+
+  // TODO: should be in the subscription zone instead
+  return observeOn.call(objectObservable, new utils.ZoneScheduler(Zone.current));
 }