最新版本: 2022 年 9 月 9 日 • 版本 3.0.0 • 发行说明

要求: iOS 11.0+ / macOS 10.13+ / tvOS 11.0+ / watchOS 4.0+ • Swift 5.7+ / Xcode 14+

要连接到数据库，请参考支持 RxGRDB 的数据库库 GRDB。

// Single<[Player]>
let players = dbQueue.rx.read { db in
    try Player.fetchAll(db)
}

players.subscribe(
    onSuccess: { (players: [Player]) in
        print("Players: \(players)")
    },
    onError: { error in ... })

// Single<Void>
let write = dbQueue.rx.write { db in 
    try Player(...).insert(db)
}

write.subscribe(
    onSuccess: { _ in
        print("Updates completed")
    },
    onError: { error in ... })

// Single<Int>
let newPlayerCount = dbQueue.rx.write { db -> Int in
    try Player(...).insert(db)
    return try Player.fetchCount(db)
}

newPlayerCount.subscribe(
    onSuccess: { (playerCount: Int) in
        print("New players count: \(playerCount)")
    },
    onError: { error in ... })

// Observable<[Player]>
let observable = ValueObservation
    .tracking { db in try Player.fetchAll(db) }
    .rx.observe(in: dbQueue)

observable.subscribe(
    onNext: { (players: [Player]) in
        print("Fresh players: \(players)")
    },
    onError: { error in ... })

// Observable<Int?>
let observable = ValueObservation
    .tracking { db in try Int.fetchOne(db, sql: "SELECT MAX(score) FROM player") }
    .rx.observe(in: dbQueue)

observable.subscribe(
    onNext: { (maxScore: Int?) in
        print("Fresh maximum score: \(maxScore)")
    },
    onError: { error in ... })

// Observable<Database>
let observable = DatabaseRegionObservation
    .tracking(Player.all())
    .rx.changes(in: dbQueue)

observable.subscribe(
    onNext: { (db: Database) in
        print("Exclusive write access to the database after players have been impacted")
    },
    onError: { error in ... })

// Observable<Database>
let observable = DatabaseRegionObservation
    .tracking(SQLRequest<Int>(sql: "SELECT MAX(score) FROM player"))
    .rx.changes(in: dbQueue)

observable.subscribe(
    onNext: { (db: Database) in
        print("Exclusive write access to the database after maximum score has been impacted")
    },
    onError: { error in ... })

• 安装
• 演示应用程序
• 异步数据库访问
• 数据库观察

要将 RxGRDB 与 Swift Package Manager 一起使用，请将依赖项添加到您的 Package.swift 文件中

let package = Package(
    dependencies: [
        .package(url: "https://github.com/RxSwiftCommunity/RxGRDB.git", ...)
    ]
)

要将 RxGRDB 与 CocoaPods 一起使用，请在您的 Podfile 中指定

# Pick only one
pod 'RxGRDB'
pod 'RxGRDB/SQLCipher'

RxGRDB 提供了执行异步数据库访问的可观察对象。

• rx.read(observeOn:value:)
• rx.write(observeOn:updates:)
• rx.write(observeOn:updates:thenRead:)

此方法返回一个 Single，它在异步获取数据库值后完成。

// Single<[Player]>
let players = dbQueue.rx.read { db in
    try Player.fetchAll(db)
}

任何修改数据库的尝试都会导致订阅以错误完成。

当您使用数据库队列或数据库快照时，读取操作必须等待此队列或快照执行的任何可能的并发数据库访问完成。

当您使用数据库连接池时，除非已达到最大并发读取数，否则读取通常是非阻塞的。在这种情况下，读取操作必须等待另一个读取操作完成。最大数量可以配置。

此可观察对象可以从任何线程订阅。每次订阅都会启动新的数据库访问。

除非您为 observeOn 参数提供特定的调度器，否则获取的值将在主队列上发布。

此方法返回一个 Single，它在数据库更新在数据库事务中成功执行后完成。

// Single<Void>
let write = dbQueue.rx.write { db in
    try Player(...).insert(db)
}

// Single<Int>
let newPlayerCount = dbQueue.rx.write { db -> Int in
    try Player(...).insert(db)
    return try Player.fetchCount(db)
}

此可观察对象可以从任何线程订阅。每次订阅都会启动新的数据库访问。

它在主队列上完成，除非您为 observeOn 参数提供特定的 调度器。

您可以忽略它的值，并使用 asCompletable 运算符将其转换为 Completable

// Completable
let write = dbQueue.rx
    .write { db in try Player(...).insert(db) }
    .asCompletable()

当您使用 数据库连接池，并且您的应用程序执行了一些数据库更新，然后执行了一些缓慢的获取操作，您可以通过 rx.write(observeOn:updates:thenRead:) 来优化调度。请参见下文。

此方法返回一个 Single，它在数据库更新在数据库事务中成功执行后完成，并且随后获取了值

// Single<Int>
let newPlayerCount = dbQueue.rx.write(
    updates: { db in try Player(...).insert(db) }
    thenRead: { db, _ in try Player.fetchCount(db) })
}

它发布的值与 rx.write(observeOn:updates:) 完全相同

// Single<Int>
let newPlayerCount = dbQueue.rx.write { db -> Int in
    try Player(...).insert(db)
    return try Player.fetchCount(db)
}

区别在于最后一次获取是在 thenRead 函数中执行的。此函数接受两个参数：只读数据库连接和 updates 函数的结果。这允许您将信息从一个函数传递到另一个函数（在上面的示例代码中被忽略）。

当您使用 数据库连接池时，此方法应用了调度优化：thenRead 函数看到的是 updates 函数留下的数据库状态，但不会阻止任何并发写入。这可以减少数据库写入争用。有关更多信息，请参见 高级 DatabasePool。

当您使用 数据库队列时，结果保证是相同的，但不会应用任何调度优化。

此可观察对象可以从任何线程订阅。每次订阅都会启动新的数据库访问。

它在主队列上完成，除非您为 observeOn 参数提供特定的 调度器。

数据库观察可观察对象基于 GRDB 的 ValueObservation 和 DatabaseRegionObservation。有关更多信息，请参考它们的文档。如果您的应用程序需要 RxGRDB 中未内置的更改通知，请查看通用的 数据库更改观察章节。

• ValueObservation.rx.observe(in:scheduling:)
• DatabaseRegionObservation.rx.changes(in:)

GRDB 的 ValueObservation 跟踪数据库值的变化。您可以将其转换为 RxSwift 可观察对象

let observation = ValueObservation.tracking { db in
    try Player.fetchAll(db)
}

// Observable<[Player]>
let observable = observation.rx.observe(in: dbQueue)

此可观察对象的行为与 ValueObservation 相同

• 它在最终更改之前通知初始值。

• 它可能会将后续更改合并为单个通知。

• 它可能会通知连续的相同值。您可以使用 distinctUntilChanged() RxSwift 运算符过滤掉不需要的重复项，但我们建议您查看 removeDuplicates() GRDB 运算符。

• 在数据库连接关闭后，它会停止发出任何值。但它永远不会完成。

• 默认情况下，它会在主线程上异步通知初始值，以及最终的更改和错误。

  这可以使用 scheduling 参数进行配置。它不接受 RxSwift 调度器，但接受 GRDB 调度器。

  例如，.immediate 调度器确保在订阅可观察对象时立即通知初始值。它可以帮助您的应用程序更新用户界面，而无需等待任何异步通知

  // Immediate notification of the initial value
  let disposable = observation.rx
      .observe(
          in: dbQueue,
          scheduling: .immediate) // <-
      .subscribe(
          onNext: { players: [Player] in print("fresh players: \(players)") },
          onError: { error in ... })
  // <- here "fresh players" is already printed.

  请注意，.immediate 调度器要求从主线程订阅可观察对象。否则会引发致命错误。

有关更多信息，请参见 ValueObservation 调度。

⚠️ ValueObservation 和数据一致性

当您使用 combineLatest 运算符将 ValueObservation 可观察对象组合在一起时，您将失去所有 数据一致性保证。

相反，将请求组合成一个 ValueObservation，如下所示

// DATA CONSISTENCY GUARANTEED
let hallOfFameObservable = ValueObservation
    .tracking { db -> HallOfFame in
        let playerCount = try Player.fetchCount(db)
        let bestPlayers = try Player.limit(10).orderedByScore().fetchAll(db)
        return HallOfFame(playerCount:playerCount, bestPlayers:bestPlayers)
    }
    .rx.observe(in: dbQueue)

有关更多信息，请参见 ValueObservation。

GRDB 的 DatabaseRegionObservation 通知所有影响跟踪数据库区域的事务。您可以将其转换为 RxSwift 可观察对象

let request = Player.all()
let observation = DatabaseRegionObservation.tracking(request)

// Observable<Database>
let observable = observation.rx.changes(in: dbQueue)

此可观察对象可以从任何线程创建和订阅。它在“受保护的调度队列”中传递数据库连接，与所有数据库更新序列化。它仅在发生数据库错误时完成。

let request = Player.all()
let disposable = DatabaseRegionObservation
    .tracking(request)
    .rx.changes(in: dbQueue)
    .subscribe(
        onNext: { (db: Database) in
            print("Players have changed.")
        },
        onError: { error in ... })

try dbQueue.write { db in
    try Player(name: "Arthur").insert(db)
    try Player(name: "Barbara").insert(db)
} 
// Prints "Players have changed."

try dbQueue.write { db in
    try Player.deleteAll(db)
}
// Prints "Players have changed."

有关更多信息，请参见 DatabaseRegionObservation。