Implement RxGRDB on top GRDB.ValueObservation

CHANGELOG.md

@@ -1,6 +1,22 @@
 Release Notes
 =============
 
+## Next Version
+
+- [#46](https://github.com/RxSwiftCommunity/RxGRDB/pull/46): Implement RxGRDB on top GRDB.ValueObservation
+
+
+### Breaking Changes
+
+- The `DatabaseWriter.rx.fetch` method has been removed. Instead, use [`ValueObservation.rx.fetch`](README.md#valueobservationrxfetchinstartimmediatelyscheduler).
+- The `distinctUntilChanged` parameter is no longer available when one creates an RxGRDB observable. Filtering of consecutive identical database values is now the default behavior.
+
+
+### New
+
+- One can now create a [values observable](README.md#values-observables) from a [DatabaseReader](https://groue.github.io/GRDB.swift/docs/3.5/Protocols/DatabaseReader.html).
+
+
 ## 0.12.1
 
 Released October 25, 2018 • [diff](https://github.com/RxSwiftCommunity/RxGRDB/compare/v0.12.0...v0.12.1)

Documentation/RxGRDBDemo/RxGRDBDemo.xcodeproj/project.pbxproj

@@ -211,22 +211,24 @@
 			files = (
 			);
 			inputPaths = (
-				"${SRCROOT}/../../Pods/Target Support Files/Pods-RxGRDBDemo/Pods-RxGRDBDemo-frameworks.sh",
+				"${PODS_ROOT}/Target Support Files/Pods-RxGRDBDemo/Pods-RxGRDBDemo-frameworks.sh",
 				"${BUILT_PRODUCTS_DIR}/GRDB.swift-iOS/GRDB.framework",
+				"${BUILT_PRODUCTS_DIR}/RxAtomic-iOS/RxAtomic.framework",
 				"${BUILT_PRODUCTS_DIR}/RxSwift-iOS/RxSwift.framework",
 				"${BUILT_PRODUCTS_DIR}/Differ/Differ.framework",
 				"${BUILT_PRODUCTS_DIR}/RxGRDB/RxGRDB.framework",
 			);
 			name = "[CP] Embed Pods Frameworks";
 			outputPaths = (
 				"${TARGET_BUILD_DIR}/${FRAMEWORKS_FOLDER_PATH}/GRDB.framework",
+				"${TARGET_BUILD_DIR}/${FRAMEWORKS_FOLDER_PATH}/RxAtomic.framework",
 				"${TARGET_BUILD_DIR}/${FRAMEWORKS_FOLDER_PATH}/RxSwift.framework",
 				"${TARGET_BUILD_DIR}/${FRAMEWORKS_FOLDER_PATH}/Differ.framework",
 				"${TARGET_BUILD_DIR}/${FRAMEWORKS_FOLDER_PATH}/RxGRDB.framework",
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 			shellPath = /bin/sh;
-			shellScript = "\"${SRCROOT}/../../Pods/Target Support Files/Pods-RxGRDBDemo/Pods-RxGRDBDemo-frameworks.sh\"\n";
+			shellScript = "\"${PODS_ROOT}/Target Support Files/Pods-RxGRDBDemo/Pods-RxGRDBDemo-frameworks.sh\"\n";
 			showEnvVarsInLog = 0;
 		};
 /* End PBXShellScriptBuildPhase section */

Podfile

@@ -3,7 +3,7 @@ workspace 'RxGRDB.xcworkspace'
 
 def common
     pod 'RxSwift', '~> 4.0'
-    pod 'GRDB.swift', '~> 3.3'
+    pod 'GRDB.swift', '~> 3.5'
 end
 
 target 'RxGRDBiOS' do

Podfile.lock

@@ -1,24 +1,27 @@
 PODS:
   - Differ (1.3.0)
-  - GRDB.swift (3.4.0)
+  - GRDB.swift (3.5.0)
+  - RxAtomic (4.4.0)
   - RxGRDB (0.12.1):
     - RxGRDB/default (= 0.12.1)
     - RxSwift (~> 4.0)
   - RxGRDB/default (0.12.1):
-    - GRDB.swift (~> 3.3)
+    - GRDB.swift (~> 3.5)
     - RxSwift (~> 4.0)
-  - RxSwift (4.3.1)
+  - RxSwift (4.4.0):
+    - RxAtomic (~> 4.4)
 
 DEPENDENCIES:
   - Differ (~> 1.0)
-  - GRDB.swift (~> 3.3)
+  - GRDB.swift (~> 3.5)
   - RxGRDB (from `.`)
   - RxSwift (~> 4.0)
 
 SPEC REPOS:
   https://github.com/cocoapods/specs.git:
     - Differ
     - GRDB.swift
+    - RxAtomic
     - RxSwift
 
 EXTERNAL SOURCES:
@@ -27,10 +30,11 @@ EXTERNAL SOURCES:
 
 SPEC CHECKSUMS:
   Differ: 9fcee7602542d7c6484ce05a0faec1c11258ed63
-  GRDB.swift: 0100cdc153f3a42348652e33e0b5697c53eee792
-  RxGRDB: 2c0e8e9f1e052bb225712c90c1a599fbdbcd7239
-  RxSwift: fe0fd770a43acdb7d0a53da411c9b892e69bb6e4
+  GRDB.swift: 225cf9c1354753e92e20c2a1fa2912b924c9da6e
+  RxAtomic: eacf60db868c96bfd63320e28619fe29c179656f
+  RxGRDB: 7c27e9cdc81185696b8ab7f29a2b6a8dc5c1b23d
+  RxSwift: 5976ecd04fc2fefd648827c23de5e11157faa973
 
-PODFILE CHECKSUM: 819b3e856a17f10d96b77a1f7e02caa41d582160
+PODFILE CHECKSUM: 1590d9a0f5d19408cda2cbca1ece0402fec7a260
 
-COCOAPODS: 1.5.3
+COCOAPODS: 1.6.0.beta.2

RxGRDB.podspec

@@ -18,12 +18,12 @@ Pod::Spec.new do |s|
 
   s.subspec 'default' do |ss|
     ss.source_files = 'RxGRDB/**/*.{h,swift}'
-    ss.dependency "GRDB.swift", "~> 3.3"
+    ss.dependency "GRDB.swift", "~> 3.5"
   end
 
   s.subspec 'GRDBCipher' do |ss|
     ss.source_files = 'RxGRDB/**/*.{h,swift}'
-    ss.dependency "GRDBCipher", "~> 3.3"
+    ss.dependency "GRDBCipher", "~> 3.5"
     ss.xcconfig = {
       'OTHER_SWIFT_FLAGS' => '$(inherited) -DSQLITE_HAS_CODEC -DUSING_SQLCIPHER',
       'OTHER_CFLAGS' => '$(inherited) -DSQLITE_HAS_CODEC -DUSING_SQLCIPHER',

RxGRDB/DatabaseReader+Rx.swift

@@ -0,0 +1,79 @@
+//#if USING_SQLCIPHER
+//    import GRDBCipher
+//#else
+//    import GRDB
+//#endif
+//import RxSwift
+//
+//extension Reactive where Base: DatabaseReader {
+//    /// Returns an Observable that emits values after each committed
+//    /// database transaction that has modified the tables, columns,
+//    /// and rows defined by some *regions*.
+//    ///
+//    ///     // When the players table is changed, fetch the ten best ones,
+//    ///     // as well as the total number of players:
+//    ///     dbQueue.rx
+//    ///         .fetch(from: [Player.all()]) { (db: Database) -> ([Player], Int) in
+//    ///             let players = try Player.order(scoreColumn.desc).limit(10).fetchAll(db)
+//    ///             let count = try Player.fetchCount(db)
+//    ///             return (players, count)
+//    ///         }
+//    ///         .subscribe(onNext: { (players, count) in
+//    ///             print("Best players out of \(count): \(players)")
+//    ///         })
+//    ///
+//    /// The `values` closure argument is called after each impactful
+//    /// transaction, and returns the values emitted by the observable. It runs
+//    /// in a protected database queue.
+//    ///
+//    /// By default, all values are emitted on the main dispatch queue. If you
+//    /// give a *scheduler*, values are emitted on that scheduler.
+//    ///
+//    /// If you set *startImmediately* to true (the default value), the first
+//    /// element is emitted right upon subscription. It is *synchronously*
+//    /// emitted if and only if the observable is subscribed on the main queue,
+//    /// and is given a nil *scheduler* argument:
+//    ///
+//    ///     // on the main queue
+//    ///     dbQueue.rx
+//    ///         .fetch(from: [request, ...]) { db in ... }
+//    ///         .subscribe(onNext: { values in
+//    ///             // on the main queue
+//    ///             print("Values have changed")
+//    ///         })
+//    ///     // <- here "Values have changed" has been printed
+//    ///
+//    ///     // on any queue
+//    ///     request.rx
+//    ///         .fetch(from: [request, ...], scheduler: MainScheduler.instance) { db in ... }
+//    ///         .subscribe(onNext: { values in
+//    ///             // on the main queue
+//    ///             print("Values have changed")
+//    ///         })
+//    ///     // <- here "Values have changed" may not be printed yet
+//    ///
+//    /// - parameter regions: The observed regions.
+//    /// - parameter startImmediately: When true (the default), the first
+//    ///   element is emitted right upon subscription.
+//    /// - parameter scheduler: The eventual scheduler on which elements
+//    ///   are emitted.
+//    /// - parameter values: A closure that returns the values emitted by
+//    ///   the observable
+//    public func fetch<T>(
+//        from regions: [DatabaseRegionConvertible],
+//        startImmediately: Bool = true,
+//        scheduler: ImmediateSchedulerType? = nil,
+//        values: @escaping (Database) throws -> T)
+//        -> Observable<T>
+//    {
+//        let region = AnyDatabaseRegionConvertible { db in
+//            try regions.reduce(into: DatabaseRegion()) {
+//                try $0.formUnion($1.databaseRegion(db))
+//            }
+//        }
+//        return ValueObservation.tracking(region, fetch: values).rx.fetch(
+//            in: base,
+//            startImmediately: startImmediately,
+//            scheduler: scheduler)
+//    }
+//}

RxGRDB/DatabaseRegionConvertible+Rx.swift

@@ -0,0 +1,73 @@
+#if USING_SQLCIPHER
+    import GRDBCipher
+#else
+    import GRDB
+#endif
+import RxSwift
+
+extension Reactive where Base: DatabaseRegionConvertible {
+    /// Returns an Observable that emits a database connection after each
+    /// committed database transaction that has modified the tables and columns
+    /// fetched by the request.
+    ///
+    /// All elements are emitted in a protected database dispatch queue,
+    /// serialized with all database updates. If you set *startImmediately* to
+    /// true (the default value), the first element is emitted synchronously
+    /// upon subscription. See [GRDB Concurrency Guide](https://github.com/groue/GRDB.swift/blob/master/README.md#concurrency)
+    /// for more information.
+    ///
+    ///     let dbQueue = DatabaseQueue()
+    ///     try dbQueue.inDatabase { db in
+    ///         try db.create(table: "persons") { t in
+    ///             t.column("id", .integer).primaryKey()
+    ///             t.column("name", .text)
+    ///         }
+    ///     }
+    ///
+    ///     let request = SQLRequest("SELECT * FROM persons")
+    ///     request.rx
+    ///         .changes(in: dbQueue)
+    ///         .subscribe(onNext: { db in
+    ///             let count = try! request.fetchCount(db)
+    ///             print("Number of persons: \(count)")
+    ///         })
+    ///     // Prints "Number of persons: 0"
+    ///
+    ///     try dbQueue.inDatabase { db in
+    ///         try db.execute("INSERT INTO persons (name) VALUES (?)", arguments: ["Arthur"])
+    ///         // Prints "Number of persons: 1"
+    ///         try db.execute("INSERT INTO persons (name) VALUES (?)", arguments: ["Barbara"])
+    ///         // Prints "Number of persons: 2"
+    ///     }
+    ///
+    ///     try dbQueue.inTransaction { db in
+    ///         try db.execute("INSERT INTO persons (name) VALUES (?)", arguments: ["Craig"])
+    ///         try db.execute("INSERT INTO persons (name) VALUES (?)", arguments: ["David"])
+    ///         return .commit
+    ///     }
+    ///     // Prints "Number of persons: 4"
+    ///
+    /// - parameter writer: A DatabaseWriter (DatabaseQueue or DatabasePool).
+    /// - parameter startImmediately: When true (the default), the first
+    ///   element is emitted synchronously, on subscription.
+    public func changes(
+        in writer: DatabaseWriter,
+        startImmediately: Bool = true)
+        -> Observable<Database>
+    {
+        return AnyDatabaseWriter(writer).rx.changes(
+            in: [base],
+            startImmediately: startImmediately)
+    }
+}
+
+/// TODO: remove
+extension Array where Element == DatabaseRegion {
+    func union() -> DatabaseRegion {
+        if let initial = first {
+            return suffix(from: 1).reduce(into: initial) { $0.formUnion($1) }
+        } else {
+            return DatabaseRegion()
+        }
+    }
+}