refactor(Observable): get rid of side-effects from rxjs throughout src

Also removed side-effects from tests to verify that code under test wouldn't
accidentally rely on a patched operator.

Author: jeffbcross

rollup-globals.js

@@ -1,6 +1,7 @@
 export default (mod) => {
   if (mod === 'rxjs') return 'Rx';
   if (mod.indexOf('rxjs/operator') === 0) return `Rx.Observable.prototype`;
+  if (mod.indexOf('rxjs/observable') === 0) return `Rx.Observable`;
   if (mod === 'rxjs/scheduler/queue') return 'Rx.Scheduler';
   if (mod.indexOf('rxjs/') === 0) return 'Rx';
 

src/angularfire2.spec.ts

@@ -17,10 +17,6 @@ import {
 } from './angularfire2';
 import { Subscription } from 'rxjs/Subscription';
 import { COMMON_CONFIG, ANON_AUTH_CONFIG } from './test-config';
-import 'rxjs/add/operator/toPromise';
-import 'rxjs/add/operator/take';
-import 'rxjs/add/operator/do';
-import 'rxjs/add/operator/delay';
 
 describe('angularfire', () => {
   var subscription:Subscription;

src/auth/auth.spec.ts

@@ -6,7 +6,7 @@ import {
   TestBed,
   inject
 } from '@angular/core/testing';
-import 'rxjs/add/operator/do';
+import { _do } from 'rxjs/operator/do';
 
 import {
   defaultFirebase,
@@ -97,9 +97,9 @@ describe('Zones', () => {
     });
     ngZone.run(() => {
       var afAuth = new AngularFireAuth(new FirebaseSdkAuthBackend(app), window.location);
-      afAuth
-        .take(1)
-        .do(_ => {
+      var authObs = afAuth.take(1);
+
+      _do.call(authObs, _ => {
           expect(Zone.current.name).toBe('ngZone');
         })
         .subscribe(() => {

src/auth/auth.ts

@@ -14,11 +14,9 @@ import {
   FirebaseAuthState,
   stripProviderId
 } from './auth_backend';
-import 'rxjs/add/operator/mergeMap';
-import 'rxjs/add/operator/take';
-import 'rxjs/add/operator/concat';
-import 'rxjs/add/operator/skip';
-import 'rxjs/add/observable/of';
+import { mergeMap } from 'rxjs/operator/mergeMap';
+import { of as observableOf } from 'rxjs/observable/of';
+import { map } from 'rxjs/operator/map';
 
 const kBufferSize = 1;
 
@@ -35,26 +33,26 @@ export class AngularFireAuth extends ReplaySubject<FirebaseAuthState> {
     super(kBufferSize);
 
     let firstPass = true;
-    this._authBackend.onAuth()
-      .mergeMap((authState: FirebaseAuthState) => {
-        if (firstPass) {
-          firstPass = false;
-          if(['http:', 'https:'].indexOf(loc.protocol) > -1) {
-            // Only call getRedirectResult() in a browser
-            return this._authBackend.getRedirectResult()
-              .map((userCredential: firebase.auth.UserCredential) => {
-                if (userCredential && userCredential.credential) {
-                  authState = attachCredentialToAuthState(authState, userCredential.credential, userCredential.credential.provider);
-                  this._credentialCache[userCredential.credential.provider] = userCredential.credential;
-                }
-                return authState;
-              });
-          }
-
+    let onAuth = this._authBackend.onAuth();
+
+    mergeMap.call(onAuth, (authState: FirebaseAuthState) => {
+      if (firstPass) {
+        firstPass = false;
+        if(['http:', 'https:'].indexOf(loc.protocol) > -1) {
+          // Only call getRedirectResult() in a browser
+          return map.call(this._authBackend.getRedirectResult(), (userCredential: firebase.auth.UserCredential) => {
+              if (userCredential && userCredential.credential) {
+                authState = attachCredentialToAuthState(authState, userCredential.credential, userCredential.credential.provider);
+                this._credentialCache[userCredential.credential.provider] = userCredential.credential;
+              }
+              return authState;
+            });
         }
-        return Observable.of(authState);
-      })
-      .subscribe((authData: FirebaseAuthState) => this._emitAuthData(authData));
+
+      }
+      return observableOf(authState);
+    })
+    .subscribe((authData: FirebaseAuthState) => this._emitAuthData(authData));
   }
 
   public login(config?: AuthConfiguration): firebase.Promise<FirebaseAuthState>;

src/auth/firebase_sdk_auth_backend.ts

@@ -20,9 +20,9 @@ const {
   TwitterAuthProvider
 } = auth;
 
-import 'rxjs/add/operator/map';
-import 'rxjs/add/observable/fromPromise';
-import 'rxjs/add/operator/observeOn';
+import { map } from 'rxjs/operator/map';
+import { fromPromise } from 'rxjs/observable/fromPromise';
+import { observeOn } from 'rxjs/operator/observeOn';
 
 @Injectable()
 export class FirebaseSdkAuthBackend extends AuthBackend {
@@ -43,20 +43,21 @@ export class FirebaseSdkAuthBackend extends AuthBackend {
   }
 
   onAuth(): Observable<FirebaseAuthState> {
-    return Observable.create((observer: Observer<FirebaseAuthState>) => {
+    let stateChange = Observable.create((observer: Observer<FirebaseAuthState>) => {
       return this._fbAuth.onAuthStateChanged(observer);
-    })
-    .map((user: firebase.User) => {
+    });
+    let authState = map.call(stateChange, (user: firebase.User) => {
       if (!user) return null;
       return authDataToAuthState(user, user.providerData[0]);
-    })
+    });
+
     /**
      * TODO: since the auth service automatically subscribes to this before
      * any user, it will run in the Angular zone, instead of the subscription
      * zone. The auth service should be refactored to capture the subscription
      * zone and not use a ReplaySubject.
     **/
-    .observeOn(new ZoneScheduler(Zone.current));
+    return observeOn.call(authState, new ZoneScheduler(Zone.current));
   }
 
   unauth(): void {
@@ -101,7 +102,7 @@ export class FirebaseSdkAuthBackend extends AuthBackend {
   }
 
   getRedirectResult(): Observable<firebase.auth.UserCredential> {
-    return Observable.fromPromise(castPromise<firebase.auth.UserCredential>(this._fbAuth.getRedirectResult()));
+    return fromPromise(castPromise<firebase.auth.UserCredential>(this._fbAuth.getRedirectResult()));
   }
 
   private _enumToAuthProvider(providerId: AuthProviders): any {

src/database/firebase_list_factory.spec.ts

@@ -24,16 +24,16 @@ import * as utils from '../utils';
 import { Query, AFUnwrappedDataSnapshot } from '../interfaces';
 import { Subscription, Observable, Subject } from 'rxjs';
 import { COMMON_CONFIG, ANON_AUTH_CONFIG } from '../test-config';
-import 'rxjs/add/operator/do';
-import 'rxjs/add/operator/skip';
-import 'rxjs/add/operator/take';
+import { _do } from 'rxjs/operator/do';
+import { skip } from 'rxjs/operator/skip';
+import { take } from 'rxjs/operator/take';
+import { toPromise } from 'rxjs/operator/toPromise';
 
 const rootDatabaseUrl = COMMON_CONFIG.databaseURL;
 
 function queryTest(observable: Observable<any>, subject: Subject<any>, done: any) {
   let nexted = false;
-  observable
-    .take(2)
+  skipAndTake(observable, 2)
     .subscribe(val => {
       if (!nexted) {
         subject.next('2');
@@ -378,7 +378,7 @@ describe('FirebaseListFactory', () => {
 
     it('should emit only when the initial data set has been loaded', (done: any) => {
       (<any>questions)._ref.set([{ initial1: true }, { initial2: true }, { initial3: true }, { initial4: true }])
-        .then(() => questions.take(1).toPromise())
+        .then(() => toPromise.call(skipAndTake(questions, 1)))
         .then((val: any[]) => {
           expect(val.length).toBe(4);
         })
@@ -389,10 +389,8 @@ describe('FirebaseListFactory', () => {
 
 
     it('should emit a new value when a child moves', (done: any) => {
-      subscription = questions
-        .skip(2)
-        .take(1)
-        .do((data: any) => {
+       let question = skipAndTake(questions, 1, 2)
+       subscription = _do.call(question, (data: any) => {
           expect(data.length).toBe(2);
           expect(data[0].push2).toBe(true);
           expect(data[1].push1).toBe(true);
@@ -412,30 +410,26 @@ describe('FirebaseListFactory', () => {
     it('should emit unwrapped data by default', (done: any) => {
       ref.remove(() => {
         ref.push({ unwrapped: true }, () => {
-          subscription = questions
-            .take(1)
-            .do((data: any) => {
-              expect(data.length).toBe(1);
-              expect(data[0].unwrapped).toBe(true);
-            })
-            .subscribe(() => {
-              done();
-            }, done.fail);
+          subscription = _do.call(skipAndTake(questions, 1), (data: any) => {
+            expect(data.length).toBe(1);
+            expect(data[0].unwrapped).toBe(true);
+          })
+          .subscribe(() => {
+            done();
+          }, done.fail);
         });
       });
     });
 
 
     it('should emit snapshots if preserveSnapshot option is true', (done: any) => {
       refSnapshotted.push('hello snapshot!', () => {
-        subscription = questionsSnapshotted
-          .take(1)
-          .do((data: any) => {
-            expect(data[0].val()).toEqual('hello snapshot!');
-          })
-          .subscribe(() => {
-            done();
-          }, done.fail);
+        subscription = _do.call(skipAndTake(questionsSnapshotted, 1),(data: any) => {
+          expect(data[0].val()).toEqual('hello snapshot!');
+        })
+        .subscribe(() => {
+          done();
+        }, done.fail);
       });
     });
 
@@ -543,9 +537,13 @@ describe('FirebaseListFactory', () => {
 
         expect(unwrappedValueLol.$key).toEqual('key');
         expect(unwrappedValueLol.$value).toEqual('lol');
-        expect(unwrappedValueLol.$exists()).toEqual(true);        
+        expect(unwrappedValueLol.$exists()).toEqual(true);
       });
     });
 
   });
 });
+
+function skipAndTake<T>(obs: Observable<T>, takeCount: number = 1, skipCount: number = 0) {
+  return take.call(skip.call(obs, skipCount), takeCount);
+}

src/database/firebase_list_factory.ts

@@ -5,8 +5,8 @@ import { database } from 'firebase';
 import { observeQuery } from './query_observable';
 import { Query, FirebaseListFactoryOpts } from '../interfaces';
 import * as utils from '../utils';
-import 'rxjs/add/operator/mergeMap';
-import 'rxjs/add/operator/map';
+import { mergeMap } from 'rxjs/operator/mergeMap';
+import { map } from 'rxjs/operator/map';
 
 export function FirebaseListFactory (
   absoluteUrlOrDbRef:string |
@@ -31,7 +31,7 @@ export function FirebaseListFactory (
 
   const queryObs = observeQuery(query);
   return new FirebaseListObservable(ref, subscriber => {
-    let sub = queryObs.map(query => {
+    let sub = mergeMap.call(map.call(queryObs, query => {
       let queried: firebase.database.Query = ref;
       // Only apply the populated keys
       // apply ordering and available querying options
@@ -90,8 +90,7 @@ export function FirebaseListFactory (
       }
 
       return queried;
-    })
-    .mergeMap((queryRef: firebase.database.Reference, ix: number) => {
+    }), (queryRef: firebase.database.Reference, ix: number) => {
       return firebaseListObservable(queryRef, { preserveSnapshot });
     })
     .subscribe(subscriber);

src/database/firebase_list_observable.spec.ts

@@ -1,7 +1,7 @@
 import { FirebaseListObservable } from './index';
 import { Observer } from 'rxjs/Observer';
 import { Observable } from 'rxjs/Observable';
-import 'rxjs/add/operator/map';
+import { map } from 'rxjs/operator/map';
 import { database } from 'firebase';
 import { unwrapMapFn } from '../utils';
 import {
@@ -46,7 +46,7 @@ describe('FirebaseObservable', () => {
   it('should return an instance of FirebaseObservable when calling operators', () => {
     O = new FirebaseListObservable(ref, (observer:Observer<any>) => {
     });
-    expect(O.map(noop) instanceof FirebaseListObservable).toBe(true);
+    expect(map.call(O, noop) instanceof FirebaseListObservable).toBe(true);
   });
 
   describe('push', () => {

src/database/firebase_object_factory.ts

@@ -5,7 +5,6 @@ import * as utils from '../utils';
 import { Query } from '../interfaces';
 import { observeQuery } from './query_observable';
 import { FirebaseObjectFactoryOpts } from '../interfaces';
-import 'rxjs/add/operator/mergeMap';
 
 export function FirebaseObjectFactory (
   absoluteUrlOrDbRef: string | firebase.database.Reference,

src/database/firebase_object_observable.spec.ts

@@ -13,7 +13,7 @@ import {
 import { COMMON_CONFIG, ANON_AUTH_CONFIG } from '../test-config';
 import { FirebaseObjectObservable } from './index';
 import { Observer } from 'rxjs/Observer';
-import 'rxjs/add/operator/map';
+import { map } from 'rxjs/operator/map';
 import { database } from 'firebase';
 
 const rootDatabaseUrl = COMMON_CONFIG.databaseURL;
@@ -44,7 +44,7 @@ describe('FirebaseObjectObservable', () => {
 
   it('should return an instance of FirebaseObservable when calling operators', () => {
     var O = new FirebaseObjectObservable((observer:Observer<any>) => {});
-    expect(O.map(noop) instanceof FirebaseObjectObservable).toBe(true);
+    expect(map.call(O, noop) instanceof FirebaseObjectObservable).toBe(true);
   });
 
   describe('set', () => {

src/database/query_observable.ts

@@ -1,7 +1,8 @@
 import { Observable } from 'rxjs/Observable';
-import 'rxjs/add/observable/of';
+import { of as observableOf } from  'rxjs/observable/of';
 import { Operator } from 'rxjs/Operator';
 import { Observer } from 'rxjs/Observer';
+import { combineLatest } from 'rxjs/operator/combineLatest';
 import { merge } from 'rxjs/operator/merge';
 import { map } from 'rxjs/operator/map';
 import {
@@ -13,18 +14,17 @@ import {
   LimitToSelection,
   Primitive
 } from '../interfaces';
-import 'rxjs/add/operator/merge';
-import 'rxjs/add/operator/combineLatest';
+
 
 export function observeQuery(query: Query): Observable<ScalarQuery> {
   if (!isPresent(query)) {
-    return Observable.of(null);
+    return observableOf(null);
   }
 
   return Observable.create((observer: Observer<ScalarQuery>) => {
 
     let obs = getOrderObservables(query) as Observable<OrderBySelection>;
-    obs.combineLatest(
+    combineLatest.call(obs,
         getStartAtObservable(query),
         getEndAtObservable(query),
         getEqualToObservable(query),
@@ -93,7 +93,7 @@ export function getOrderObservables(query: Query): Observable<OrderBySelection>
   if (observables.length === 1) {
     return observables[0];
   } else if (observables.length > 1) {
-    return observables[0].merge(observables.slice(1));
+    return merge.call(observables[0], observables.slice(1));
   } else {
     return new Observable<OrderBySelection>(subscriber => {
       subscriber.next(null);
@@ -110,7 +110,7 @@ export function getLimitToObservables(query: Query): Observable<LimitToSelection
   if (observables.length === 1) {
     return observables[0];
   } else if (observables.length > 1) {
-    const mergedObs = observables[0].merge(observables.slice(1));
+    const mergedObs = merge.call(observables[0], observables.slice(1));
     return mergedObs;
   } else {
     return new Observable<LimitToSelection>(subscriber => {