RxFirebase is a wrapper for warping Firebase APIs to Reactive Programming in swift

MockFirebase.swift

import Firebase

class MockFirebase: Firebase {

  let mockError = NSError(domain: "This is a test error", code: 0, userInfo: nil)
  let mockSnapshot = FDataSnapshot()

  var makeApiFail = false
  var observerRomoved = false

  override func setValue(value: AnyObject!, withCompletionBlock block: ((NSError!, Firebase!) -> Void)!) {
    block(makeApiFail ? mockError : nil, self)
  }

  override func updateChildValues(values: [NSObject: AnyObject]!, withCompletionBlock block: ((NSError!, Firebase!) -> Void)!) {
    block(makeApiFail ? mockError : nil, self)
  }

  override func removeValueWithCompletionBlock(block: ((NSError!, Firebase!) -> Void)!) {
    block(makeApiFail ? mockError : nil, self)
  }

  override func runTransactionBlock(block: ((FMutableData!) -> FTransactionResult!)!, andCompletionBlock completionBlock: ((NSError!, Bool, FDataSnapshot!) -> Void)!) {
    completionBlock(makeApiFail ? mockError : nil, makeApiFail, mockSnapshot)
  }

  override func observeEventType(eventType: FEventType, withBlock block: ((FDataSnapshot!) -> Void)!, withCancelBlock cancelBlock: ((NSError!) -> Void)!) -> UInt {
    if (makeApiFail) {
      cancelBlock(mockError)
    } else {
      block(mockSnapshot)
    }
    return 0
  }

  override func observeSingleEventOfType(eventType: FEventType, withBlock block: ((FDataSnapshot!) -> Void)!, withCancelBlock cancelBlock: ((NSError!) -> Void)!) {
    if (makeApiFail) {
      cancelBlock(mockError)
    } else {
      block(mockSnapshot)
    }
  }
  
  override func removeAllObservers() {
    observerRomoved = true
  }

}

RxFirebase.swift

import RxSwift
import Firebase

class RxFirebse {

  /**
   Update the values according to the given reference and map. The result of the operation
   will be notified in main thread and cannot be changed.

   - parameter ref:   The `Firebase` reference.
   - parameter value: The target values.

   - returns: `Observable` of `Firebase` where the data updated.
   */
  static func updateChildValues(ref: Firebase, value: [NSObject: AnyObject]) -> Observable<Firebase> {
    return Observable<Firebase>.create { observer in
      let disposable = BooleanDisposable()
      ref.updateChildValues(value) { error, ref in
        if let realError = error where !disposable.disposed {
          observer.onError(realError)
          return
        }
        if !disposable.disposed {
          observer.onNext(ref)
          observer.onCompleted()
        }
      }
      return disposable
    }
  }

  /**
   Set the value according to the given reference and object. The result of the operation
   will be notified in main thread and cannot be changed.

   - parameter ref:   The `Firebase` reference.
   - parameter value: The target value.

   - returns: `Observable` of `Firebase` where the data updated.
   */
  static func setValue(ref: Firebase, value: AnyObject) -> Observable<Firebase> {
    return Observable<Firebase>.create { observer in
      let disposable = BooleanDisposable()
      ref.setValue(value) { error, ref in
        if let realError = error where !disposable.disposed {
          observer.onError(realError)
          return
        }
        if !disposable.disposed {
          observer.onNext(ref)
          observer.onCompleted()
        }
      }
      return disposable
    }
  }

  /**
   Remove a value from the given reference.

   - parameter ref:  The `Firebase` reference.

   - returns: `Observable` of `Firebase` where the data removed.
   */
  static func removeValue(ref: Firebase) -> Observable<Firebase> {
    return Observable<Firebase>.create { observer in
      let disposable = BooleanDisposable()
      ref.removeValueWithCompletionBlock { error, ref in
        if let realError = error where !disposable.disposed {
          observer.onError(realError)
          return
        }
        if !disposable.disposed {
          observer.onNext(ref)
          observer.onCompleted()
        }
      }
      return disposable
    }
  }

  /**
   Run the transaction for the given reference.

   - parameter ref:   The `Firebase` reference.
   - parameter block: The block receives the current data at this location and must return an instance of FTransactionResult.

   - returns: `Observable` of `FDataSnapshot` which indicates what the current value of the data at this location is.
   */
  static func runTransaction(ref: Firebase, block: ((FMutableData!) -> FTransactionResult!)!) -> Observable<FDataSnapshot> {
    return Observable<FDataSnapshot>.create { observer in
      let disposable = BooleanDisposable()
      ref.runTransactionBlock(block) { error, committed, snapshot in
        if let realError = error where !disposable.disposed {
          observer.onError(realError)
          return
        }
        if !disposable.disposed {
          observer.onNext(snapshot)
          observer.onCompleted()
        }
      }
      return disposable
    }
  }

  /**
   Observe on the given reference with given event type. The events are coming from the main thread and cannot be changed.

   - parameter ref:       The `Firebase` reference.
   - paramtetr eventType: The target `FEventType`.

   - returns: `Observable` of `FDataSnapshot` which will omit new data.
   */
  static func observeEvent(ref: FQuery, eventType: FEventType) -> Observable<FDataSnapshot> {
    return Observable<FDataSnapshot>.create { observer in
      ref.observeEventType(eventType, withBlock: { snapshot in
        observer.onNext(snapshot)
        observer.onCompleted()
        }, withCancelBlock: { error in observer.onError(error) })
      return AnonymousDisposable { ref.removeAllObservers() }
    }
  }

  /**
   Observe on the given {@link Query} with single value event. The single value events are coming
   from the main thread and cannot be changed.

   - parameter ref: The `Firebase` reference.

   - returns: `Observable` of `FDataSnapshot` which will omit new data.
   */
  static func observeSingleValueEvent(ref: FQuery) -> Observable<FDataSnapshot> {
    return Observable<FDataSnapshot>.create { observer in
      ref.observeSingleEventOfType(FEventType.Value, withBlock: { snapshot in
        observer.onNext(snapshot)
        observer.onCompleted()
        }, withCancelBlock: { error in observer.onError(error) })
      return AnonymousDisposable { ref.removeAllObservers() }
    }
  }

}

RxFirebaseTest.swift

import XCTest
import RxTests
import Firebase

class RxFirebaseTest: XCTestCase {

  var mockFirebase: MockFirebase!
  var scheduler: TestScheduler!

  override func setUp() {
    super.setUp()
    mockFirebase = MockFirebase()
    scheduler = TestScheduler(initialClock: 0)
  }

  func testSetValue() {
    let res = scheduler.start {
      return RxFirebse.setValue(self.mockFirebase, value: "test").subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
  }

  func testSetValueWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.setValue(self.mockFirebase, value: "test").subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
  }

  func testUpdateChildValues() {
    let res = scheduler.start {
      return RxFirebse.updateChildValues(self.mockFirebase, value: ["test": "a"]).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
  }

  func testUpdateChildValuesWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.updateChildValues(self.mockFirebase, value: ["test": "a"]).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
  }

  func testRemoveValue() {
    let res = scheduler.start {
      return RxFirebse.removeValue(self.mockFirebase).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
  }

  func testRemoveValueWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.removeValue(self.mockFirebase).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
  }

  func testRunTransaction() {
    let res = scheduler.start {
      return RxFirebse.runTransaction(self.mockFirebase, block: nil).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase.mockSnapshot),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
  }

  func testRunTransactionWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.runTransaction(self.mockFirebase, block: nil).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
  }

  func testObserveEvent() {
    let res = scheduler.start {
      return RxFirebse.observeEvent(self.mockFirebase, eventType: FEventType.Value).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase.mockSnapshot),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
    XCTAssertTrue(mockFirebase.observerRomoved)
  }

  func testObserveEventWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.observeEvent(self.mockFirebase, eventType: FEventType.Value).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
    XCTAssertTrue(mockFirebase.observerRomoved)
  }

  func testObserveSingleEvent() {
    let res = scheduler.start {
      return RxFirebse.observeSingleValueEvent(self.mockFirebase).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      next(TestScheduler.Defaults.subscribed + 1, mockFirebase.mockSnapshot),
      completed(TestScheduler.Defaults.subscribed + 1)
    ])
    XCTAssertTrue(mockFirebase.observerRomoved)
  }

  func testObserveSingleWithError() {
    mockFirebase.makeApiFail = true
    let res = scheduler.start {
      return RxFirebse.observeSingleValueEvent(self.mockFirebase).subscribeOn(self.scheduler)
    }
    XCTAssertEqual(res.events, [
      error(TestScheduler.Defaults.subscribed + 1, self.mockFirebase.mockError)
    ])
    XCTAssertTrue(mockFirebase.observerRomoved)
  }

}