Skip to content

Commit 238853b

Browse files
simolus3simolus3
authored
Make watch queries easier to cancel (#92)
* Refactor watch queries to allow cancelling them * Fix handling pause immediately after emit * Review feedback
1 parent ae7dc0e commit 238853b

File tree

4 files changed

+187
-71
lines changed

4 files changed

+187
-71
lines changed

‎packages/sqlite_async/lib/src/sqlite_queries.dart‎

Lines changed: 14 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -44,25 +44,23 @@ mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
4444
Stream<sqlite.ResultSet> watch(String sql,
4545
{List<Object?> parameters = const [],
4646
Duration throttle = const Duration(milliseconds: 30),
47-
Iterable<String>? triggerOnTables}) async* {
47+
Iterable<String>? triggerOnTables}) {
4848
assert(updates != null,
4949
'updates stream must be provided to allow query watching');
50-
final tables =
51-
triggerOnTables ?? await getSourceTables(this, sql, parameters);
52-
final filteredStream =
53-
updates!.transform(UpdateNotification.filterTablesTransformer(tables));
54-
final throttledStream = UpdateNotification.throttleStream(
55-
filteredStream, throttle,
56-
addOne: UpdateNotification.empty());
5750

58-
// FIXME:
59-
// When the subscription is cancelled, this performs a final query on the next
60-
// update.
61-
// The loop only stops once the "yield" is reached.
62-
// Using asyncMap instead of a generator would solve it, but then the body
63-
// here can't be async for getSourceTables().
64-
await for (var _ in throttledStream) {
65-
yield await getAll(sql, parameters);
51+
Stream<sqlite.ResultSet> watchInner(Iterable<String> trigger) {
52+
return onChange(
53+
trigger,
54+
throttle: throttle,
55+
triggerImmediately: true,
56+
).asyncMap((_) => getAll(sql, parameters));
57+
}
58+
59+
if (triggerOnTables case final knownTrigger?) {
60+
return watchInner(knownTrigger);
61+
} else {
62+
return Stream.fromFuture(getSourceTables(this, sql, parameters))
63+
.asyncExpand(watchInner);
6664
}
6765
}
6866

‎packages/sqlite_async/lib/src/update_notification.dart‎

Lines changed: 113 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -52,10 +52,13 @@ class UpdateNotification {
5252
static Stream<UpdateNotification> throttleStream(
5353
Stream<UpdateNotification> input, Duration timeout,
5454
{UpdateNotification? addOne}) {
55-
return _throttleStream(input, timeout, addOne: addOne, throttleFirst: true,
56-
add: (a, b) {
57-
return a.union(b);
58-
});
55+
return _throttleStream(
56+
input: input,
57+
timeout: timeout,
58+
throttleFirst: true,
59+
add: (a, b) => a.union(b),
60+
addOne: addOne,
61+
);
5962
}
6063

6164
/// Filter an update stream by specific tables.
@@ -67,62 +70,120 @@ class UpdateNotification {
6770
}
6871
}
6972

70-
/// Given a broadcast stream, return a singular throttled stream that is throttled.
71-
/// This immediately starts listening.
73+
/// Throttles an [input] stream to not emit events more often than with a
74+
/// frequency of 1/[timeout].
7275
///
73-
/// Behaviour:
74-
/// If there was no event in "timeout", and one comes in, it is pushed immediately.
75-
/// Otherwise, we wait until the timeout is over.
76-
Stream<T> _throttleStream<T extends Object>(Stream<T> input, Duration timeout,
77-
{bool throttleFirst = false, T Function(T, T)? add, T? addOne}) async* {
78-
var nextPing = Completer<void>();
79-
var done = false;
80-
T? lastData;
81-
82-
var listener = input.listen((data) {
83-
if (lastData != null && add != null) {
84-
lastData = add(lastData!, data);
85-
} else {
86-
lastData = data;
76+
/// When an event is received and no timeout window is active, it is forwarded
77+
/// downstream and a timeout window is started. For events received within a
78+
/// timeout window, [add] is called to fold events. Then when the window
79+
/// expires, pending events are emitted.
80+
/// The subscription to the [input] stream is never paused.
81+
///
82+
/// When the returned stream is paused, an active timeout window is reset and
83+
/// restarts after the stream is resumed.
84+
///
85+
/// If [addOne] is not null, that event will always be added when the stream is
86+
/// subscribed to.
87+
/// When [throttleFirst] is true, a timeout window begins immediately after
88+
/// listening (so that the first event, apart from [addOne], is emitted no
89+
/// earlier than after [timeout]).
90+
Stream<T> _throttleStream<T extends Object>({
91+
required Stream<T> input,
92+
required Duration timeout,
93+
required bool throttleFirst,
94+
required T Function(T, T) add,
95+
required T? addOne,
96+
}) {
97+
return Stream.multi((listener) {
98+
T? pendingData;
99+
Timer? activeTimeoutWindow;
100+
var needsTimeoutWindowAfterResume = false;
101+
102+
/// Add pending data, bypassing the active timeout window.
103+
///
104+
/// This is used to forward error and done events immediately.
105+
bool addPendingEvents() {
106+
if (pendingData case final data?) {
107+
pendingData = null;
108+
listener.addSync(data);
109+
activeTimeoutWindow?.cancel();
110+
activeTimeoutWindow = null;
111+
return true;
112+
} else {
113+
return false;
114+
}
87115
}
88-
if (!nextPing.isCompleted) {
89-
nextPing.complete();
116+
117+
late void Function() setTimeout;
118+
119+
/// Emits [pendingData] if no timeout window is active, and then starts a
120+
/// timeout window if necessary.
121+
void maybeEmit() {
122+
if (activeTimeoutWindow == null && !listener.isPaused) {
123+
final didAdd = addPendingEvents();
124+
if (didAdd) {
125+
// Schedule a pause after resume if the subscription was paused
126+
// directly in response to receiving the event. Otherwise, begin the
127+
// timeout window immediately.
128+
if (listener.isPaused) {
129+
needsTimeoutWindowAfterResume = true;
130+
} else {
131+
setTimeout();
132+
}
133+
}
134+
}
90135
}
91-
}, onDone: () {
92-
if (!nextPing.isCompleted) {
93-
nextPing.complete();
136+
137+
setTimeout = () {
138+
activeTimeoutWindow = Timer(timeout, () {
139+
activeTimeoutWindow = null;
140+
maybeEmit();
141+
});
142+
};
143+
144+
void onData(T data) {
145+
pendingData = switch (pendingData) {
146+
null => data,
147+
final pending => add(pending, data),
148+
};
149+
maybeEmit();
94150
}
95151

96-
done = true;
97-
});
152+
void onError(Object error, StackTrace trace) {
153+
addPendingEvents();
154+
listener.addErrorSync(error, trace);
155+
}
156+
157+
void onDone() {
158+
addPendingEvents();
159+
listener.closeSync();
160+
}
161+
162+
final subscription = input.listen(onData, onError: onError, onDone: onDone);
163+
164+
listener.onPause = () {
165+
needsTimeoutWindowAfterResume = activeTimeoutWindow != null;
166+
activeTimeoutWindow?.cancel();
167+
activeTimeoutWindow = null;
168+
};
169+
listener.onResume = () {
170+
if (needsTimeoutWindowAfterResume) {
171+
setTimeout();
172+
} else {
173+
maybeEmit();
174+
}
175+
};
176+
listener.onCancel = () async {
177+
activeTimeoutWindow?.cancel();
178+
return subscription.cancel();
179+
};
98180

99-
try {
100181
if (addOne != null) {
101-
yield addOne;
182+
// This must not be sync, we're doing this directly in onListen
183+
listener.add(addOne);
102184
}
103185
if (throttleFirst) {
104-
await Future.delayed(timeout);
105-
}
106-
while (!done) {
107-
// If a value is available now, we'll use it immediately.
108-
// If not, this waits for it.
109-
await nextPing.future;
110-
if (done) break;
111-
112-
// Capture any new values coming in while we wait.
113-
nextPing = Completer<void>();
114-
T data = lastData as T;
115-
// Clear before we yield, so that we capture new changes while yielding
116-
lastData = null;
117-
yield data;
118-
// Wait a minimum of this duration between tasks
119-
await Future.delayed(timeout);
186+
setTimeout();
120187
}
121-
} finally {
122-
if (lastData case final data?) {
123-
yield data;
124-
}
125-
126-
await listener.cancel();
127-
}
188+
});
128189
}

‎packages/sqlite_async/test/update_notification_test.dart‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,61 @@ void main() {
4747
});
4848
});
4949

50+
test('increases delay after pause', () {
51+
fakeAsync((control) {
52+
final source = StreamController<UpdateNotification>(sync: true);
53+
final events = <UpdateNotification>[];
54+
55+
final sub = UpdateNotification.throttleStream(source.stream, timeout)
56+
.listen(null);
57+
sub.onData((event) {
58+
events.add(event);
59+
sub.pause();
60+
});
61+
62+
source.add(UpdateNotification({'a'}));
63+
control.elapse(timeout);
64+
expect(events, hasLength(1));
65+
66+
// Assume the stream stays paused for the timeout window that would
67+
// be created after emitting the notification.
68+
control.elapse(timeout * 2);
69+
source.add(UpdateNotification({'b'}));
70+
control.elapse(timeout * 2);
71+
72+
// A full timeout needs to pass after resuming before a new item is
73+
// emitted.
74+
sub.resume();
75+
expect(events, hasLength(1));
76+
77+
control.elapse(halfTimeout);
78+
expect(events, hasLength(1));
79+
control.elapse(halfTimeout);
80+
expect(events, hasLength(2));
81+
});
82+
});
83+
84+
test('does not introduce artificial delay in pause', () {
85+
fakeAsync((control) {
86+
final source = StreamController<UpdateNotification>(sync: true);
87+
final events = <UpdateNotification>[];
88+
89+
final sub = UpdateNotification.throttleStream(source.stream, timeout)
90+
.listen(events.add);
91+
92+
// Await the initial delay
93+
control.elapse(timeout);
94+
95+
sub.pause();
96+
source.add(UpdateNotification({'a'}));
97+
// Resuming should not introduce a timeout window because no window
98+
// was active when the stream was paused.
99+
sub.resume();
100+
control.flushMicrotasks();
101+
expect(events, hasLength(1));
102+
});
103+
});
104+
50105
test('merges events', () {
51106
fakeAsync((control) {
52107
final source = StreamController<UpdateNotification>(sync: true);

‎packages/sqlite_async/test/watch_test.dart‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,10 @@ void main() {
113113
lastCount = count;
114114
}
115115

116-
// The number of read queries must not be greater than the number of writes overall.
117-
expect(numberOfQueries, lessThanOrEqualTo(results.last.first['count']));
116+
// The number of read queries must not be greater than the number of
117+
// writes overall, plus one for the initial stream emission.
118+
expect(numberOfQueries,
119+
lessThanOrEqualTo(results.last.first['count'] + 1));
118120

119121
DateTime? lastTime;
120122
for (var r in times) {
@@ -283,7 +285,7 @@ void main() {
283285
});
284286
await Future.delayed(delay);
285287

286-
subscription.cancel();
288+
await subscription.cancel();
287289

288290
expect(
289291
counts,

0 commit comments

Comments
 (0)