zeus
906 lines
39 KiB
Swift
906 lines
39 KiB
Swift
import Foundation
|
|
|
|
/// `ValueConcurrentObserver` observes the database for `ValueObservation`, in
|
|
/// a `DatabasePool`.
|
|
///
|
|
/// It performs the following database observation cycle:
|
|
///
|
|
/// 1. Start observation or detect a database change
|
|
/// 2. Fetch
|
|
/// 3. Reduce
|
|
/// 4. Notify
|
|
///
|
|
/// **Fetch** is performed concurrently (hence the name of this observer).
|
|
///
|
|
/// **Reduce** is the operation that turns the fetched database values into the
|
|
/// observed values. Those are not the same. Consider, for example, the `map()`
|
|
/// and `removeDuplicates()` operators: they perform their job during the
|
|
/// reducing stage.
|
|
///
|
|
/// **Notify** is calling user callbacks, in case of database change or error.
|
|
final class ValueConcurrentObserver<Reducer: ValueReducer, Scheduler: ValueObservationScheduler> {
|
|
// MARK: - Configuration
|
|
//
|
|
// Configuration is not mutable.
|
|
|
|
/// How to schedule observed values and errors.
|
|
private let scheduler: Scheduler
|
|
|
|
/// Configures the tracked database region.
|
|
private let trackingMode: ValueObservationTrackingMode
|
|
|
|
// MARK: - Mutable State
|
|
//
|
|
// The observer has four distinct mutable states that evolve independently,
|
|
// and are made thread-safe with various mechanisms:
|
|
//
|
|
// - A `DatabaseAccess`: ability to access the database. It is constant but
|
|
// turns nil after the observation fails or is cancelled, in order to
|
|
// release memory and resources when the observation completes. It is
|
|
// guarded by `lock`, because observation can fail or be cancelled from
|
|
// multiple threads.
|
|
//
|
|
// - A `NotificationCallbacks`: ability to notify observation events. It is
|
|
// constant but turns nil when failure or cancellation is notified, in
|
|
// order to release memory and resources when the observation completes.
|
|
// It is guarded by `lock`, because observation can fail or be cancelled
|
|
// from multiple threads.
|
|
//
|
|
// - An `ObservationState`: relationship with the `TransactionObserver`
|
|
// protocol. It is only accessed from the serialized writer
|
|
// dispatch queue.
|
|
//
|
|
// - A `Reducer`: the observation reducer, only accessed from the
|
|
// serialized dispatch queue `reduceQueue`.
|
|
//
|
|
// The `reduceQueue` guarantees that fresh value notifications have the same
|
|
// order as transactions. It is different from the serialized writer
|
|
// dispatch queue because we do not want to lock the database as
|
|
// computations (`map`, `removeDuplicates()`, etc.) are performed.
|
|
//
|
|
// Despite being protected by the same lock, `DatabaseAccess` and
|
|
// `NotificationCallbacks` are not merged together. This is because the
|
|
// observer does not lose `DatabaseAccess` at the same time it
|
|
// looses `NotificationCallbacks`:
|
|
//
|
|
// - In case of cancellation, `NotificationCallbacks` is lost first, and
|
|
// `DatabaseAccess` is lost asynchronously, after the observer could
|
|
// resign as a transaction observer. See `cancel()`.
|
|
//
|
|
// - In case of error, `DatabaseAccess` may be lost synchronously, and
|
|
// `NotificationCallbacks` is lost asynchronously, after the error could
|
|
// be notified. See error catching clauses.
|
|
|
|
/// Ability to access the database
|
|
private struct DatabaseAccess {
|
|
/// The observed DatabasePool.
|
|
let dbPool: DatabasePool
|
|
|
|
/// A reducer that fetches database values.
|
|
private let reducer: Reducer
|
|
|
|
init(dbPool: DatabasePool, reducer: Reducer) {
|
|
self.dbPool = dbPool
|
|
self.reducer = reducer
|
|
}
|
|
|
|
func fetch(_ db: Database) throws -> Reducer.Fetched {
|
|
try db.isolated(readOnly: true) {
|
|
try reducer._fetch(db)
|
|
}
|
|
}
|
|
|
|
func fetchRecordingObservedRegion(_ db: Database) throws -> (Reducer.Fetched, DatabaseRegion) {
|
|
var region = DatabaseRegion()
|
|
let fetchedValue = try db.isolated(readOnly: true) {
|
|
try db.recordingSelection(®ion) {
|
|
try reducer._fetch(db)
|
|
}
|
|
}
|
|
return try (fetchedValue, region.observableRegion(db))
|
|
}
|
|
}
|
|
|
|
/// The fetching state for observation of constant regions.
|
|
enum FetchingState {
|
|
/// No need to fetch.
|
|
case idle
|
|
|
|
/// Waiting for a fetched value.
|
|
case fetching
|
|
|
|
/// Waiting for a fetched value, and for a subsequent fetch after
|
|
/// that, because a change has been detected as we were fetching.
|
|
case fetchingAndNeedsFetch
|
|
}
|
|
|
|
/// Ability to notify observation events
|
|
private struct NotificationCallbacks {
|
|
let events: ValueObservationEvents
|
|
let onChange: (Reducer.Value) -> Void
|
|
}
|
|
|
|
/// Relationship with the `TransactionObserver` protocol
|
|
private struct ObservationState {
|
|
var region: DatabaseRegion?
|
|
var isModified = false
|
|
|
|
static var notObserving: Self { .init(region: nil, isModified: false) }
|
|
}
|
|
|
|
/// Protects `databaseAccess` and `notificationCallbacks`.
|
|
///
|
|
/// Check out this compiler bug:
|
|
/// - <https://github.com/groue/GRDB.swift/issues/1026>
|
|
/// - <https://github.com/groue/GRDB.swift/pull/1025>
|
|
private let lock = NSLock()
|
|
|
|
/// The dispatch queue where database values are reduced into observed
|
|
/// values before being notified. Protects `reducer`.
|
|
private let reduceQueue: DispatchQueue
|
|
|
|
/// Access to the database, protected by `lock`.
|
|
private var databaseAccess: DatabaseAccess?
|
|
|
|
/// Ability to notify observation events, protected by `lock`.
|
|
private var notificationCallbacks: NotificationCallbacks?
|
|
|
|
/// The fetching state for observation of constant regions.
|
|
@LockedBox private var fetchingState = FetchingState.idle
|
|
|
|
/// Support for `TransactionObserver`, protected by the serialized writer
|
|
/// dispatch queue.
|
|
private var observationState = ObservationState.notObserving
|
|
|
|
/// Protected by `reduceQueue`.
|
|
private var reducer: Reducer
|
|
|
|
init(
|
|
dbPool: DatabasePool,
|
|
scheduler: Scheduler,
|
|
trackingMode: ValueObservationTrackingMode,
|
|
reducer: Reducer,
|
|
events: ValueObservationEvents,
|
|
onChange: @escaping (Reducer.Value) -> Void)
|
|
{
|
|
// Configuration
|
|
self.scheduler = scheduler
|
|
self.trackingMode = trackingMode
|
|
|
|
// State
|
|
self.databaseAccess = DatabaseAccess(
|
|
dbPool: dbPool,
|
|
// ValueReducer semantics guarantees that reducer._fetch
|
|
// is independent from the reducer state
|
|
reducer: reducer)
|
|
self.notificationCallbacks = NotificationCallbacks(events: events, onChange: onChange)
|
|
self.reducer = reducer
|
|
self.reduceQueue = DispatchQueue(
|
|
label: dbPool.configuration.identifier(
|
|
defaultLabel: "GRDB",
|
|
purpose: "ValueObservation"),
|
|
qos: dbPool.configuration.readQoS)
|
|
}
|
|
}
|
|
|
|
// MARK: - Starting the Observation
|
|
//
|
|
// When we start an observation from a `DatabasePool`, we do not wait for an
|
|
// access to the writer connection before fetching the initial value. That is
|
|
// because the user of a `DatabasePool` expects to be notified with the initial
|
|
// value as fast as possible, even if a long write transaction is running in the
|
|
// background.
|
|
//
|
|
// We will thus perform the initial fetch from a reader connection, and only
|
|
// then access the writer connection, and start database observation.
|
|
//
|
|
// Between this initial fetch, and the beginning of database observation, any
|
|
// number of unobserved writes may occur. We must notify the changes that happen
|
|
// during this unobserved window. But how do we spot them, since we were not
|
|
// observing the database yet?
|
|
//
|
|
// The solution depends on the presence of the `SQLITE_ENABLE_SNAPSHOT`
|
|
// SQLite compilation flag.
|
|
//
|
|
// Without `SQLITE_ENABLE_SNAPSHOT`, we have NO WAY to detect if the database
|
|
// was changed or not between the initial fetch and the beginning of database
|
|
// observation. We will thus always perform a secondary fetch from the initial
|
|
// access to the writer connection. Even if no change was performed. We may end
|
|
// up notifying the same value twice. Such stuttering is a documented glitch,
|
|
// and the user can perform deduplication with the
|
|
// `removeDuplicates()` operator.
|
|
//
|
|
// With `SQLITE_ENABLE_SNAPSHOT`, we can detect if the database was not changed
|
|
// at all between the initial fetch and the beginning of database observation.
|
|
// If the database was changed, we perform a secondary fetch from the initial
|
|
// access to the writer connection. It is possible that the change was not
|
|
// related to the observed value. Actually we have NO WAY to know. So we may end
|
|
// up notifying the same value twice. Such stuttering is a documented glitch,
|
|
// and the user can perform deduplication with the
|
|
// `removeDuplicates()` operator.
|
|
//
|
|
// This is how we can both:
|
|
// 1. Start the observation without waiting for a write access (the expected
|
|
// benefit of `DatabasePool`).
|
|
// 2. Make sure we do not miss a change (a documented guarantee)
|
|
//
|
|
// Support for `SQLITE_ENABLE_SNAPSHOT` is implemented by our
|
|
// `WALSnapshot` class.
|
|
extension ValueConcurrentObserver {
|
|
// Starts the observation
|
|
func start() -> AnyDatabaseCancellable {
|
|
let (notificationCallbacks, databaseAccess) = lock.synchronized {
|
|
(self.notificationCallbacks, self.databaseAccess)
|
|
}
|
|
guard let notificationCallbacks, let databaseAccess else {
|
|
// Likely a GRDB bug: during a synchronous start, user is not
|
|
// able to cancel observation.
|
|
fatalError("can't start a cancelled or failed observation")
|
|
}
|
|
|
|
if scheduler.immediateInitialValue() {
|
|
do {
|
|
// Start the observation in an synchronous way
|
|
let initialValue = try syncStart(from: databaseAccess)
|
|
|
|
// Notify the initial value from the dispatch queue the
|
|
// observation was started from
|
|
notificationCallbacks.onChange(initialValue)
|
|
} catch {
|
|
// Notify error from the dispatch queue the observation
|
|
// was started from.
|
|
notificationCallbacks.events.didFail?(error)
|
|
|
|
// Early return!
|
|
return AnyDatabaseCancellable { /* nothing to cancel */ }
|
|
}
|
|
} else {
|
|
// Start the observation in an asynchronous way
|
|
asyncStart(from: databaseAccess)
|
|
}
|
|
|
|
// Make sure the returned cancellable cancels the observation
|
|
// when deallocated. We can't relying on the deallocation of
|
|
// self to trigger early cancellation, because self may be retained by
|
|
// some closure waiting to run in some DispatchQueue.
|
|
return AnyDatabaseCancellable(self)
|
|
}
|
|
|
|
private func startObservation(_ writerDB: Database, observedRegion: DatabaseRegion) {
|
|
observationState.region = observedRegion
|
|
assert(observationState.isModified == false)
|
|
writerDB.add(transactionObserver: self, extent: .observerLifetime)
|
|
}
|
|
}
|
|
|
|
// swiftlint:disable:next line_length
|
|
#if SQLITE_ENABLE_SNAPSHOT || (!GRDBCUSTOMSQLITE && !GRDBCIPHER && (compiler(>=5.7.1) || !(os(macOS) || targetEnvironment(macCatalyst))))
|
|
extension ValueConcurrentObserver {
|
|
/// Synchronously starts the observation, and returns the initial value.
|
|
///
|
|
/// Unlike `asyncStart()`, this method does not notify the initial value or error.
|
|
private func syncStart(from databaseAccess: DatabaseAccess) throws -> Reducer.Value {
|
|
// Start from a read access. The whole point of using a DatabasePool
|
|
// for observing the database is to be able to fetch the initial value
|
|
// without having to wait for an eventual long-running write
|
|
// transaction to complete.
|
|
//
|
|
// We perform the initial read from a long-lived WAL snapshot
|
|
// transaction, because it is a handy way to keep a read transaction
|
|
// open until we grab a write access, and compare the database versions.
|
|
let initialFetchTransaction: WALSnapshotTransaction
|
|
do {
|
|
initialFetchTransaction = try databaseAccess.dbPool.walSnapshotTransaction()
|
|
} catch DatabaseError.SQLITE_ERROR {
|
|
// We can't create a WAL snapshot. The WAL file is probably
|
|
// missing, or is truncated. Let's degrade the observation
|
|
// by not using any snapshot.
|
|
// For more information, see <https://github.com/groue/GRDB.swift/issues/1383>
|
|
return try syncStartWithoutWALSnapshot(from: databaseAccess)
|
|
}
|
|
|
|
let (fetchedValue, initialRegion): (Reducer.Fetched, DatabaseRegion) = try initialFetchTransaction.read { db in
|
|
switch trackingMode {
|
|
case let .constantRegion(regions):
|
|
let fetchedValue = try databaseAccess.fetch(db)
|
|
let region = try DatabaseRegion.union(regions)(db)
|
|
let initialRegion = try region.observableRegion(db)
|
|
return (fetchedValue, initialRegion)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
let (fetchedValue, initialRegion) = try databaseAccess.fetchRecordingObservedRegion(db)
|
|
return (fetchedValue, initialRegion)
|
|
}
|
|
}
|
|
|
|
// Reduce
|
|
let initialValue = try reduceQueue.sync {
|
|
guard let initialValue = try reducer._value(fetchedValue) else {
|
|
fatalError("Broken contract: reducer has no initial value")
|
|
}
|
|
return initialValue
|
|
}
|
|
|
|
// Start observation
|
|
asyncStartObservation(
|
|
from: databaseAccess,
|
|
initialFetchTransaction: initialFetchTransaction,
|
|
initialRegion: initialRegion)
|
|
|
|
return initialValue
|
|
}
|
|
|
|
/// Asynchronously starts the observation
|
|
///
|
|
/// Unlike `syncStart()`, this method does notify the initial value or error.
|
|
private func asyncStart(from databaseAccess: DatabaseAccess) {
|
|
// Start from a read access. The whole point of using a DatabasePool
|
|
// for observing the database is to be able to fetch the initial value
|
|
// without having to wait for an eventual long-running write
|
|
// transaction to complete.
|
|
//
|
|
// We perform the initial read from a long-lived WAL snapshot
|
|
// transaction, because it is a handy way to keep a read transaction
|
|
// open until we grab a write access, and compare the database versions.
|
|
databaseAccess.dbPool.asyncWALSnapshotTransaction { result in
|
|
let (isNotifying, databaseAccess) = self.lock.synchronized {
|
|
(self.notificationCallbacks != nil, self.databaseAccess)
|
|
}
|
|
guard isNotifying, let databaseAccess else { return /* Cancelled */ }
|
|
|
|
do {
|
|
let initialFetchTransaction = try result.get()
|
|
// Second async jump because that's how
|
|
// `DatabasePool.asyncWALSnapshotTransaction` has to be used.
|
|
initialFetchTransaction.asyncRead { db in
|
|
do {
|
|
let fetchedValue: Reducer.Fetched
|
|
let initialRegion: DatabaseRegion
|
|
|
|
switch self.trackingMode {
|
|
case let .constantRegion(regions):
|
|
fetchedValue = try databaseAccess.fetch(db)
|
|
let region = try DatabaseRegion.union(regions)(db)
|
|
initialRegion = try region.observableRegion(db)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
(fetchedValue, initialRegion) = try databaseAccess.fetchRecordingObservedRegion(db)
|
|
}
|
|
|
|
// Reduce
|
|
//
|
|
// Reducing is performed asynchronously, so that we do not lock
|
|
// a database dispatch queue longer than necessary.
|
|
self.reduceQueue.async {
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
do {
|
|
guard let initialValue = try self.reducer._value(fetchedValue) else {
|
|
fatalError("Broken contract: reducer has no initial value")
|
|
}
|
|
|
|
// Notify
|
|
self.scheduler.schedule {
|
|
let onChange = self.lock.synchronized { self.notificationCallbacks?.onChange }
|
|
guard let onChange else { return /* Cancelled */ }
|
|
onChange(initialValue)
|
|
}
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
|
|
// Start observation
|
|
self.asyncStartObservation(
|
|
from: databaseAccess,
|
|
initialFetchTransaction: initialFetchTransaction,
|
|
initialRegion: initialRegion)
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
} catch DatabaseError.SQLITE_ERROR {
|
|
// We can't create a WAL snapshot. The WAL file is probably
|
|
// missing, or is truncated. Let's degrade the observation
|
|
// by not using any snapshot.
|
|
// For more information, see <https://github.com/groue/GRDB.swift/issues/1383>
|
|
self.asyncStartWithoutWALSnapshot(from: databaseAccess)
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func asyncStartObservation(
|
|
from databaseAccess: DatabaseAccess,
|
|
initialFetchTransaction: WALSnapshotTransaction,
|
|
initialRegion: DatabaseRegion)
|
|
{
|
|
// We'll start the observation when we can access the writer
|
|
// connection. Until then, maybe the database has been modified
|
|
// since the initial fetch: we'll then need to notify a fresh value.
|
|
//
|
|
// To know if the database has been modified between the initial
|
|
// fetch and the writer access, we'll compare WAL snapshots.
|
|
//
|
|
// WAL snapshots can only be compared if the database is not
|
|
// checkpointed. That's why we'll keep `initialFetchTransaction`
|
|
// alive until the comparison is done.
|
|
//
|
|
// However, we want to release `initialFetchTransaction` as soon as
|
|
// possible, so that the reader connection it holds becomes
|
|
// available for other reads. It will be released when this optional
|
|
// is set to nil:
|
|
var initialFetchTransaction: WALSnapshotTransaction? = initialFetchTransaction
|
|
|
|
databaseAccess.dbPool.asyncWriteWithoutTransaction { writerDB in
|
|
let events = self.lock.synchronized { self.notificationCallbacks?.events }
|
|
guard let events else { return /* Cancelled */ }
|
|
|
|
do {
|
|
var observedRegion = initialRegion
|
|
|
|
try writerDB.isolated(readOnly: true) {
|
|
// Was the database modified since the initial fetch?
|
|
let isModified: Bool
|
|
if let currentWALSnapshot = try? WALSnapshot(writerDB) {
|
|
let ordering = initialFetchTransaction!.walSnapshot.compare(currentWALSnapshot)
|
|
assert(ordering <= 0, "Unexpected snapshot ordering")
|
|
isModified = ordering < 0
|
|
} else {
|
|
// Can't compare: assume the database was modified.
|
|
isModified = true
|
|
}
|
|
|
|
// Comparison done: end the WAL snapshot transaction
|
|
// and release its reader connection.
|
|
initialFetchTransaction = nil
|
|
|
|
if isModified {
|
|
events.databaseDidChange?()
|
|
|
|
// Fetch
|
|
let fetchedValue: Reducer.Fetched
|
|
|
|
switch self.trackingMode {
|
|
case .constantRegion:
|
|
fetchedValue = try databaseAccess.fetch(writerDB)
|
|
events.willTrackRegion?(initialRegion)
|
|
self.startObservation(writerDB, observedRegion: initialRegion)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
(fetchedValue, observedRegion) = try databaseAccess.fetchRecordingObservedRegion(writerDB)
|
|
events.willTrackRegion?(observedRegion)
|
|
self.startObservation(writerDB, observedRegion: observedRegion)
|
|
}
|
|
|
|
// Reduce
|
|
//
|
|
// Reducing is performed asynchronously, so that we do not lock
|
|
// the writer dispatch queue longer than necessary.
|
|
//
|
|
// Important: reduceQueue.async guarantees the same ordering
|
|
// between transactions and notifications!
|
|
self.reduceQueue.async {
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
do {
|
|
let value = try self.reducer._value(fetchedValue)
|
|
|
|
// Notify
|
|
if let value {
|
|
self.scheduler.schedule {
|
|
let onChange = self.lock.synchronized { self.notificationCallbacks?.onChange }
|
|
guard let onChange else { return /* Cancelled */ }
|
|
onChange(value)
|
|
}
|
|
}
|
|
} catch {
|
|
let dbPool = self.lock.synchronized { self.databaseAccess?.dbPool }
|
|
dbPool?.asyncWriteWithoutTransaction { writerDB in
|
|
self.stopDatabaseObservation(writerDB)
|
|
}
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
} else {
|
|
events.willTrackRegion?(initialRegion)
|
|
self.startObservation(writerDB, observedRegion: initialRegion)
|
|
}
|
|
}
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
#else
|
|
extension ValueConcurrentObserver {
|
|
private func syncStart(from databaseAccess: DatabaseAccess) throws -> Reducer.Value {
|
|
try syncStartWithoutWALSnapshot(from: databaseAccess)
|
|
}
|
|
|
|
private func asyncStart(from databaseAccess: DatabaseAccess) {
|
|
asyncStartWithoutWALSnapshot(from: databaseAccess)
|
|
}
|
|
}
|
|
#endif
|
|
|
|
extension ValueConcurrentObserver {
|
|
/// Synchronously starts the observation, and returns the initial value.
|
|
///
|
|
/// Unlike `asyncStartWithoutWALSnapshot()`, this method does not notify the initial value or error.
|
|
private func syncStartWithoutWALSnapshot(from databaseAccess: DatabaseAccess) throws -> Reducer.Value {
|
|
// Start from a read access. The whole point of using a DatabasePool
|
|
// for observing the database is to be able to fetch the initial value
|
|
// without having to wait for an eventual long-running write
|
|
// transaction to complete.
|
|
let (fetchedValue, initialRegion) = try databaseAccess.dbPool.read { db -> (Reducer.Fetched, DatabaseRegion) in
|
|
switch trackingMode {
|
|
case let .constantRegion(regions):
|
|
let fetchedValue = try databaseAccess.fetch(db)
|
|
let region = try DatabaseRegion.union(regions)(db)
|
|
let initialRegion = try region.observableRegion(db)
|
|
return (fetchedValue, initialRegion)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
let (fetchedValue, initialRegion) = try databaseAccess.fetchRecordingObservedRegion(db)
|
|
return (fetchedValue, initialRegion)
|
|
}
|
|
}
|
|
|
|
// Reduce
|
|
let initialValue = try reduceQueue.sync {
|
|
guard let initialValue = try reducer._value(fetchedValue) else {
|
|
fatalError("Broken contract: reducer has no initial value")
|
|
}
|
|
return initialValue
|
|
}
|
|
|
|
// Start observation
|
|
asyncStartObservationWithoutWALSnapshot(
|
|
from: databaseAccess,
|
|
initialRegion: initialRegion)
|
|
|
|
return initialValue
|
|
}
|
|
|
|
/// Asynchronously starts the observation
|
|
///
|
|
/// Unlike `syncStartWithoutWALSnapshot()`, this method does notify the initial value or error.
|
|
private func asyncStartWithoutWALSnapshot(from databaseAccess: DatabaseAccess) {
|
|
// Start from a read access. The whole point of using a DatabasePool
|
|
// for observing the database is to be able to fetch the initial value
|
|
// without having to wait for an eventual long-running write
|
|
// transaction to complete.
|
|
databaseAccess.dbPool.asyncRead { dbResult in
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
do {
|
|
// Fetch
|
|
let fetchedValue: Reducer.Fetched
|
|
let initialRegion: DatabaseRegion
|
|
let db = try dbResult.get()
|
|
switch self.trackingMode {
|
|
case let .constantRegion(regions):
|
|
fetchedValue = try databaseAccess.fetch(db)
|
|
let region = try DatabaseRegion.union(regions)(db)
|
|
initialRegion = try region.observableRegion(db)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
(fetchedValue, initialRegion) = try databaseAccess.fetchRecordingObservedRegion(db)
|
|
}
|
|
|
|
// Reduce
|
|
//
|
|
// Reducing is performed asynchronously, so that we do not lock
|
|
// a database dispatch queue longer than necessary.
|
|
self.reduceQueue.async {
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
do {
|
|
guard let initialValue = try self.reducer._value(fetchedValue) else {
|
|
fatalError("Broken contract: reducer has no initial value")
|
|
}
|
|
|
|
// Notify
|
|
self.scheduler.schedule {
|
|
let onChange = self.lock.synchronized { self.notificationCallbacks?.onChange }
|
|
guard let onChange else { return /* Cancelled */ }
|
|
onChange(initialValue)
|
|
}
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
|
|
// Start observation
|
|
self.asyncStartObservationWithoutWALSnapshot(
|
|
from: databaseAccess,
|
|
initialRegion: initialRegion)
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func asyncStartObservationWithoutWALSnapshot(
|
|
from databaseAccess: DatabaseAccess,
|
|
initialRegion: DatabaseRegion)
|
|
{
|
|
databaseAccess.dbPool.asyncWriteWithoutTransaction { writerDB in
|
|
let events = self.lock.synchronized { self.notificationCallbacks?.events }
|
|
guard let events else { return /* Cancelled */ }
|
|
events.databaseDidChange?()
|
|
|
|
do {
|
|
try writerDB.isolated(readOnly: true) {
|
|
// Fetch
|
|
let fetchedValue: Reducer.Fetched
|
|
let observedRegion: DatabaseRegion
|
|
switch self.trackingMode {
|
|
case .constantRegion:
|
|
fetchedValue = try databaseAccess.fetch(writerDB)
|
|
observedRegion = initialRegion
|
|
events.willTrackRegion?(initialRegion)
|
|
self.startObservation(writerDB, observedRegion: initialRegion)
|
|
|
|
case .constantRegionRecordedFromSelection,
|
|
.nonConstantRegionRecordedFromSelection:
|
|
(fetchedValue, observedRegion) = try databaseAccess.fetchRecordingObservedRegion(writerDB)
|
|
events.willTrackRegion?(observedRegion)
|
|
self.startObservation(writerDB, observedRegion: observedRegion)
|
|
}
|
|
|
|
// Reduce
|
|
//
|
|
// Reducing is performed asynchronously, so that we do not lock
|
|
// the writer dispatch queue longer than necessary.
|
|
//
|
|
// Important: reduceQueue.async guarantees the same ordering
|
|
// between transactions and notifications!
|
|
self.reduceQueue.async {
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
do {
|
|
let value = try self.reducer._value(fetchedValue)
|
|
|
|
// Notify
|
|
if let value {
|
|
self.scheduler.schedule {
|
|
let onChange = self.lock.synchronized { self.notificationCallbacks?.onChange }
|
|
guard let onChange else { return /* Cancelled */ }
|
|
onChange(value)
|
|
}
|
|
}
|
|
} catch {
|
|
let dbPool = self.lock.synchronized { self.databaseAccess?.dbPool }
|
|
dbPool?.asyncWriteWithoutTransaction { writerDB in
|
|
self.stopDatabaseObservation(writerDB)
|
|
}
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
} catch {
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// MARK: - Observing Database Transactions
|
|
|
|
extension ValueConcurrentObserver: TransactionObserver {
|
|
func observes(eventsOfKind eventKind: DatabaseEventKind) -> Bool {
|
|
if let region = observationState.region {
|
|
return region.isModified(byEventsOfKind: eventKind)
|
|
} else {
|
|
return false
|
|
}
|
|
}
|
|
|
|
func databaseDidChange() {
|
|
// Database was modified!
|
|
observationState.isModified = true
|
|
// We can stop observing the current transaction
|
|
stopObservingDatabaseChangesUntilNextTransaction()
|
|
}
|
|
|
|
func databaseDidChange(with event: DatabaseEvent) {
|
|
if let region = observationState.region, region.isModified(by: event) {
|
|
// Database was modified!
|
|
observationState.isModified = true
|
|
// We can stop observing the current transaction
|
|
stopObservingDatabaseChangesUntilNextTransaction()
|
|
}
|
|
}
|
|
|
|
func databaseDidCommit(_ writerDB: Database) {
|
|
// Ignore transaction unless database was modified
|
|
guard observationState.isModified else { return }
|
|
|
|
// Reset the isModified flag until next transaction
|
|
observationState.isModified = false
|
|
|
|
// Ignore transaction unless we are still notifying database events, and
|
|
// we can still access the database.
|
|
let (events, databaseAccess) = lock.synchronized {
|
|
(notificationCallbacks?.events, self.databaseAccess)
|
|
}
|
|
guard let events, let databaseAccess else { return /* Cancelled */ }
|
|
|
|
events.databaseDidChange?()
|
|
|
|
// Fetch
|
|
switch trackingMode {
|
|
case .constantRegion, .constantRegionRecordedFromSelection:
|
|
setNeedsFetching(databaseAccess: databaseAccess)
|
|
|
|
case .nonConstantRegionRecordedFromSelection:
|
|
// When the tracked region is not constant, we can't perform
|
|
// concurrent fetches of observed values.
|
|
//
|
|
// This is because after a concurrent fetch has acquired snapshot
|
|
// isolation, and before it completes, a change can be performed
|
|
// in the *next* tracked region. When this happens, the
|
|
// concurrent fetch has loaded an obsolete value, and we need to
|
|
// perform a new fetch, with the latest values. But the
|
|
// observation was not triggered by the change because we didn't
|
|
// know that this change was about to be tracked! This means
|
|
// that we'd miss a change, and fail notifying the latest value.
|
|
//
|
|
// Conclusion: fetch from the writer connection, and update the
|
|
// tracked region.
|
|
do {
|
|
let (fetchedValue, observedRegion) = try databaseAccess.fetchRecordingObservedRegion(writerDB)
|
|
|
|
// Don't spam the user with region tracking events: wait for an actual change
|
|
if let willTrackRegion = events.willTrackRegion, observedRegion != observationState.region {
|
|
willTrackRegion(observedRegion)
|
|
}
|
|
|
|
observationState.region = observedRegion
|
|
reduce(.success(fetchedValue))
|
|
} catch {
|
|
stopDatabaseObservation(writerDB)
|
|
notifyError(error)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
private func setNeedsFetching(databaseAccess: DatabaseAccess) {
|
|
$fetchingState.update { state in
|
|
switch state {
|
|
case .idle:
|
|
state = .fetching
|
|
asyncFetch(databaseAccess: databaseAccess)
|
|
|
|
case .fetching:
|
|
state = .fetchingAndNeedsFetch
|
|
|
|
case .fetchingAndNeedsFetch:
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
private func asyncFetch(databaseAccess: DatabaseAccess) {
|
|
databaseAccess.dbPool.asyncRead { [self] dbResult in
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
let fetchResult = dbResult.flatMap { db in
|
|
Result { try databaseAccess.fetch(db) }
|
|
}
|
|
|
|
self.reduce(fetchResult)
|
|
|
|
$fetchingState.update { state in
|
|
switch state {
|
|
case .idle:
|
|
// GRDB bug
|
|
preconditionFailure()
|
|
|
|
case .fetching:
|
|
state = .idle
|
|
|
|
case .fetchingAndNeedsFetch:
|
|
state = .fetching
|
|
asyncFetch(databaseAccess: databaseAccess)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private func reduce(_ fetchResult: Result<Reducer.Fetched, Error>) {
|
|
reduceQueue.async {
|
|
do {
|
|
let fetchedValue = try fetchResult.get()
|
|
|
|
let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
|
|
guard isNotifying else { return /* Cancelled */ }
|
|
|
|
let value = try self.reducer._value(fetchedValue)
|
|
|
|
// Notify value
|
|
if let value {
|
|
self.scheduler.schedule {
|
|
let onChange = self.lock.synchronized { self.notificationCallbacks?.onChange }
|
|
guard let onChange else { return /* Cancelled */ }
|
|
onChange(value)
|
|
}
|
|
}
|
|
} catch {
|
|
let dbPool = self.lock.synchronized { self.databaseAccess?.dbPool }
|
|
dbPool?.asyncWriteWithoutTransaction { writerDB in
|
|
self.stopDatabaseObservation(writerDB)
|
|
}
|
|
self.notifyError(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
func databaseDidRollback(_ db: Database) {
|
|
// Reset the isModified flag until next transaction
|
|
observationState.isModified = false
|
|
}
|
|
}
|
|
|
|
// MARK: - Ending the Observation
|
|
|
|
extension ValueConcurrentObserver: DatabaseCancellable {
|
|
func cancel() {
|
|
// Notify cancellation
|
|
let (events, dbPool): (ValueObservationEvents?, DatabasePool?) = lock.synchronized {
|
|
let events = notificationCallbacks?.events
|
|
// Set callbacks to nil so that we can't notify anything after
|
|
// the cancellation.
|
|
notificationCallbacks = nil
|
|
return (events, databaseAccess?.dbPool)
|
|
}
|
|
|
|
guard let events else { return /* Cancelled or failed */ }
|
|
events.didCancel?()
|
|
|
|
// Stop observing the database
|
|
// Do it asynchronously, so that we do not block the current thread:
|
|
// cancellation may be triggered while a long write access is executing.
|
|
guard let dbPool else { return /* Failed */ }
|
|
dbPool.asyncWriteWithoutTransaction { db in
|
|
self.stopDatabaseObservation(db)
|
|
}
|
|
}
|
|
|
|
func notifyError(_ error: Error) {
|
|
scheduler.schedule {
|
|
let events = self.lock.synchronized {
|
|
let events = self.notificationCallbacks?.events
|
|
self.notificationCallbacks = nil
|
|
return events
|
|
}
|
|
guard let events else { return /* Cancelled */ }
|
|
events.didFail?(error)
|
|
}
|
|
}
|
|
|
|
private func stopDatabaseObservation(_ writerDB: Database) {
|
|
writerDB.remove(transactionObserver: self)
|
|
observationState = .notObserving
|
|
lock.synchronized {
|
|
databaseAccess = nil
|
|
}
|
|
}
|
|
}
|