Skip to content

Commit 3cc0942

Browse files
apackinapackin
committed
fix(graphql): deduplicate pollers
Prevents multiple of the same query from being added to intervalQueries, which dictates which queries are being polled.
1 parent 312dc6c commit 3cc0942

File tree

3 files changed

+164
-3
lines changed

3 files changed

+164
-3
lines changed

‎packages/graphql/lib/src/cache/cache.dart‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ class GraphQLCache extends NormalizingDataProxy {
138138
/// Write normalized data into the cache,
139139
/// deeply merging maps with existing values
140140
///
141-
/// Called from [witeQuery] and [writeFragment].
141+
/// Called from [writeQuery] and [writeFragment].
142142
void writeNormalized(String dataId, Map<String, dynamic>? value) {
143143
if (value is Map<String, Object>) {
144144
final existing = store.get(dataId);

‎packages/graphql/lib/src/scheduler/scheduler.dart‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ class QueryScheduler {
1818

1919
/// Map going from poling interval to the query ids that fire on that interval.
2020
/// These query ids are associated with a [ObservableQuery] in the registeredQueries.
21-
Map<Duration?, List<String>> intervalQueries = <Duration?, List<String>>{};
21+
Map<Duration?, Set<String>> intervalQueries = <Duration?, Set<String>>{};
2222

2323
/// Map going from polling interval durations to polling timers.
2424
final Map<Duration?, Timer> _pollingTimers = <Duration?, Timer>{};
@@ -75,7 +75,7 @@ class QueryScheduler {
7575
if (intervalQueries.containsKey(interval)) {
7676
intervalQueries[interval]!.add(queryId);
7777
} else {
78-
intervalQueries[interval] = <String>[queryId];
78+
intervalQueries[interval] = Set<String>.of([queryId]);
7979

8080
_pollingTimers[interval] = Timer.periodic(
8181
interval!,

‎packages/graphql/test/observable_query_test.dart‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
import 'package:test/test.dart';
2+
import 'package:mockito/mockito.dart';
3+
4+
import 'package:graphql/client.dart';
5+
import 'package:gql/language.dart';
6+
7+
import './helpers.dart';
8+
9+
void main() {
10+
const String readSingle = r'''
11+
query ReadSingle() {
12+
single() {
13+
id,
14+
__typename,
15+
name
16+
}
17+
}
18+
''';
19+
const data = {
20+
'single': {
21+
'id': '1',
22+
'__typename': 'Single',
23+
'name': 'initialQueryName',
24+
},
25+
};
26+
const pollDuration = Duration(milliseconds: 20);
27+
28+
final queryResponseMatcher = isA<QueryResult>().having(
29+
(result) => result.data!['single']['name'],
30+
'name',
31+
'initialQueryName',
32+
);
33+
34+
late MockLink link;
35+
late GraphQLClient client;
36+
37+
group('observable query ', () {
38+
setUp(() {
39+
link = MockLink();
40+
41+
client = GraphQLClient(
42+
cache: getTestCache(),
43+
link: link,
44+
);
45+
46+
final queryResponse = Response(data: data, response: {});
47+
when(
48+
link.request(any),
49+
).thenAnswer(
50+
(_) => Stream.fromIterable(
51+
[queryResponse],
52+
),
53+
);
54+
});
55+
test('can start poller', () async {
56+
final observable = client.watchQuery(
57+
WatchQueryOptions(
58+
document: parseString(readSingle),
59+
),
60+
);
61+
expect(
62+
observable.stream,
63+
emitsInOrder(
64+
[queryResponseMatcher, queryResponseMatcher, emitsDone],
65+
),
66+
);
67+
observable.startPolling(pollDuration);
68+
await Future<void>.delayed(pollDuration * 2.1);
69+
observable.close();
70+
}, timeout: Timeout(Duration(seconds: 1)));
71+
test('can stop poller', () async {
72+
final observable = client.watchQuery(
73+
WatchQueryOptions(
74+
document: parseString(readSingle),
75+
),
76+
);
77+
expect(
78+
observable.stream,
79+
neverEmits(
80+
isA<QueryResult>().having(
81+
(result) => result.data!['single']['name'],
82+
'name',
83+
'initialQueryName',
84+
),
85+
),
86+
);
87+
observable.startPolling(pollDuration);
88+
observable.stopPolling();
89+
await Future<void>.delayed(pollDuration * 2.1);
90+
observable.close();
91+
}, timeout: Timeout(Duration(seconds: 1)));
92+
test('can deduplicate startPolling calls with the same duration', () async {
93+
final observable = client.watchQuery(
94+
WatchQueryOptions(
95+
document: parseString(readSingle),
96+
),
97+
);
98+
expect(
99+
observable.stream,
100+
emitsInOrder(
101+
[queryResponseMatcher, queryResponseMatcher, emitsDone],
102+
),
103+
);
104+
observable.startPolling(pollDuration);
105+
observable.startPolling(pollDuration);
106+
observable.startPolling(pollDuration);
107+
await Future<void>.delayed(pollDuration * 2.1);
108+
observable.close();
109+
}, timeout: Timeout(Duration(seconds: 1)));
110+
test('can deduplicate startPolling calls with different durations',
111+
() async {
112+
final observable = client.watchQuery(
113+
WatchQueryOptions(
114+
document: parseString(readSingle),
115+
),
116+
);
117+
expect(
118+
observable.stream,
119+
emitsInOrder(
120+
[
121+
queryResponseMatcher,
122+
queryResponseMatcher,
123+
queryResponseMatcher,
124+
emitsDone
125+
],
126+
),
127+
);
128+
observable.startPolling(Duration(milliseconds: 10));
129+
observable.startPolling(Duration(milliseconds: 20));
130+
observable.startPolling(Duration(milliseconds: 30));
131+
observable.startPolling(pollDuration);
132+
await Future<void>.delayed(pollDuration * 3.1);
133+
observable.close();
134+
}, timeout: Timeout(Duration(seconds: 1)));
135+
test('can stop pollers in quick succession', () async {
136+
final observable = client.watchQuery(
137+
WatchQueryOptions(
138+
document: parseString(readSingle),
139+
),
140+
);
141+
expect(
142+
observable.stream,
143+
emitsInOrder(
144+
[
145+
queryResponseMatcher,
146+
queryResponseMatcher,
147+
queryResponseMatcher,
148+
emitsDone
149+
],
150+
),
151+
);
152+
observable.startPolling(pollDuration);
153+
observable.stopPolling();
154+
observable.startPolling(pollDuration);
155+
observable.stopPolling();
156+
observable.startPolling(pollDuration);
157+
await Future<void>.delayed(pollDuration * 3.1);
158+
observable.close();
159+
}, timeout: Timeout(Duration(seconds: 1)));
160+
});
161+
}

0 commit comments

Comments
 (0)