* 0.1.3 first draft

Example/RxRealm_Tests/RxRealmTests.swift

@@ -11,6 +11,11 @@ import RealmSwift
 import RxRealm
 import RxTests
 
+func delay(delay: Double, closure: () -> Void) {
+    dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(delay * Double(NSEC_PER_SEC))),
+                   dispatch_get_main_queue(), closure)
+}
+
 class Message: Object, Equatable {
     dynamic var text = ""
     convenience init(_ text: String) {
@@ -25,25 +30,19 @@ func ==(lhs: Message, rhs: Message) -> Bool {
 
 class RxRealm_Tests: XCTestCase {
 
-    var bag: DisposeBag! = DisposeBag()
-
-    override func setUp() {
-        super.setUp()
-        Realm.Configuration.defaultConfiguration.inMemoryIdentifier = "MemoryRealm"
+    private func realmInMemory(name: String) -> Realm {
+        var conf = Realm.Configuration()
+        conf.inMemoryIdentifier = name
+        return try! Realm(configuration: conf)
     }
 
-    override func tearDown() {
-        bag = nil
-        super.tearDown()
-    }
-
     private func clearRealm(realm: Realm) {
         try! realm.write {
             realm.deleteAll()
         }
     }
 
-    private func addObject(realm: Realm, text: String) {
+    private func addMessage(realm: Realm, text: String) {
         try! realm.write {
             realm.add(Message(text))
         }
@@ -53,8 +52,9 @@ class RxRealm_Tests: XCTestCase {
         let expectation1 = expectationWithDescription("Results<Message> first")
         let expectation2 = expectationWithDescription("Results<Message> second")
 
-        let realm = try! Realm()
+        let realm = realmInMemory(#function)
         clearRealm(realm)
+        let bag = DisposeBag()
 
         let scheduler = TestScheduler(initialClock: 0)
         let observer = scheduler.createObserver(Results<Message>)
@@ -70,27 +70,30 @@ class RxRealm_Tests: XCTestCase {
 
         messages$.subscribe(observer).addDisposableTo(bag)
 
-        addObject(realm, text: "first")
+        addMessage(realm, text: "first(Results)")
 
-        performSelector(#selector(addSecondMessage), withObject: nil, afterDelay: 0.1)
+        delay(0.1) {
+            self.addMessage(realm, text: "second(Results)")
+        }
 
         scheduler.start()
 
         waitForExpectationsWithTimeout(0.5) {error in
             XCTAssertTrue(error == nil)
             XCTAssertEqual(observer.events.count, 2)
             let results = observer.events.last!.value.element!
-            XCTAssertTrue(results.first! == Message("first"))
-            XCTAssertTrue(results.last! == Message("second"))
+            XCTAssertTrue(results.first! == Message("first(Results)"))
+            XCTAssertTrue(results.last! == Message("second(Results)"))
         }
     }
 
     func testEmittedArrayValues() {
         let expectation1 = expectationWithDescription("Array<Message> first")
         let expectation2 = expectationWithDescription("Array<Message> second")
 
-        let realm = try! Realm()
+        let realm = realmInMemory(#function)
         clearRealm(realm)
+        let bag = DisposeBag()
 
         let scheduler = TestScheduler(initialClock: 0)
         let observer = scheduler.createObserver(Array<Message>)
@@ -106,21 +109,142 @@ class RxRealm_Tests: XCTestCase {
 
         messages$.subscribe(observer).addDisposableTo(bag)
 
-        addObject(realm, text: "first")
+        addMessage(realm, text: "first(Array)")
 
-        performSelector(#selector(addSecondMessage), withObject: nil, afterDelay: 0.1)
+        delay(0.1) {
+            self.addMessage(realm, text: "second(Array)")
+        }
 
         scheduler.start()
 
         waitForExpectationsWithTimeout(0.5) {error in
             XCTAssertTrue(error == nil)
             XCTAssertEqual(observer.events.count, 2)
-            XCTAssertTrue(observer.events.first!.value.element! == [Message("first")])
-            XCTAssertTrue(observer.events.last!.value.element! == [Message("first"), Message("second")])
+            XCTAssertTrue(observer.events.first!.value.element! == [Message("first(Array)")])
+            XCTAssertTrue(observer.events.last!.value.element! == [Message("first(Array)"), Message("second(Array)")])
         }
     }
 
-    func addSecondMessage() {
-        addObject(try! Realm(), text: "second")
+    func testEmittedChangeset() {
+        let expectation1 = expectationWithDescription("did emit all changeset values")
+
+        let realm = realmInMemory(#function)
+        clearRealm(realm)
+        let bag = DisposeBag()
+
+        let scheduler = TestScheduler(initialClock: 0)
+        let observer = scheduler.createObserver(String)
+
+        //initial data
+        addMessage(realm, text: "first(Changeset)")
+
+        let messages$ = realm.objects(Message).asObservableChangeset().shareReplay(1)
+        messages$.scan(0) { count, _ in
+            return count+1
+        }
+        .filter {$0 == 3}
+        .subscribeNext {_ in expectation1.fulfill() }
+        .addDisposableTo(bag)
+
+        messages$
+            .map {result, changes in
+                if let changes = changes {
+                    return "count:\(result.count) inserted:\(changes.inserted) deleted:\(changes.deleted) updated:\(changes.updated)"
+                } else {
+                    return "count:\(result.count)"
+                }
+            }
+            .subscribe(observer).addDisposableTo(bag)
+
+        //insert
+        delay(0.25) {
+            self.addMessage(realm, text: "second(Changeset)")
+        }
+        //update
+        delay(0.5) {
+            try! realm.write {
+                realm.delete(realm.objects(Message).filter("text='first(Changeset)'").first!)
+                realm.objects(Message).filter("text='second(Changeset)'").first!.text = "third(Changeset)"
+            }
+        }
+        //coalesced
+        delay(0.7) {
+            self.addMessage(realm, text: "first(Changeset)")
+        }
+        delay(0.7) {
+            try! realm.write {
+                realm.delete(realm.objects(Message).filter("text='first(Changeset)'").first!)
+            }
+        }
+
+        waitForExpectationsWithTimeout(0.75) {error in
+            XCTAssertTrue(error == nil)
+            XCTAssertEqual(observer.events.count, 3)
+            XCTAssertEqual(observer.events[0].value.element!, "count:1")
+            XCTAssertEqual(observer.events[1].value.element!, "count:2 inserted:[1] deleted:[] updated:[]")
+            XCTAssertEqual(observer.events[2].value.element!, "count:1 inserted:[] deleted:[0] updated:[1]")
+        }
+    }
+
+    func testEmittedArrayChangeset() {
+        let expectation1 = expectationWithDescription("did emit all array changeset values")
+
+        let realm = realmInMemory(#function)
+        clearRealm(realm)
+        let bag = DisposeBag()
+
+        let scheduler = TestScheduler(initialClock: 0)
+        let observer = scheduler.createObserver(String)
+
+        //initial data
+        addMessage(realm, text: "first(ArrayChangeset)")
+
+        let messages$ = realm.objects(Message).asObservableArrayChangeset().shareReplay(1)
+        messages$.scan(0) { count, _ in
+            return count+1
+            }
+            .filter {$0 == 3}
+            .subscribeNext {_ in expectation1.fulfill() }
+            .addDisposableTo(bag)
+
+        messages$
+            .map {result, changes in
+                if let changes = changes {
+                    return "count:\(result.count) inserted:\(changes.inserted) deleted:\(changes.deleted) updated:\(changes.updated)"
+                } else {
+                    return "count:\(result.count)"
+                }
+            }
+            .subscribe(observer).addDisposableTo(bag)
+
+        //insert
+        delay(0.25) {
+            self.addMessage(realm, text: "second(ArrayChangeset)")
+        }
+        //update
+        delay(0.5) {
+            try! realm.write {
+                realm.delete(realm.objects(Message).filter("text='first(ArrayChangeset)'").first!)
+                realm.objects(Message).filter("text='second(ArrayChangeset)'").first!.text = "third(ArrayChangeset)"
+            }
+        }
+        //coalesced
+        delay(0.7) {
+            self.addMessage(realm, text: "first(ArrayChangeset)")
+        }
+        delay(0.7) {
+            try! realm.write {
+                realm.delete(realm.objects(Message).filter("text='first(ArrayChangeset)'").first!)
+            }
+        }
+
+        waitForExpectationsWithTimeout(0.75) {error in
+            XCTAssertTrue(error == nil)
+            XCTAssertEqual(observer.events.count, 3)
+            XCTAssertEqual(observer.events[0].value.element!, "count:1")
+            XCTAssertEqual(observer.events[1].value.element!, "count:2 inserted:[1] deleted:[] updated:[]")
+            XCTAssertEqual(observer.events[2].value.element!, "count:1 inserted:[] deleted:[0] updated:[1]")
+        }
     }
+
 }

Pod/Classes/RxRealm.swift

@@ -16,6 +16,12 @@ extension List: NotificationEmitter {}
 extension AnyRealmCollection: NotificationEmitter {}
 extension Results: NotificationEmitter {}
 
+public struct RealmChangeset {
+    public let deleted: [Int]
+    public let inserted: [Int]
+    public let updated: [Int]
+}
+
 public extension NotificationEmitter where Self: RealmCollectionType {
 
     public func asObservable() -> Observable<Self> {
@@ -48,4 +54,29 @@ public extension NotificationEmitter where Self: RealmCollectionType {
     public func asObservableArray() -> Observable<Array<Self.Generator.Element>> {
         return asObservable().map { Array($0) }
     }
+
+    public func asObservableChangeset() -> Observable<(Self, RealmChangeset?)> {
+        return Observable.create {observer in
+            let token = self.addNotificationBlock {changeset in
+
+                switch changeset {
+                case .Initial(let value):
+                    observer.onNext((value, nil))
+                case .Update(let value, let deletes, let inserts, let updates):
+                    observer.onNext((value, RealmChangeset(deleted: deletes, inserted: inserts, updated: updates)))
+                case .Error(let error):
+                    observer.onError(error)
+                    return
+                }
+            }
+
+            return AnonymousDisposable {
+                token.stop()
+            }
+        }
+    }
+
+    public func asObservableArrayChangeset() -> Observable<(Array<Self.Generator.Element>, RealmChangeset?)> {
+        return asObservableChangeset().map { (Array($0), $1) }
+    }
 }

README.md

@@ -8,9 +8,9 @@
 
 This library is a very thin wrapper around the reactive collection types __RealmSwift__ provides: `Results`, `List` and `AnyRealmCollection`. 
 
-The extension adds two  methods to all of the above:
+The extension adds these methods to all of the above:
 
-### asObservable()
+#### asObservable()
 `asObservable()` - emits every time the collection changes:
 
 ```swift
@@ -22,7 +22,7 @@ realm.objects(Lap).asObservable()
   }
 ```
 
-### asObservableArray()
+#### asObservableArray()
 `asObservableArray()` - fetches the a snapshot of a Realm collection and converts it to an array value (for example if you want to use array methods on the collection):
 
 ```swift
@@ -36,29 +36,48 @@ realm.objects(Lap).asObservableArray()
   }
 ```
 
+#### asObservableChangeset()
+`asObservableChangeset()` - emits every time the collection changes and provides the exact indexes that has been deleted, inserted or updated:
 
-## Example app
+```swift
+let realm = try! Realm()
+realm.objects(Lap).asObservableChangeset()
+  .subscribeNext {result, changes in
+    if let changes = changes {
+	  //it's an update
+	  print(result)
+	  print("deleted: \(changes.deleted) inserted: \(changes.inserted) updated: \(changes.updated)")
+	} else {
+	  //it's the initial data
+	  print(result)
+	}
+  }
+```
 
-To run the example project, clone the repo, and run `pod install` from the Example directory first. The app uses RxSwift, RxCocoa using RealmSwift, RxRealm to observe Results from Realm.
+#### asObservableArrayChangeset()
+
+`asObservableArrayChangeset()` combines the result of `asObservableArray()` and `asObservableChangeset()` returning an `Observable<Array<T>, RealmChangeset?>`.
 
-## Requirements
+#### Example app
 
-This library depends on both __RxSwift__ and __RealmSwift__.
+To run the example project, clone the repo, and run `pod install` from the Example directory first. The app uses RxSwift, RxCocoa using RealmSwift, RxRealm to observe Results from Realm.
 
 ## Installation
 
-### CocoaPods
+This library depends on both __RxSwift__ and __RealmSwift__ 0.99+.
+
+#### CocoaPods
 RxRealm is available through [CocoaPods](http://cocoapods.org). To install it, simply add the following line to your Podfile:
 
 ```ruby
 pod "RxRealm"
 ```
 
-### Carthage
+#### Carthage
 
 Feel free to send a PR
 
-### As Source
+#### As Source
 
 You can grab the __RxRealm.swift__ file from this repo and include it in your project.
 
@@ -71,6 +90,7 @@ This library belongs to _RxSwiftCommunity_ and is based on the work of [@fpillet
 * Carthage
 * Add `asObservable()` to the Realm class
 * Test add platforms and add compatibility for the pod
+* Document the source code
 
 ## License