Skip to content

Commit 38b0e22

Browse files
wickwirewwickwirew
committed
better observation
1 parent 3626561 commit 38b0e22

7 files changed

Lines changed: 135 additions & 98 deletions

File tree

‎Sources/Feather/ConnectionPool.swift‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -82,10 +82,8 @@ public actor ConnectionPool: Sendable {
8282
}
8383

8484
extension ConnectionPool: Database {
85-
public func observe(
86-
subscriber: any DatabaseSubscriber
87-
) throws(FeatherError) {
88-
try observer.subscribe(subscriber: subscriber)
85+
public nonisolated func observe(subscriber: any DatabaseSubscriber) {
86+
observer.subscribe(subscriber: subscriber)
8987
}
9088

9189
public nonisolated func cancel(subscriber: any DatabaseSubscriber) {

‎Sources/Feather/Database.swift‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
//
77

88
public protocol Database: Actor {
9-
func observe(subscriber: DatabaseSubscriber) throws(FeatherError)
9+
nonisolated func observe(subscriber: DatabaseSubscriber)
1010

1111
nonisolated func cancel(subscriber: DatabaseSubscriber)
1212

‎Sources/Feather/DatabaseObserver.swift‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,14 +25,14 @@ class DatabaseObserver: @unchecked Sendable {
2525
private let lock = NSLock()
2626
private var subscribers: [ObjectIdentifier: any DatabaseSubscriber] = [:]
2727

28-
func subscribe(subscriber: any DatabaseSubscriber) throws(FeatherError) {
28+
func subscribe(subscriber: any DatabaseSubscriber) {
2929
lock.lock()
3030
defer { lock.unlock() }
3131

3232
let id = ObjectIdentifier(subscriber)
3333

3434
guard subscribers[id] == nil else {
35-
throw .subscriptionAlreadyStarted
35+
return
3636
}
3737

3838
subscribers[id] = subscriber
@@ -42,7 +42,6 @@ class DatabaseObserver: @unchecked Sendable {
4242
lock.lock()
4343
defer { lock.unlock() }
4444

45-
subscriber.onCancel()
4645
subscribers[ObjectIdentifier(subscriber)] = nil
4746
}
4847

@@ -55,7 +54,7 @@ class DatabaseObserver: @unchecked Sendable {
5554
}
5655
}
5756

58-
nonisolated func installHooks(into connection: Connection) {
57+
func installHooks(into connection: Connection) {
5958
sqlite3_update_hook(
6059
connection.sqliteConnection,
6160
{ selfPointer, operation, dbName, tableName, rowId in
@@ -74,7 +73,7 @@ class DatabaseObserver: @unchecked Sendable {
7473
)
7574
}
7675

77-
nonisolated func receiveSqliteUpdateHook(
76+
func receiveSqliteUpdateHook(
7877
operation: Int32,
7978
dbName: UnsafePointer<CChar>?,
8079
tableName: UnsafePointer<CChar>?,
@@ -97,5 +96,4 @@ class DatabaseObserver: @unchecked Sendable {
9796

9897
public protocol DatabaseSubscriber: AnyObject {
9998
func receive(event: DatabaseEvent)
100-
func onCancel()
10199
}

‎Sources/Feather/Queries.swift‎

Lines changed: 22 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -23,12 +23,8 @@ public enum Queries {
2323
return try await base.execute(with: input, in: database)
2424
}
2525

26-
public func observe(
27-
with input: Input,
28-
handle: @Sendable @escaping (Output) -> Void,
29-
cancelled: @Sendable @escaping () -> Void
30-
) -> QueryObservation<Input, Output> {
31-
return base.observe(with: input, in: database, handle: handle, cancelled: cancelled)
26+
public func observe(with input: Input) -> any QueryObservation<Output> {
27+
return base.observe(with: input, in: database)
3228
}
3329
}
3430

@@ -59,7 +55,7 @@ public enum Queries {
5955
}
6056

6157
/// Applies a transform to the queries result
62-
public struct Just<Input, Output>: DatabaseQuery
58+
public struct Just<Input, Output>: Query
6359
where Input: Sendable, Output: Sendable
6460
{
6561
let output: Output
@@ -68,22 +64,29 @@ public enum Queries {
6864
self.output = output
6965
}
7066

71-
public var transactionKind: TransactionKind {
72-
return .read
67+
public func execute(with input: Input) async throws -> Output {
68+
return output
7369
}
7470

75-
public func execute(
76-
with input: Input,
77-
in _: ()
78-
) async throws -> Output {
79-
return output
71+
public func observe(with input: Input) -> any QueryObservation<Output> {
72+
return Observation(output: output)
8073
}
8174

82-
public func execute(
83-
with input: Input,
84-
tx: borrowing Transaction
85-
) throws -> Output {
86-
return output
75+
final class Observation: QueryObservation {
76+
let output: Output
77+
78+
init(output: Output) {
79+
self.output = output
80+
}
81+
82+
func start(
83+
onChange: @escaping (Output) -> Void,
84+
onError: @escaping (any Error) -> Void
85+
) {
86+
onChange(output)
87+
}
88+
89+
func cancel() {}
8790
}
8891
}
8992

‎Sources/Feather/Query.swift‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -11,22 +11,11 @@ public protocol Query<Input, Output> {
1111

1212
func execute(with input: Input) async throws -> Output
1313

14-
func observe(
15-
with input: Input,
16-
handle: @Sendable @escaping (Output) -> Void,
17-
cancelled: @Sendable @escaping () -> Void
18-
) -> QueryObservation<Input, Output>
14+
func observe(with input: Input) -> any QueryObservation<Output>
1915
}
2016

2117
public extension Query where Input == () {
2218
func execute() async throws -> Output {
2319
return try await execute(with: ())
2420
}
25-
26-
func observe(
27-
handle: @Sendable @escaping (Output) -> Void,
28-
cancelled: @Sendable @escaping () -> Void
29-
) -> QueryObservation<Input, Output> {
30-
return observe(with: (), handle: handle, cancelled: cancelled)
31-
}
3221
}

‎Sources/Feather/QueryObservation.swift‎

Lines changed: 101 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -5,50 +5,131 @@
55
// Created by Wes Wickwire on 3/10/25.
66
//
77

8-
public final class QueryObservation<Input, Output>: DatabaseSubscriber, Sendable
8+
import Foundation
9+
10+
public final class DatabaseQueryObservation<Input, Output>: DatabaseSubscriber, QueryObservation, @unchecked Sendable
911
where Input: Sendable, Output: Sendable
1012
{
1113
private let query: any DatabaseQuery<Input, Output>
1214
private let input: Input
1315
private let database: any Database
14-
private let handle: @Sendable (Output) -> Void
15-
private let cancelled: @Sendable () -> Void
16+
private let lock = NSLock()
17+
private let queue = Queue()
18+
19+
private var onChange: (@Sendable (Output) -> Void)?
20+
private var onError: (@Sendable (Error) -> Void)?
1621

1722
init(
1823
query: any DatabaseQuery<Input, Output>,
1924
input: Input,
20-
database: any Database,
21-
handle: @Sendable @escaping (Output) -> Void,
22-
cancelled: @Sendable @escaping () -> Void
25+
database: any Database
2326
) {
2427
self.query = query
2528
self.input = input
2629
self.database = database
27-
self.handle = handle
28-
self.cancelled = cancelled
2930
}
3031

3132
public func receive(event: DatabaseEvent) {
32-
Task {
33-
try await handle(query.execute(with: input, in: database))
33+
enqueueNext()
34+
}
35+
36+
public func cancel() {
37+
lock.withLock {
38+
onChange = nil
39+
onError = nil
40+
}
41+
42+
database.cancel(subscriber: self)
43+
}
44+
45+
public func start(
46+
onChange: @escaping @Sendable (Output) -> Void,
47+
onError: @escaping @Sendable (Error) -> Void
48+
) {
49+
lock.withLock {
50+
self.onChange = onChange
51+
self.onError = onError
3452
}
53+
54+
database.observe(subscriber: self)
55+
enqueueNext()
3556
}
3657

37-
public func onCancel() {
38-
cancelled()
58+
private func emitNext() async {
59+
guard let onChange else {
60+
return assertionFailure("Started without handle set")
61+
}
62+
63+
do {
64+
let output = try await query.execute(with: input, in: database)
65+
onChange(output)
66+
} catch {
67+
onError?(error)
68+
cancel()
69+
}
3970
}
4071

41-
public func cancel() {
42-
database.cancel(subscriber: self)
72+
private func enqueueNext() {
73+
queue.enqueue { [weak self] in
74+
await self?.emitNext()
75+
}
76+
}
77+
}
78+
79+
public protocol QueryObservation<Output>: Sendable, AsyncSequence {
80+
associatedtype Output: Sendable
81+
82+
func start(
83+
onChange: @escaping @Sendable (Output) -> Void,
84+
onError: @escaping @Sendable (Error) -> Void
85+
)
86+
87+
func cancel()
88+
}
89+
90+
extension QueryObservation {
91+
public func makeAsyncIterator() -> AsyncThrowingStream<Output, Error>.AsyncIterator {
92+
return asStream().makeAsyncIterator()
93+
}
94+
95+
func asStream() -> AsyncThrowingStream<Output, Error> {
96+
AsyncThrowingStream<Output, Error> { continuation in
97+
start { output in
98+
continuation.yield(output)
99+
} onError: { error in
100+
continuation.finish(throwing: error)
101+
}
102+
103+
continuation.onTermination = { _ in
104+
cancel()
105+
}
106+
}
107+
}
108+
}
109+
110+
final class Queue: Sendable {
111+
typealias Action = @Sendable () async -> Void
112+
113+
private let task: Task<(), Never>
114+
private let stream: AsyncStream<Action>
115+
private let continuation: AsyncStream<Action>.Continuation
116+
117+
init() {
118+
let (stream, continuation) = AsyncStream<Action>.makeStream()
119+
self.stream = stream
120+
self.continuation = continuation
121+
self.task = Task {
122+
for await action in stream {
123+
await action()
124+
}
125+
}
43126
}
44127

45-
public func start() async throws {
46-
try await database.observe(subscriber: self)
47-
try await emitNext()
128+
deinit {
129+
task.cancel()
48130
}
49131

50-
private func emitNext() async throws {
51-
let output = try await query.execute(with: input, in: (database))
52-
handle(output)
132+
func enqueue(_ action: @escaping Action) {
133+
continuation.yield(action)
53134
}
54135
}

‎Sources/Feather/Queryable.swift‎

Lines changed: 4 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -34,38 +34,14 @@ public extension DatabaseQuery {
3434

3535
func observe(
3636
with input: Input,
37-
in database: any Database,
38-
handle: @Sendable @escaping (Output) -> Void,
39-
cancelled: @Sendable @escaping () -> Void
40-
) -> QueryObservation<Input, Output> {
41-
return QueryObservation(
37+
in database: any Database
38+
) -> any QueryObservation<Output> {
39+
return DatabaseQueryObservation(
4240
query: self,
4341
input: input,
44-
database: database,
45-
handle: handle,
46-
cancelled: cancelled
42+
database: database
4743
)
4844
}
49-
50-
func stream(
51-
with input: Input,
52-
in database: any Database
53-
) -> AsyncThrowingStream<Output, Error> {
54-
return AsyncThrowingStream<Output, Error> { continuation in
55-
let observation = self.observe(
56-
with: input,
57-
in: database
58-
) { output in
59-
continuation.yield(output)
60-
} cancelled: {
61-
// Nothing to do
62-
}
63-
64-
continuation.onTermination = { _ in
65-
observation.cancel()
66-
}
67-
}
68-
}
6945
}
7046

7147
extension DatabaseQuery where Input == () {
@@ -76,12 +52,4 @@ extension DatabaseQuery where Input == () {
7652
func execute(tx: borrowing Transaction) throws -> Output {
7753
return try execute(with: (), tx: tx)
7854
}
79-
80-
func observe(
81-
in database: any Database,
82-
handle: @Sendable @escaping (Output) -> Void,
83-
cancelled: @Sendable @escaping () -> Void
84-
) -> QueryObservation<Input, Output> {
85-
return observe(with: (), in: database, handle: handle, cancelled: cancelled)
86-
}
8755
}

0 commit comments

Comments
 (0)