fix

README.md

@@ -20,6 +20,179 @@ it, simply add the following line to your Podfile:
 pod 'RxFirebase'
 ```
 
+## Usage
+
+```swift
+        import RxFirebase
+```
+
+### Firestore
+
+Setting data:
+```swift
+        let db = Firestore.firestore()
+
+        // Add a new document in collection "cities"
+        db.collection("cities")
+            .document("SF")
+            .rx
+            .setData([
+                "name": "San Francisco",
+                "state": "CA",
+                "country": "USA",
+                "capital": false,
+                "population": 860000
+                ]).subscribe(onError: { error in
+                    print("Error setting data: \(error)")
+                }).disposed(by: disposeBag)
+
+        // Add a new document with a generated id.
+        db.collection("cities")
+            .rx
+            .addDocument(data: [
+                "name": "San Francisco",
+                "state": "CA",
+                "country": "USA",
+                "capital": false,
+                "population": 860000
+                ]).subscribe(onNext: { ref in
+                    print("Document added with ID: \(ref.documentID)")
+                }, onError: { error in
+                    print("Error adding document: \(error)")
+                }).disposed(by: disposeBag)
+
+        // Set the "capital" field of the city 'SF'
+        db.collection("cities")
+            .document("SF")
+            .rx
+            .updateData([
+                "capital": true
+                ]).subscribe(onNext: {
+                    print("Document successfully updated")
+                }, onError: { error in
+                    print("Error updating document: \(error)")
+                }).disposed(by: disposeBag)
+
+        // https://firebase.google.com/docs/firestore/manage-data/add-data
+```
+
+Get a document
+```swift
+        let db = Firestore.firestore()
+
+        db.collection("cities")
+            .document("SF")
+            .rx
+            .getDocument()
+            .subscribe(onNext: { document in
+                if let document = document {
+                    print("Document data: \(document.data())")
+                } else {
+                    print("Document does not exist")
+                }
+            }, onError: { error in
+                print("Error fetching snapshots: \(error)")
+            }).disposed(by: disposeBag)
+
+        // https://firebase.google.com/docs/firestore/query-data/get-data
+```
+
+Get Realtime Updates
+```swift
+        let db = Firestore.firestore()
+
+        // Document
+        db.collection("cities")
+            .document("SF")
+            .rx
+            .listen()
+            .subscribe(onNext: { document in
+                print("Current data: \(document.data())")
+            }, onError: { error in
+                print("Error fetching snapshots: \(error)")
+            }).disposed(by: disposeBag)
+
+        // Collection
+        db.collection("cities")
+            .rx
+            .listen()
+            .subscribe(onNext: { snapshot in
+                snapshot.documentChanges.forEach { diff in
+                    if (diff.type == .added) {
+                        print("New city: \(diff.document.data())")
+                    }
+                    if (diff.type == .modified) {
+                        print("Modified city: \(diff.document.data())")
+                    }
+                    if (diff.type == .removed) {
+                        print("Removed city: \(diff.document.data())")
+                    }
+                }
+            }, onError: { error in
+                print("Error fetching snapshots: \(error)")
+            }).disposed(by: disposeBag)
+
+        // https://firebase.google.com/docs/firestore/query-data/listen
+```
+
+Batched writes
+```swift
+        let db = Firestore.firestore()
+
+        // Get new write batch
+        let batch = db.batch()
+
+        // Update the population of 'SF'
+        let sfRef = db.collection("cities").document("SF")
+        batch.updateData(["population": 1000000 ], forDocument: sfRef)
+
+        // Commit the batch
+        batch.rx
+            .commit()
+            .subscribe(onNext: {
+                print("Batch write succeeded.")
+            }, onError: { error in
+                print("Error writing batch \(error)")
+            }).disposed(by: disposeBag)
+
+        // https://firebase.google.com/docs/firestore/manage-data/transactions
+```
+
+Transactions
+```swift
+        let db = Firestore.firestore()
+        let sfReference = db.collection("cities").document("SF")
+
+        db.rx.runTransaction { transaction, errorPointer in
+            let sfDocument: DocumentSnapshot
+            do {
+                try sfDocument = transaction.getDocument(sfReference)
+            } catch let fetchError as NSError {
+                errorPointer?.pointee = fetchError
+                return nil
+            }
+
+            guard let oldPopulation = sfDocument.data()?["population"] as? Int else {
+                let error = NSError(
+                    domain: "AppErrorDomain",
+                    code: -1,
+                    userInfo: [
+                        NSLocalizedDescriptionKey: "Unable to retrieve population from snapshot \(sfDocument)"
+                    ]
+                )
+                errorPointer?.pointee = error
+                return nil
+            }
+
+            transaction.updateData(["population": oldPopulation + 1], forDocument: sfReference)
+            return nil
+            }.subscribe(onNext: { _ in
+                print("Transaction successfully committed!")
+            }, onError: { error in
+                print("Transaction failed: \(error)")
+            }).disposed(by: disposeBag)
+``
+
 ## Author
 
 Arnaud Dorgans, arnaud.dorgans@gmail.com

RxFirebase/Classes/Firestore/FIRCollectionReference+Rx.swift

@@ -25,15 +25,14 @@ extension Reactive where Base: CollectionReference {
     public func addDocument(data: [String: Any]) -> Observable<DocumentReference> {
         return Observable<DocumentReference>.create { observer in
             var ref: DocumentReference?
-            let _ref = self.base.addDocument(data: data) { error in
+            ref = self.base.addDocument(data: data) { error in
                 if let error = error {
                     observer.onError(error)
                 } else if let ref = ref {
                     observer.onNext(ref)
                 }
                 observer.onCompleted()
             }
-            ref = _ref
             return Disposables.create()
         }
     }

RxFirebase/Classes/Firestore/FIRDocumentReference+Rx.swift

@@ -113,14 +113,13 @@ extension Reactive where Base: DocumentReference {
      *
      * @param completion a block to execute once the document has been successfully read.
      */
-    public func getDocument() -> Observable<DocumentSnapshot> {
-        return Observable<DocumentSnapshot>.create { observer in
+    public func getDocument() -> Observable<DocumentSnapshot?> {
+        return Observable.create { observer in
             self.base.getDocument { snapshot, error in
                 if let error = error {
                     observer.onError(error)
-                } else if let snapshot = snapshot {
-                    observer.onNext(snapshot)
                 }
+                observer.onNext(snapshot)
                 observer.onCompleted()
             }
             return Disposables.create()
@@ -135,15 +134,14 @@ extension Reactive where Base: DocumentReference {
      *
      * @return A FIRListenerRegistration that can be used to remove this listener.
      */
-    public func addSnapshotListener(options: DocumentListenOptions? = nil) -> Observable<DocumentSnapshot> {
+    public func listen(options: DocumentListenOptions? = nil) -> Observable<DocumentSnapshot> {
         return Observable<DocumentSnapshot>.create { observer in
             let listener = self.base.addSnapshotListener(options: options) { snapshot, error in
                 if let error = error {
                     observer.onError(error)
                 } else if let snapshot = snapshot {
                     observer.onNext(snapshot)
                 }
-                observer.onCompleted()
             }
             return Disposables.create {
                 listener.remove()

RxFirebase/Classes/Firestore/FIRFirestore+Rx.swift

@@ -17,7 +17,7 @@ extension Reactive where Base: Firestore {
      * will return results from cache and any write operations will be queued until the network is
      * restored. The completion block, if provided, will be called once network usage has been disabled.
      */
-    func disableNetwork() -> Observable<Void> {
+    public func disableNetwork() -> Observable<Void> {
         return Observable.create { observer in
             self.base.disableNetwork(completion: { error in
                 guard let error = error else {
@@ -36,7 +36,7 @@ extension Reactive where Base: Firestore {
      * `disableNetworkWithCompletion`. Completion block, if provided, will be called once network uasge
      * has been enabled.
      */
-    func enableNetwork() -> Observable<Void> {
+    public func enableNetwork() -> Observable<Void> {
         return Observable.create { observer in
             self.base.enableNetwork(completion: { error in
                 guard let error = error else {
@@ -78,15 +78,15 @@ extension Reactive where Base: Firestore {
      * @param completion The block to call with the result or error of the transaction. This
      *     block will run even if the client is offline, unless the process is killed.
      */
-    func runTransaction(_ updateBlock: @escaping (Transaction, NSErrorPointer)->Any?) -> Observable<Any> {
+    public func runTransaction(_ updateBlock: @escaping (Transaction, NSErrorPointer)->Any?) -> Observable<Any?> {
         return Observable.create { observer in
             self.base.runTransaction(updateBlock) { value, error in
-                if let error = error {
-                    observer.onError(error)
-                } else if let value = value {
+                guard let error = error else {
                     observer.onNext(value)
+                    observer.onCompleted()
+                    return 
                 }
-                observer.onCompleted()
+                observer.onError(error)
             }
             return Disposables.create()
         }

RxFirebase/Classes/Firestore/FIRQuery+Rx.swift

@@ -39,15 +39,14 @@ extension Reactive where Base: Query {
      *
      * @return A FIRListenerRegistration that can be used to remove this listener.
      */
-    public func addSnapshotListener(options: QueryListenOptions? = nil) -> Observable<QuerySnapshot> {
+    public func listen(options: QueryListenOptions? = nil) -> Observable<QuerySnapshot> {
         return Observable<QuerySnapshot>.create { observer in
             let listener = self.base.addSnapshotListener(options: options) { snapshot, error in
                 if let error = error {
                     observer.onError(error)
                 } else if let snapshot = snapshot {
                     observer.onNext(snapshot)
                 }
-                observer.onCompleted()
             }
             return Disposables.create {
                 listener.remove()