Add CI/CD configuration and API documentation
This commit is contained in:
@@ -0,0 +1,316 @@
|
||||
#if canImport(Combine)
|
||||
import Combine
|
||||
#endif
|
||||
import Foundation
|
||||
|
||||
public struct DatabaseRegionObservation {
|
||||
/// A closure that is evaluated when the observation starts, and returns
|
||||
/// the observed database region.
|
||||
var observedRegion: (Database) throws -> DatabaseRegion
|
||||
}
|
||||
|
||||
extension DatabaseRegionObservation {
|
||||
/// Creates a `DatabaseRegionObservation` that notifies all transactions
|
||||
/// that modify one of the provided regions.
|
||||
///
|
||||
/// For example:
|
||||
///
|
||||
/// ```swift
|
||||
/// // An observation that tracks the 'player' table
|
||||
/// let observation = DatabaseRegionObservation(tracking: Player.all())
|
||||
/// ```
|
||||
///
|
||||
/// - parameter regions: A list of observed regions.
|
||||
public init(tracking regions: any DatabaseRegionConvertible...) {
|
||||
self.init(tracking: regions)
|
||||
}
|
||||
|
||||
/// Creates a `DatabaseRegionObservation` that notifies all transactions
|
||||
/// that modify one of the provided regions.
|
||||
///
|
||||
/// For example:
|
||||
///
|
||||
/// ```swift
|
||||
/// // An observation that tracks the 'player' table
|
||||
/// let observation = DatabaseRegionObservation(tracking: [Player.all()])
|
||||
/// ```
|
||||
///
|
||||
/// - parameter regions: An array of observed regions.
|
||||
public init(tracking regions: [any DatabaseRegionConvertible]) {
|
||||
self.init(observedRegion: DatabaseRegion.union(regions))
|
||||
}
|
||||
}
|
||||
|
||||
extension DatabaseRegionObservation {
|
||||
/// The state of a started DatabaseRegionObservation
|
||||
private enum ObservationState {
|
||||
case cancelled
|
||||
case pending
|
||||
case started(DatabaseRegionObserver)
|
||||
}
|
||||
|
||||
/// Starts observing the database.
|
||||
///
|
||||
/// The observation lasts until the returned cancellable is cancelled
|
||||
/// or deallocated.
|
||||
///
|
||||
/// For example:
|
||||
///
|
||||
/// ```swift
|
||||
/// let observation = DatabaseRegionObservation(tracking: Player.all())
|
||||
///
|
||||
/// let cancellable = try observation.start(in: dbQueue) { error in
|
||||
/// // handle error
|
||||
/// } onChange: { (db: Database) in
|
||||
/// print("A modification of the player table has just been committed.")
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// If this method is called from the writer dispatch queue of `writer` (see
|
||||
/// ``DatabaseWriter``), the observation starts immediately. Otherwise, it
|
||||
/// blocks the current thread until a write access can be established.
|
||||
///
|
||||
/// Both `onError` and `onChange` closures are executed in the writer
|
||||
/// dispatch queue, serialized with all database updates performed
|
||||
/// by `writer`.
|
||||
///
|
||||
/// The ``Database`` argument to `onChange` is valid only during the
|
||||
/// execution of the closure. Do not store or return the database connection
|
||||
/// for later use.
|
||||
///
|
||||
/// - parameter writer: A DatabaseWriter.
|
||||
/// - parameter onError: The closure to execute when the observation fails.
|
||||
/// - parameter onChange: The closure to execute when a transaction has
|
||||
/// modified the observed region.
|
||||
/// - returns: A DatabaseCancellable that can stop the observation.
|
||||
public func start(
|
||||
in writer: some DatabaseWriter,
|
||||
onError: @escaping (Error) -> Void,
|
||||
onChange: @escaping (Database) -> Void)
|
||||
-> AnyDatabaseCancellable
|
||||
{
|
||||
@LockedBox var state = ObservationState.pending
|
||||
|
||||
// Use unsafeReentrantWrite so that observation can start from any
|
||||
// dispatch queue.
|
||||
writer.unsafeReentrantWrite { db in
|
||||
do {
|
||||
let region = try observedRegion(db).observableRegion(db)
|
||||
$state.update {
|
||||
let observer = DatabaseRegionObserver(region: region, onChange: {
|
||||
if case .cancelled = state {
|
||||
return
|
||||
}
|
||||
onChange($0)
|
||||
})
|
||||
|
||||
// Use the `.observerLifetime` extent so that we can cancel
|
||||
// the observation by deallocating the observer. This is
|
||||
// a simpler way to cancel the observation than waiting for
|
||||
// *another* write access in order to explicitly remove
|
||||
// the observer.
|
||||
db.add(transactionObserver: observer, extent: .observerLifetime)
|
||||
|
||||
$0 = .started(observer)
|
||||
}
|
||||
} catch {
|
||||
onError(error)
|
||||
}
|
||||
}
|
||||
|
||||
return AnyDatabaseCancellable {
|
||||
// Deallocates the transaction observer. This makes sure that the
|
||||
// `onChange` callback will never be called again, because the
|
||||
// observation was started with the `.observerLifetime` extent.
|
||||
state = .cancelled
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#if canImport(Combine)
|
||||
@available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
|
||||
extension DatabaseRegionObservation {
|
||||
// MARK: - Publishing Impactful Transactions
|
||||
|
||||
/// Returns a publisher that observes the database.
|
||||
///
|
||||
/// The publisher publishes ``Database`` connections on the writer dispatch
|
||||
/// queue of `writer` (see ``DatabaseWriter``). Those connections are valid
|
||||
/// only when published. Do not store or return them for later use.
|
||||
///
|
||||
/// Do not reschedule the publisher with `receive(on:options:)` or any
|
||||
/// `Publisher` method that schedules publisher elements.
|
||||
@available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
|
||||
public func publisher(in writer: some DatabaseWriter) -> DatabasePublishers.DatabaseRegion {
|
||||
DatabasePublishers.DatabaseRegion(self, in: writer)
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
private class DatabaseRegionObserver: TransactionObserver {
|
||||
let region: DatabaseRegion
|
||||
let onChange: (Database) -> Void
|
||||
var isChanged = false
|
||||
|
||||
init(region: DatabaseRegion, onChange: @escaping (Database) -> Void) {
|
||||
self.region = region
|
||||
self.onChange = onChange
|
||||
}
|
||||
|
||||
func observes(eventsOfKind eventKind: DatabaseEventKind) -> Bool {
|
||||
region.isModified(byEventsOfKind: eventKind)
|
||||
}
|
||||
|
||||
func databaseDidChange() {
|
||||
isChanged = true
|
||||
stopObservingDatabaseChangesUntilNextTransaction()
|
||||
}
|
||||
|
||||
func databaseDidChange(with event: DatabaseEvent) {
|
||||
if region.isModified(by: event) {
|
||||
isChanged = true
|
||||
stopObservingDatabaseChangesUntilNextTransaction()
|
||||
}
|
||||
}
|
||||
|
||||
func databaseDidCommit(_ db: Database) {
|
||||
guard isChanged else { return }
|
||||
isChanged = false
|
||||
|
||||
onChange(db)
|
||||
}
|
||||
|
||||
func databaseDidRollback(_ db: Database) {
|
||||
isChanged = false
|
||||
}
|
||||
}
|
||||
|
||||
#if canImport(Combine)
|
||||
@available(iOS 13, macOS 10.15, tvOS 13, watchOS 6, *)
|
||||
extension DatabasePublishers {
|
||||
/// A publisher that tracks transactions that modify a database region.
|
||||
///
|
||||
/// You build such a publisher from ``DatabaseRegionObservation``.
|
||||
public struct DatabaseRegion: Publisher {
|
||||
public typealias Output = Database
|
||||
public typealias Failure = Error
|
||||
|
||||
let writer: any DatabaseWriter
|
||||
let observation: DatabaseRegionObservation
|
||||
|
||||
init(_ observation: DatabaseRegionObservation, in writer: some DatabaseWriter) {
|
||||
self.writer = writer
|
||||
self.observation = observation
|
||||
}
|
||||
|
||||
public func receive<S>(subscriber: S) where S: Subscriber, Failure == S.Failure, Output == S.Input {
|
||||
let subscription = DatabaseRegionSubscription(
|
||||
writer: writer,
|
||||
observation: observation,
|
||||
downstream: subscriber)
|
||||
subscriber.receive(subscription: subscription)
|
||||
}
|
||||
}
|
||||
|
||||
private class DatabaseRegionSubscription<Downstream: Subscriber>: Subscription
|
||||
where Downstream.Failure == Error, Downstream.Input == Database
|
||||
{
|
||||
private struct WaitingForDemand {
|
||||
let downstream: Downstream
|
||||
let writer: any DatabaseWriter
|
||||
let observation: DatabaseRegionObservation
|
||||
}
|
||||
|
||||
private struct Observing {
|
||||
let downstream: Downstream
|
||||
let writer: any DatabaseWriter // Retain writer until subscription is finished
|
||||
var remainingDemand: Subscribers.Demand
|
||||
}
|
||||
|
||||
private enum State {
|
||||
// Waiting for demand, not observing the database.
|
||||
case waitingForDemand(WaitingForDemand)
|
||||
|
||||
// Observing the database.
|
||||
case observing(Observing)
|
||||
|
||||
// Completed or cancelled, not observing the database.
|
||||
case finished
|
||||
}
|
||||
|
||||
// cancellable is not stored in self.state because we must enter the
|
||||
// .observing state *before* the observation starts.
|
||||
private var cancellable: AnyDatabaseCancellable?
|
||||
private var state: State
|
||||
private var lock = NSRecursiveLock() // Allow re-entrancy
|
||||
|
||||
init(
|
||||
writer: some DatabaseWriter,
|
||||
observation: DatabaseRegionObservation,
|
||||
downstream: Downstream)
|
||||
{
|
||||
state = .waitingForDemand(WaitingForDemand(
|
||||
downstream: downstream,
|
||||
writer: writer,
|
||||
observation: observation))
|
||||
}
|
||||
|
||||
func request(_ demand: Subscribers.Demand) {
|
||||
lock.synchronized {
|
||||
switch state {
|
||||
case let .waitingForDemand(info):
|
||||
guard demand > 0 else {
|
||||
return
|
||||
}
|
||||
state = .observing(Observing(
|
||||
downstream: info.downstream,
|
||||
writer: info.writer,
|
||||
remainingDemand: demand))
|
||||
cancellable = info.observation.start(
|
||||
in: info.writer,
|
||||
onError: { [weak self] in self?.receive(failure: $0) },
|
||||
onChange: { [weak self] in self?.receive($0) })
|
||||
|
||||
case var .observing(info):
|
||||
info.remainingDemand += demand
|
||||
state = .observing(info)
|
||||
|
||||
case .finished:
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func cancel() {
|
||||
lock.synchronized {
|
||||
cancellable = nil
|
||||
state = .finished
|
||||
}
|
||||
}
|
||||
|
||||
private func receive(_ value: Database) {
|
||||
lock.synchronized {
|
||||
if case let .observing(info) = state,
|
||||
info.remainingDemand > .none
|
||||
{
|
||||
let additionalDemand = info.downstream.receive(value)
|
||||
if case var .observing(info) = state {
|
||||
info.remainingDemand += additionalDemand
|
||||
info.remainingDemand -= 1
|
||||
state = .observing(info)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func receive(failure error: Error) {
|
||||
lock.synchronized {
|
||||
if case let .observing(info) = state {
|
||||
state = .finished
|
||||
info.downstream.receive(completion: .failure(error))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
Reference in New Issue
Block a user