Merge pull request #14 from RxSwiftCommunity/mt-1.0.5

Mt 1.0.5

Example/Podfile

@@ -1,14 +1,19 @@
 source 'https://github.com/CocoaPods/Specs.git'
 use_frameworks!
 
-target 'RxRealm_Example', :exclusive => true do
-  platform :ios, '8.3'
+target 'RxRealm_Example' do
+  platform :ios, '8.0'
   pod 'RxRealm', :path => '../'
   pod 'RxSwift'
   pod 'RxCocoa'
   pod 'RealmSwift'
 end
 
-target 'RxRealm_Tests', :exclusive => true do
+target 'RxRealm_Tests' do
   pod 'RxTests'
+
+  pod 'RxRealm', :path => '../'
+  pod 'RxSwift'
+  pod 'RxCocoa'
+  pod 'RealmSwift'
 end

Example/RxRealm_Tests/RxRealmRealmTests.swift

@@ -0,0 +1,95 @@
+//
+//  RxRealmRealmTests.swift
+//  RxRealm
+//
+//  Created by Marin Todorov on 5/22/16.
+//  Copyright © 2016 CocoaPods. All rights reserved.
+//
+
+import XCTest
+
+import RxSwift
+import RealmSwift
+import RxRealm
+import RxTests
+
+class RxRealmRealmTests: XCTestCase {
+    private func realmInMemory(name: String) -> Realm {
+        var conf = Realm.Configuration()
+        conf.inMemoryIdentifier = name
+        return try! Realm(configuration: conf)
+    }
+
+    func testRealmDidChangeNotifications() {
+        let expectation1 = expectationWithDescription("Realm notification")
+
+        let realm = realmInMemory(#function)
+        let bag = DisposeBag()
+
+        let scheduler = TestScheduler(initialClock: 0)
+        let observer = scheduler.createObserver((Realm, Notification))
+
+        let realm$ = realm.asObservable().shareReplay(1)
+        realm$.scan(0, accumulator: {acc, _ in return acc+1})
+            .filter { $0 == 2 }.map {_ in ()}.subscribeNext(expectation1.fulfill).addDisposableTo(bag)
+        realm$
+            .subscribe(observer).addDisposableTo(bag)
+
+        //interact with Realm here
+        delay(0.1) {
+            try! realm.write {
+                realm.add(Message("first"))
+            }
+        }
+        delayInBackground(0.3) {[unowned self] in
+            let realm = self.realmInMemory(#function)
+            try! realm.write {
+                realm.add(Message("second"))
+            }
+        }
+
+        scheduler.start()
+
+        waitForExpectationsWithTimeout(0.5) {error in
+            XCTAssertTrue(error == nil)
+            XCTAssertEqual(observer.events.count, 2)
+            XCTAssertEqual(observer.events[0].value.element!.1, Notification.DidChange)
+            XCTAssertEqual(observer.events[1].value.element!.1, Notification.DidChange)
+        }
+    }
+
+    func testRealmRefreshRequiredNotifications() {
+        let expectation1 = expectationWithDescription("Realm notification")
+
+        let realm = realmInMemory(#function)
+        realm.autorefresh = false
+
+        let bag = DisposeBag()
+
+        let scheduler = TestScheduler(initialClock: 0)
+        let observer = scheduler.createObserver((Realm, Notification))
+
+        let realm$ = realm.asObservable().shareReplay(1)
+        realm$.scan(0, accumulator: {acc, _ in return acc+1})
+            .filter { $0 == 1 }.map {_ in ()}.subscribeNext(expectation1.fulfill).addDisposableTo(bag)
+        realm$
+            .subscribe(observer).addDisposableTo(bag)
+
+        //interact with Realm here from background
+        delayInBackground(0.1) {[unowned self] in
+            let realm = self.realmInMemory(#function)
+            try! realm.write {
+                realm.add(Message("second"))
+            }
+        }
+
+        scheduler.start()
+
+        waitForExpectationsWithTimeout(0.5) {error in
+            XCTAssertTrue(error == nil)
+            XCTAssertEqual(observer.events.count, 1)
+            XCTAssertEqual(observer.events[0].value.element!.1, Notification.RefreshRequired)
+        }
+    }
+
+}

Example/RxRealm_Tests/RxRealmTests.swift

@@ -17,6 +17,12 @@ func delay(delay: Double, closure: () -> Void) {
                    dispatch_get_main_queue(), closure)
 }
 
+func delayInBackground(delay: Double, closure: () -> Void) {
+    dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(delay * Double(NSEC_PER_SEC))),
+                   dispatch_get_global_queue(QOS_CLASS_BACKGROUND, 0), closure)
+}
+
+
 class RxRealm_Tests: XCTestCase {
 
     private func realmInMemory(name: String) -> Realm {
@@ -109,8 +115,9 @@ class RxRealm_Tests: XCTestCase {
         waitForExpectationsWithTimeout(0.5) {error in
             XCTAssertTrue(error == nil)
             XCTAssertEqual(observer.events.count, 2)
-            XCTAssertTrue(observer.events.first!.value.element! == [Message("first(Array)")])
-            XCTAssertTrue(observer.events.last!.value.element! == [Message("first(Array)"), Message("second(Array)")])
+
+            XCTAssertTrue(observer.events[0].value.element!.equalTo([Message("first(Array)")]))
+            XCTAssertTrue(observer.events[1].value.element!.equalTo([Message("first(Array)"), Message("second(Array)")]))
         }
     }
 

Example/RxRealm_Tests/TestModels.swift

@@ -11,7 +11,7 @@ import Foundation
 import RealmSwift
 
 //MARK: Message
-class Message: Object, Equatable {
+class Message: Object {
 
     dynamic var text = ""
 
@@ -24,12 +24,23 @@ class Message: Object, Equatable {
     }
 }
 
+extension Array where Element: Message {
+    func equalTo(to: [Message]) -> Bool {
+        guard count == to.count else {return false}
+        let (result, _) = reduce((true, 0)) {acc, el in
+            guard acc.0 && self[acc.1] == to[acc.1] else {return (false, 0)}
+            return (true, acc.1+1)
+        }
+        return result
+    }
+}
+
 func ==(lhs: Message, rhs: Message) -> Bool {
     return lhs.text == rhs.text
 }
 
 //MARK: User
-class User: Object, Equatable {
+class User: Object {
     dynamic var name = ""
     dynamic var lastMessage: Message?
 

Pod/Classes/RxRealm.swift

@@ -8,7 +8,20 @@ import Foundation
 import RealmSwift
 import RxSwift
 
+/**
+ `NotificationEmitter` is a faux protocol to allow for Realm's collections to be handled in a generic way.
+
+  All collections already include a `addNotificationBlock(_:)` method - making them conform to `NotificationEmitter` just makes it easier to add Rx methods to them.
+
+  The methods of essence in this protocol are `asObservable(...)`, which allow for observing for changes on Realm's collections.
+*/
 public protocol NotificationEmitter {
+
+    /**
+     Returns a `NotificationToken`, which while retained enables change notifications for the current collection.
+
+     - returns: `NotificationToken` - retain this value to keep notifications being emitted for the current collection.
+     */
     func addNotificationBlock(block: (RealmCollectionChange<Self>) -> ()) -> NotificationToken
 }
 
@@ -62,6 +75,7 @@ public extension NotificationEmitter where Self: RealmCollectionType {
             }
 
             return AnonymousDisposable {
+                observer.onCompleted()
                 token.stop()
             }
         }
@@ -103,6 +117,7 @@ public extension NotificationEmitter where Self: RealmCollectionType {
             }
 
             return AnonymousDisposable {
+                observer.onCompleted()
                 token.stop()
             }
         }
@@ -122,4 +137,34 @@ public extension NotificationEmitter where Self: RealmCollectionType {
     public func asObservableArrayChangeset() -> Observable<(Array<Self.Generator.Element>, RealmChangeset?)> {
         return asObservableChangeset().map { (Array($0), $1) }
     }
+}
+
+public extension Realm {
+
+    /**
+     Returns an `Observable<(Realm, Notification)>` that emits each time the Realm emits a notification.
+
+     The Observable you will get emits a tuple made out of:
+
+     * the realm that emitted the event
+     * the notification type: this can be either `.DidChange` which occurs after a refresh or a write transaction ends, 
+     or `.RefreshRequired` which happens when a write transaction occurs from a different thread on the same realm file
+
+     For more information look up: [Notification](https://realm.io/docs/swift/latest/api/Enums/Notification.html)
+
+     - returns: `Observable<(Realm, Notification)>`, which you can subscribe to.
+     */
+    public func asObservable() -> Observable<(Realm, Notification)> {
+        return Observable.create {observer in
+            let token = self.addNotificationBlock {(notification: Notification, realm: Realm) in
+                observer.onNext(realm, notification)
+            }
+
+            return AnonymousDisposable {
+                observer.onCompleted()
+                token.stop()
+            }
+        }
+    }
+
 }

RxRealm.podspec

@@ -1,7 +1,7 @@
 Pod::Spec.new do |s|
 
   s.name             = "RxRealm"
-  s.version          = "0.1.4"
+  s.version          = "0.1.5"
   s.summary          = "An Rx wrapper of Realm's collection type"
 
   s.description      = <<-DESC
@@ -22,7 +22,7 @@ Pod::Spec.new do |s|
   s.source_files = 'Pod/Classes/*.swift'
 
   s.frameworks = 'Foundation'
-  s.dependency 'RealmSwift', '~> 0.100'
+  s.dependency 'RealmSwift', '~> 0.103'
   s.dependency 'RxSwift'
 
 end