(firestore) Update to rxjs pipeable operators

src/firestore/collection/changes.ts

@@ -1,9 +1,7 @@
 import { fromCollectionRef } from '../observable/fromRef';
 import { Query, DocumentChangeType, DocumentChange } from '@firebase/firestore-types';
-import { Observable } from 'rxjs/Observable';
-import 'rxjs/add/operator/map';
-import 'rxjs/add/operator/filter';
-import 'rxjs/add/operator/scan';
+import { Observable } from 'rxjs';
+import { map, filter, scan } from 'rxjs/operators';
 
 import { DocumentChangeAction, Action } from '../interfaces';
 
@@ -14,9 +12,10 @@ import { DocumentChangeAction, Action } from '../interfaces';
  */
 export function docChanges(query: Query): Observable<DocumentChangeAction[]> {
   return fromCollectionRef(query)
-    .map(action =>
-      action.payload.docChanges
-        .map(change => ({ type: change.type, payload: change })));
+    .pipe(
+      map(action =>
+        action.payload.docChanges
+          .map(change => ({ type: change.type, payload: change }))));
 }
 
 /**
@@ -25,9 +24,10 @@ export function docChanges(query: Query): Observable<DocumentChangeAction[]> {
  */
 export function sortedChanges(query: Query, events: DocumentChangeType[]): Observable<DocumentChangeAction[]> {
   return fromCollectionRef(query)
-    .map(changes => changes.payload.docChanges)
-    .scan((current, changes) => combineChanges(current, changes, events), [])
-    .map(changes => changes.map(c => ({ type: c.type, payload: c })));
+    .pipe(
+      map(changes => changes.payload.docChanges),
+      scan((current, changes) => combineChanges(current, changes, events), []),
+      map(changes => changes.map(c => ({ type: c.type, payload: c }))));
 }
 
 /**

src/firestore/collection/collection.spec.ts

@@ -6,11 +6,8 @@ import { AngularFirestoreCollection } from './collection';
 import { QueryFn } from '../interfaces';
 
 import { FirebaseApp as FBApp } from '@firebase/app-types';
-import { Observable } from 'rxjs/Observable';
-import { BehaviorSubject } from 'rxjs/BehaviorSubject';
-import { of } from 'rxjs/observable/of';
-import { Subscription } from 'rxjs/Subscription';
-import 'rxjs/add/operator/skip';
+import { Observable, BehaviorSubject, Subscription } from 'rxjs';
+import { skip, take, switchMap } from 'rxjs/operators';
 
 import { TestBed, inject } from '@angular/core/testing';
 import { COMMON_CONFIG } from '../test-config';
@@ -80,7 +77,7 @@ describe('AngularFirestoreCollection', () => {
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.valueChanges();
       const sub = changes.subscribe(() => {}).add(
-        changes.take(1).subscribe(data => {
+        changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
           sub.unsubscribe();
         })
@@ -93,8 +90,8 @@ describe('AngularFirestoreCollection', () => {
       const ITEMS = 4;
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.valueChanges();
-      changes.take(1).subscribe(() => {}).add(() => {
-        const sub = changes.take(1).subscribe(data => {
+      changes.pipe(take(1)).subscribe(() => {}).add(() => {
+        const sub = changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
         }).add(() => {
           deleteThemAll(names, ref).then(done).catch(done.fail);
@@ -110,9 +107,9 @@ describe('AngularFirestoreCollection', () => {
       const randomCollectionName = randomName(afs.firestore);
       const ref = afs.firestore.collection(`${randomCollectionName}`);
       let names = await createRandomStocks(afs.firestore, ref, ITEMS);
-      const sub = pricefilter$.switchMap(price => {
+      const sub = pricefilter$.pipe(switchMap(price => {
         return afs.collection(randomCollectionName, ref => price ? ref.where('price', '==', price) : ref).valueChanges()
-      }).subscribe(data => {
+      })).subscribe(data => {
         count = count + 1;
         // the first time should all be 'added'
         if(count === 1) {
@@ -161,7 +158,7 @@ describe('AngularFirestoreCollection', () => {
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.snapshotChanges();
       const sub = changes.subscribe(() => {}).add(
-        changes.take(1).subscribe(data => {
+        changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
           sub.unsubscribe();
         })
@@ -174,8 +171,8 @@ describe('AngularFirestoreCollection', () => {
       const ITEMS = 4;
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.snapshotChanges();
-      changes.take(1).subscribe(() => {}).add(() => {
-        const sub = changes.take(1).subscribe(data => {
+      changes.pipe(take(1)).subscribe(() => {}).add(() => {
+        const sub = changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
         }).add(() => {
           deleteThemAll(names, ref).then(done).catch(done.fail);
@@ -214,7 +211,7 @@ describe('AngularFirestoreCollection', () => {
       const ITEMS = 10;
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
 
-      const sub = stocks.snapshotChanges(['modified']).skip(1).subscribe(data => {
+      const sub = stocks.snapshotChanges(['modified']).pipe(skip(1)).subscribe(data => {
         sub.unsubscribe();
         const change = data.filter(x => x.payload.doc.id === names[0])[0];
         expect(data.length).toEqual(1);
@@ -231,7 +228,7 @@ describe('AngularFirestoreCollection', () => {
       let { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const nextId = ref.doc('a').id;
 
-      const sub = stocks.snapshotChanges(['added']).skip(1).subscribe(data => {
+      const sub = stocks.snapshotChanges(['added']).pipe(skip(1)).subscribe(data => {
         sub.unsubscribe();
         const change = data.filter(x => x.payload.doc.id === nextId)[0];
         expect(data.length).toEqual(ITEMS + 1);
@@ -252,7 +249,7 @@ describe('AngularFirestoreCollection', () => {
       const nextId = ref.doc('a').id;
       let count = 0;
 
-      const sub = stocks.snapshotChanges(['added', 'modified']).skip(1).take(2).subscribe(data => {
+      const sub = stocks.snapshotChanges(['added', 'modified']).pipe(skip(1),take(2)).subscribe(data => {
         count += 1;
         if (count == 1) {
           const change = data.filter(x => x.payload.doc.id === nextId)[0];
@@ -279,7 +276,7 @@ describe('AngularFirestoreCollection', () => {
       const ITEMS = 10;
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
 
-      const sub = stocks.snapshotChanges(['added', 'removed']).skip(1).subscribe(data => {
+      const sub = stocks.snapshotChanges(['added', 'removed']).pipe(skip(1)).subscribe(data => {
         sub.unsubscribe();
         const change = data.filter(x => x.payload.doc.id === names[0]);
         expect(data.length).toEqual(ITEMS - 1);
@@ -339,7 +336,7 @@ describe('AngularFirestoreCollection', () => {
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.stateChanges();
       const sub = changes.subscribe(() => {}).add(
-        changes.take(1).subscribe(data => {
+        changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
           sub.unsubscribe();
         })
@@ -352,8 +349,8 @@ describe('AngularFirestoreCollection', () => {
       const ITEMS = 4;
       const { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
       const changes = stocks.stateChanges();
-      changes.take(1).subscribe(() => {}).add(() => {
-        const sub = changes.take(1).subscribe(data => {
+      changes.pipe(take(1)).subscribe(() => {}).add(() => {
+        const sub = changes.pipe(take(1)).subscribe(data => {
           expect(data.length).toEqual(ITEMS);
         }).add(() => {
           deleteThemAll(names, ref).then(done).catch(done.fail);
@@ -383,7 +380,7 @@ describe('AngularFirestoreCollection', () => {
       let count = 0;
       let { randomCollectionName, ref, stocks, names } = await collectionHarness(afs, ITEMS);
 
-      const sub = stocks.stateChanges(['added']).skip(1).subscribe(data => {
+      const sub = stocks.stateChanges(['added']).pipe(skip(1)).subscribe(data => {
         sub.unsubscribe();
         expect(data.length).toEqual(1);
         expect(data[0].payload.doc.data().price).toEqual(2);

src/firestore/collection/collection.ts

@@ -1,9 +1,7 @@
 import { DocumentChangeType, CollectionReference, Query, DocumentReference } from '@firebase/firestore-types';
-import { Observable } from 'rxjs/Observable';
-import { Subscriber } from 'rxjs/Subscriber';
+import { Observable, Subscriber } from 'rxjs';
 import { fromCollectionRef } from '../observable/fromRef';
-import 'rxjs/add/operator/map';
-import 'rxjs/add/operator/filter';
+import { map, filter } from 'rxjs/operators';
 
 import { Injectable } from '@angular/core';
 
@@ -12,8 +10,6 @@ import { docChanges, sortedChanges } from './changes';
 import { AngularFirestoreDocument } from '../document/document';
 import { AngularFirestore } from '../firestore';
 
-import 'rxjs/add/observable/of';
-
 export function validateEventsArray(events?: DocumentChangeType[]) {
   if(!events || events!.length === 0) {
     events = ['added', 'removed', 'modified'];
@@ -79,8 +75,10 @@ export class AngularFirestoreCollection<T> {
           docChanges(this.query)
         )
       )
-      .map(actions => actions.filter(change => events.indexOf(change.type) > -1))
-      .filter(changes =>  changes.length > 0);
+      .pipe(
+        map(actions => actions.filter(change => events.indexOf(change.type) > -1)),
+        filter(changes =>  changes.length > 0)
+      );
   }
 
   /**
@@ -111,7 +109,9 @@ export class AngularFirestoreCollection<T> {
     const fromCollectionRef$ = fromCollectionRef(this.query);
     const scheduled$ = this.afs.scheduler.runOutsideAngular(fromCollectionRef$);
     return this.afs.scheduler.keepUnstableUntilFirst(scheduled$)
-      .map(actions => actions.payload.docs.map(a => a.data()) as T[]);
+      .pipe(
+        map(actions => actions.payload.docs.map(a => a.data()) as T[])
+      );
   }
 
   /**

src/firestore/document/document.spec.ts

@@ -4,8 +4,8 @@ import { AngularFirestoreModule } from '../firestore.module';
 import { AngularFirestoreDocument } from '../document/document';
 
 import { FirebaseApp as FBApp } from '@firebase/app-types';
-import { Observable } from 'rxjs/Observable';
-import { Subscription } from 'rxjs/Subscription';
+import { Observable, Subscription } from 'rxjs';
+import { take } from 'rxjs/operators';
 
 import { TestBed, inject } from '@angular/core/testing';
 import { COMMON_CONFIG } from '../test-config';
@@ -57,7 +57,7 @@ describe('AngularFirestoreDocument', () => {
     const stock = new AngularFirestoreDocument<Stock>(ref, afs);
     await stock.set(FAKE_STOCK_DATA);
     const obs$ = stock.valueChanges();
-    obs$.take(1).subscribe(async (data: Stock) => {
+    obs$.pipe(take(1)).subscribe(async (data: Stock) => {
       expect(JSON.stringify(data)).toBe(JSON.stringify(FAKE_STOCK_DATA));
       stock.delete().then(done).catch(done.fail);
     });

src/firestore/document/document.ts

@@ -1,9 +1,8 @@
 import { DocumentReference, SetOptions, DocumentSnapshot } from '@firebase/firestore-types';
-import { Observable } from 'rxjs/Observable';
-import { Subscriber } from 'rxjs/Subscriber';
+import { Observable, Subscriber } from 'rxjs';
 import { QueryFn, AssociatedReference, Action } from '../interfaces';
 import { fromDocRef } from '../observable/fromRef';
-import 'rxjs/add/operator/map';
+import { map } from 'rxjs/operators';
 
 import { Injectable } from '@angular/core';
 
@@ -90,8 +89,10 @@ export class AngularFirestoreDocument<T> {
    * Listen to unwrapped snapshot updates from the document.
    */
   valueChanges(): Observable<T|null> {
-    return this.snapshotChanges().map(action => {
-      return action.payload.exists ? action.payload.data() as T : null;
-    });
+    return this.snapshotChanges().pipe(
+      map(action => {
+        return action.payload.exists ? action.payload.data() as T : null;
+      })
+    );
   }
 }

src/firestore/firestore.spec.ts

@@ -4,8 +4,7 @@ import { AngularFirestoreModule } from './firestore.module';
 import { AngularFirestoreDocument } from './document/document';
 import { AngularFirestoreCollection } from './collection/collection';
 
-import { Observable } from 'rxjs/Observable';
-import { Subscription } from 'rxjs/Subscription';
+import { Observable, Subscription } from 'rxjs';
 
 import { TestBed, inject } from '@angular/core/testing';
 import { COMMON_CONFIG } from './test-config';

src/firestore/firestore.ts

@@ -1,13 +1,10 @@
 import { InjectionToken, NgZone, PLATFORM_ID } from '@angular/core';
 import { FirebaseFirestore, CollectionReference, DocumentReference } from '@firebase/firestore-types';
 
-import { Observable } from 'rxjs/Observable';
-import { Subscriber } from 'rxjs/Subscriber';
-import { from } from 'rxjs/observable/from';
+import { Observable, Subscriber } from 'rxjs';
+import { map, catchError } from 'rxjs/operators';
 import { of } from 'rxjs/observable/of';
-import 'rxjs/add/operator/map';
-import 'rxjs/add/operator/catch';
-
+import { from } from 'rxjs/observable/from';
 import { FirebaseOptions } from '@firebase/app-types';
 import { Injectable, Inject, Optional } from '@angular/core';
 
@@ -125,7 +122,9 @@ export class AngularFirestore {
         shouldEnablePersistence ? from(this.firestore.enablePersistence().then(() => true, () => false))
                                 : of(false)
       )
-      .catch(() => of(false)); // https://github.com/firebase/firebase-js-sdk/issues/608
+      .pipe(
+        catchError(() => of(false))
+      ); // https://github.com/firebase/firebase-js-sdk/issues/608
   }
 
   /**

src/firestore/interfaces.ts

@@ -1,4 +1,4 @@
-import { Subscriber } from 'rxjs/Subscriber';
+import { Subscriber } from 'rxjs';
 import { DocumentChangeType, DocumentChange, CollectionReference, Query } from '@firebase/firestore-types';
 
 export interface DocumentChangeAction {

src/firestore/observable/fromRef.ts

@@ -1,10 +1,7 @@
 import { DocumentReference, Query, QuerySnapshot, DocumentSnapshot } from '@firebase/firestore-types';
-import { Observable } from 'rxjs/Observable';
-import { Subscriber } from 'rxjs/Subscriber';
+import { Observable, Subscriber } from 'rxjs';
 import { Action, Reference } from '../interfaces';
-
-import 'rxjs/add/operator/map';
-import 'rxjs/add/operator/share';
+import { map, share } from 'rxjs/operators';
 
 function _fromRef<T, R>(ref: Reference<T>): Observable<R> {
   return new Observable(subscriber => {
@@ -14,14 +11,16 @@ function _fromRef<T, R>(ref: Reference<T>): Observable<R> {
 }
 
 export function fromRef<R>(ref: DocumentReference | Query) {
-  return _fromRef<typeof ref, R>(ref).share();
+  return _fromRef<typeof ref, R>(ref).pipe(share());
 }
 
 export function fromDocRef(ref: DocumentReference): Observable<Action<DocumentSnapshot>>{
   return fromRef<DocumentSnapshot>(ref)
-    .map(payload => ({ payload, type: 'value' }));
+    .pipe(
+      map(payload => ({ payload, type: 'value' }))
+    );
 }
 
 export function fromCollectionRef(ref: Query): Observable<Action<QuerySnapshot>> {
-  return fromRef<QuerySnapshot>(ref).map(payload => ({ payload, type: 'query' }))
+  return fromRef<QuerySnapshot>(ref).pipe(map(payload => ({ payload, type: 'query' })));
 }