380 lines
14 KiB
Swift
380 lines
14 KiB
Swift
import Foundation
|
|
|
|
/// The extent of the shared subscription to a ``SharedValueObservation``.
|
|
public enum SharedValueObservationExtent: Sendable {
|
|
/// The ``SharedValueObservation`` starts a single database observation,
|
|
/// which stops when the `SharedValueObservation` is deallocated and all
|
|
/// subscriptions terminated.
|
|
///
|
|
/// This extent prevents the shared observation from recovering from
|
|
/// database errors. To recover from database errors, you must create a new
|
|
/// shared `SharedValueObservation` instance.
|
|
case observationLifetime
|
|
|
|
/// The ``SharedValueObservation`` stops database observation when the
|
|
/// number of subscriptions drops down to zero. Database observation
|
|
/// restarts on the next subscription.
|
|
///
|
|
/// Database errors can be recovered by resubscribing to the
|
|
/// shared observation.
|
|
case whileObserved
|
|
}
|
|
|
|
extension ValueObservation {
|
|
/// Returns a shared value observation that spares database resources by
|
|
/// sharing a single underlying ``ValueObservation`` subscription.
|
|
///
|
|
/// - note: [**🔥 EXPERIMENTAL**](https://github.com/groue/GRDB.swift/blob/master/README.md#what-are-experimental-features)
|
|
///
|
|
/// For example:
|
|
///
|
|
/// ```swift
|
|
/// let observation = ValueObservation.tracking { db in
|
|
/// try Player.fetchAll(db)
|
|
/// }
|
|
///
|
|
/// let sharedObservation = observation.shared(in: dbQueue)
|
|
///
|
|
/// let cancellable = try sharedObservation.start { error in
|
|
/// // handle error
|
|
/// } onChange: { (players: [Player]) in
|
|
/// print("Fresh players: \(players)")
|
|
/// }
|
|
/// ```
|
|
///
|
|
/// The underlying subscription is shared if and only if you start observing
|
|
/// the database from the same `SharedValueObservation` instance:
|
|
///
|
|
/// ```swift
|
|
/// // Shared
|
|
/// let sharedObservation = ValueObservation.tracking { db in ... }.shared(in: dbQueue)
|
|
/// let cancellable1 = sharedObservation.start(...)
|
|
/// let cancellable2 = sharedObservation.start(...)
|
|
///
|
|
/// // NOT shared
|
|
/// let cancellable1 = ValueObservation.tracking { db in ... }.shared(in: dbQueue).start(...)
|
|
/// let cancellable2 = ValueObservation.tracking { db in ... }.shared(in: dbQueue).start(...)
|
|
/// ```
|
|
///
|
|
/// A shared observation starts observing the database as soon as it is
|
|
/// subscribed. You can choose if database observation should stop, or not,
|
|
/// when its number of subscriptions drops down to zero, with the `extent`
|
|
/// parameter:
|
|
///
|
|
/// ```swift
|
|
/// // The default: stops observing the database when the number of
|
|
/// // subscriptions drops down to zero, and restart database observation
|
|
/// // on the next subscription.
|
|
/// //
|
|
/// // Database errors can be recovered by resubscribing to the
|
|
/// // shared observation.
|
|
/// let sharedObservation = ValueObservation
|
|
/// .tracking { db in try Player.fetchAll(db) }
|
|
/// .shared(in: dbQueue, extent: .whileObserved)
|
|
///
|
|
/// // Only stops observing the database when the shared observation
|
|
/// // is deinitialized, and all subscriptions are cancelled.
|
|
/// //
|
|
/// // This extent prevents the shared observation from recovering
|
|
/// // from database errors. To recover from database errors, create a new
|
|
/// // shared SharedValueObservation instance.
|
|
/// let sharedObservation = ValueObservation
|
|
/// .tracking { db in try Player.fetchAll(db) }
|
|
/// .shared(in: dbQueue, extent: .observationLifetime)
|
|
/// ```
|
|
///
|
|
/// By default, fresh values are dispatched asynchronously on the
|
|
/// main dispatch queue. You can change this behavior by providing a
|
|
/// scheduler.
|
|
///
|
|
/// For example, the ``ValueObservationScheduler/immediate`` scheduler
|
|
/// notifies all values on the main dispatch queue, and notifies the first
|
|
/// one immediately when the
|
|
/// ``SharedValueObservation/start(onError:onChange:)`` method is called.
|
|
/// The `immediate` scheduling requires that the observation starts from the
|
|
/// main thread (a fatal error is raised otherwise):
|
|
///
|
|
/// ```swift
|
|
/// let observation = ValueObservation.tracking { db in
|
|
/// try Player.fetchAll(db)
|
|
/// }
|
|
///
|
|
/// let sharedObservation = observation.shared(
|
|
/// in: dbQueue,
|
|
/// scheduling: .immediate)
|
|
///
|
|
/// let cancellable = try sharedObservation.start { error in
|
|
/// // handle error
|
|
/// } onChange: { (players: [Player]) in
|
|
/// print("Fresh players: \(players)")
|
|
/// }
|
|
/// // <- here "Fresh players" is already printed.
|
|
/// ```
|
|
///
|
|
/// Note that the `.immediate` scheduler requires that the observation is
|
|
/// subscribed from the main thread. It raises a fatal error otherwise.
|
|
///
|
|
/// - parameter reader: A DatabaseReader.
|
|
/// - parameter scheduler: A Scheduler. By default, fresh values are
|
|
/// dispatched asynchronously on the main queue.
|
|
/// - parameter extent: The extent of the shared database observation.
|
|
/// - returns: A `SharedValueObservation`
|
|
public func shared(
|
|
in reader: some DatabaseReader,
|
|
scheduling scheduler: some ValueObservationScheduler = .async(onQueue: .main),
|
|
extent: SharedValueObservationExtent = .whileObserved)
|
|
-> SharedValueObservation<Reducer.Value>
|
|
where Reducer: ValueReducer
|
|
{
|
|
SharedValueObservation(scheduling: scheduler, extent: extent) { onError, onChange in
|
|
self.start(in: reader, scheduling: scheduler, onError: onError, onChange: onChange)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A shared value observation spares database resources by sharing a single
|
|
/// underlying ``ValueObservation`` subscription.
|
|
///
|
|
/// - note: [**🔥 EXPERIMENTAL**](https://github.com/groue/GRDB.swift/blob/master/README.md#what-are-experimental-features)
|
|
///
|
|
/// You build a `SharedValueObservation` with the ``ValueObservation`` method
|
|
/// ``ValueObservation/shared(in:scheduling:extent:)``. For example:
|
|
///
|
|
/// ```swift
|
|
/// let observation = ValueObservation.tracking { db in
|
|
/// try Player.fetchAll(db)
|
|
/// }
|
|
///
|
|
/// let sharedObservation = observation.shared(in: dbQueue)
|
|
///
|
|
/// let cancellable = try sharedObservation.start { error in
|
|
/// // handle error
|
|
/// } onChange: { (players: [Player]) in
|
|
/// print("Fresh players: \(players)")
|
|
/// }
|
|
/// ```
|
|
///
|
|
/// The underlying subscription is shared if and only if you start observing
|
|
/// the database from the same `SharedValueObservation` instance:
|
|
///
|
|
/// ```swift
|
|
/// // Shared
|
|
/// let sharedObservation = ValueObservation.tracking { db in ... }.shared(in: dbQueue)
|
|
/// let cancellable1 = sharedObservation.start(...)
|
|
/// let cancellable2 = sharedObservation.start(...)
|
|
///
|
|
/// // NOT shared
|
|
/// let cancellable1 = ValueObservation.tracking { db in ... }.shared(in: dbQueue).start(...)
|
|
/// let cancellable2 = ValueObservation.tracking { db in ... }.shared(in: dbQueue).start(...)
|
|
/// ```
|
|
public final class SharedValueObservation<Element> {
|
|
private let scheduler: any ValueObservationScheduler
|
|
private let extent: SharedValueObservationExtent
|
|
private let startObservation: ValueObservationStart<Element>
|
|
private let lock = NSRecursiveLock() // support synchronous observation events
|
|
|
|
// protected by lock
|
|
private var clients: [Client]
|
|
private var isObserving = false
|
|
private var cancellable: AnyDatabaseCancellable?
|
|
private var lastResult: Result<Element, Error>?
|
|
|
|
private final class Client {
|
|
var onError: (Error) -> Void
|
|
var onChange: (Element) -> Void
|
|
|
|
init(onError: @escaping (Error) -> Void, onChange: @escaping (Element) -> Void) {
|
|
self.onError = onError
|
|
self.onChange = onChange
|
|
}
|
|
}
|
|
|
|
fileprivate init(
|
|
scheduling scheduler: some ValueObservationScheduler,
|
|
extent: SharedValueObservationExtent,
|
|
startObservation: @escaping ValueObservationStart<Element>)
|
|
{
|
|
self.scheduler = scheduler
|
|
self.extent = extent
|
|
self.startObservation = startObservation
|
|
self.clients = []
|
|
}
|
|
|
|
/// Starts observing the database.
|
|
///
|
|
/// The observation lasts until the returned cancellable is cancelled
|
|
/// or deallocated.
|
|
///
|
|
/// For example:
|
|
///
|
|
/// ```swift
|
|
/// let sharedObservation = ValueObservation
|
|
/// .tracking { db in try Player.fetchAll(db) }
|
|
/// .shared(in: dbQueue)
|
|
///
|
|
/// let cancellable = try sharedObservation.start { error in
|
|
/// // handle error
|
|
/// } onChange: { (players: [Player]) in
|
|
/// print("fresh players: \(players)")
|
|
/// }
|
|
/// ```
|
|
///
|
|
/// - parameter onError: The closure to execute when the observation fails.
|
|
/// - parameter onChange: The closure to execute on receipt of a
|
|
/// fresh value.
|
|
/// - returns: A DatabaseCancellable that can stop the observation.
|
|
public func start(
|
|
onError: @escaping (Error) -> Void,
|
|
onChange: @escaping (Element) -> Void)
|
|
-> AnyDatabaseCancellable
|
|
{
|
|
synchronized {
|
|
// Support for reentrancy: a shared immediate observation is
|
|
// started from the first value notification of that same shared
|
|
// immediate observation. Yeah, users are nasty.
|
|
// In this case, self.cancellable is still nil, because we are
|
|
// still waiting for the upstream ValueObservation to start.
|
|
// But we must not start another one.
|
|
let needsStart = !isObserving
|
|
|
|
// State change
|
|
let client = Client(onError: onError, onChange: onChange)
|
|
clients.append(client)
|
|
isObserving = true
|
|
|
|
// Side effect
|
|
if needsStart {
|
|
// Self retains the cancellable, so don't have the cancellable retain self.
|
|
cancellable = startObservation(
|
|
// onError
|
|
{ [weak self] error in
|
|
self?.handleError(error)
|
|
},
|
|
// onChange
|
|
{ [weak self] element in
|
|
self?.handleChange(element)
|
|
})
|
|
} else if let result = lastResult {
|
|
// Notify last result as an initial value
|
|
scheduler.scheduleInitial {
|
|
switch result {
|
|
case let .failure(error):
|
|
onError(error)
|
|
case let .success(value):
|
|
onChange(value)
|
|
}
|
|
}
|
|
}
|
|
|
|
return AnyDatabaseCancellable {
|
|
// Retain shared observation (self) until client cancels
|
|
self.handleCancel(client)
|
|
}
|
|
}
|
|
}
|
|
|
|
#if canImport(Combine)
|
|
/// Returns a publisher of observed values.
|
|
///
|
|
/// For example:
|
|
///
|
|
/// ```swift
|
|
/// let observation = ValueObservation
|
|
/// .tracking { db in try Player.fetchAll(db) }
|
|
/// .shared(in: dbQueue)
|
|
///
|
|
/// let publisher = observation.publisher()
|
|
///
|
|
/// let cancellable = publisher.sink { completion in
|
|
/// // handle completion
|
|
/// } receiveValue: { (players: [Player]) in
|
|
/// print("fresh players: \(players)")
|
|
/// }
|
|
/// ```
|
|
@available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
|
|
public func publisher() -> DatabasePublishers.Value<Element> {
|
|
DatabasePublishers.Value { onError, onChange in
|
|
self.start(onError: onError, onChange: onChange)
|
|
}
|
|
}
|
|
#endif
|
|
|
|
private func handleError(_ error: Error) {
|
|
synchronized {
|
|
let notifiedClients = clients
|
|
|
|
// State change
|
|
clients = []
|
|
if extent == .whileObserved {
|
|
isObserving = false
|
|
cancellable = nil
|
|
lastResult = nil
|
|
} else {
|
|
lastResult = .failure(error)
|
|
}
|
|
|
|
// Side effect
|
|
for client in notifiedClients {
|
|
client.onError(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func handleChange(_ value: Element) {
|
|
synchronized {
|
|
// State change
|
|
lastResult = .success(value)
|
|
|
|
// Side effect
|
|
for client in clients {
|
|
client.onChange(value)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func handleCancel(_ client: Client) {
|
|
synchronized {
|
|
// State change
|
|
clients.removeFirst(where: { $0 === client })
|
|
if clients.isEmpty && extent == .whileObserved {
|
|
isObserving = false
|
|
cancellable = nil
|
|
lastResult = nil
|
|
}
|
|
}
|
|
}
|
|
|
|
private func synchronized<T>(_ execute: () throws -> T) rethrows -> T {
|
|
lock.lock()
|
|
defer { lock.unlock() }
|
|
return try execute()
|
|
}
|
|
}
|
|
|
|
extension SharedValueObservation {
|
|
// MARK: - Asynchronous Observation
|
|
/// Returns an asynchronous sequence of observed values.
|
|
///
|
|
/// - note: [**🔥 EXPERIMENTAL**](https://github.com/groue/GRDB.swift/blob/master/README.md#what-are-experimental-features)
|
|
///
|
|
/// For example:
|
|
///
|
|
/// ```swift
|
|
/// let sharedObservation = ValueObservation
|
|
/// .tracking { db in try Player.fetchAll(db) }
|
|
/// .shared(in: dbQueue)
|
|
///
|
|
/// for try await players in sharedObservation.values() {
|
|
/// print("Fresh players: \(players)")
|
|
/// }
|
|
/// ```
|
|
@available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
|
|
public func values(bufferingPolicy: AsyncValueObservation<Element>.BufferingPolicy = .unbounded)
|
|
-> AsyncValueObservation<Element>
|
|
{
|
|
AsyncValueObservation(bufferingPolicy: bufferingPolicy) { onError, onChange in
|
|
self.start(onError: onError, onChange: onChange)
|
|
}
|
|
}
|
|
}
|