RxGRDB

对 SQLite、GRDB.swift 和 RxSwift 的一组扩展

最新发布：2021年10月17日 • 版本 2.1.0 • 发布说明

要求：iOS 11.0+ / OSX 10.10+ / tvOS 9.0+ / watchOS 2.0+ • Swift 5.2+ / Xcode 11.4+

使用方法

要连接到数据库，请参阅支持 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 ... })

文档

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

安装

要在 Swift 包管理器中使用 RxGRDB，请在您的 Package.swift 文件中添加依赖项

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

要使用 CocoaPods 使用 RxGRDB，请在您的 Podfile 中指定

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

异步数据库访问

RxGRDB 提供执行异步数据库访问的操作符。

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

DatabaseReader.rx.read(observeOn:value:)

这些方法返回一个在从数据库异步获取数据库值后完成的 Single。

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

尝试修改数据库时会导致订阅出错。

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

当您使用数据库池时，读取通常是非阻塞的，除非达到最大并发读取数。在此情况下，读取必须等待其他读取完成。可以通过配置此最大数目。

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

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

DatabaseWriter.rx.write(observeOn:updates:)

此方法返回一个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:)利用优化后的调度。见下文。

DatabaseWriter.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` 函数留下的数据库状态，同时不会阻塞任何并发写入。这可以减少数据库写入冲突。更多信息请参阅 高级数据库连接池。

当您使用数据库队列时，结果保证相同，但未应用调度优化。

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

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

数据库观察

数据库观察可观察对象基于 GRDB 的 ValueObservation 和 DatabaseRegionObservation。有关更多信息，请参阅它们的文档。如果您的应用需要 RxGRDB 中没有构建的更改通知，请查看 General Database Changes Observation 部分。

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

ValueObservation.rx.observe(in:scheduling:)

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 操作符过滤掉不想要的重复项，但我们也建议您查看 GRDB 操作符的 removeDuplicates()。

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

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

  这可以通过 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 中，如下所示

// 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。

DatabaseRegionObservation.rx.changes(in:)

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。