fix: Stream broadcast
MODIFY
packages/graphql/lib/src/core/query_manager.dart
packages/graphql/lib/src/core/query_manager.dart
@@ -92,7 +92,9 @@ class QueryManager {
}
try {
- yield* link.request(request).map((response) {
+ yield* link
+ .request(request)
+ .map((response) {
QueryResult<TParsed>? queryResult;
bool rereadFromCache = false;
try {
@@ -127,13 +129,18 @@ class QueryManager {
}
return queryResult;
- }).transform(StreamTransformer.fromHandlers(
+ })
+ .transform<QueryResult<TParsed>>(StreamTransformer.fromHandlers(
handleError: (err, trace, sink) => sink.add(_wrapFailure(
err,
trace,
options.parserFn,
)),
- ));
+ ))
+ .map((QueryResult<TParsed> queryResult) {
+ maybeRebroadcastQueries();
+ return queryResult;
+ });
} catch (ex, trace) {
yield* Stream.fromIterable([
_wrapFailure(

MODIFY
packages/graphql/test/graphql_client_test.dart
packages/graphql/test/graphql_client_test.dart
@@ -657,29 +657,109 @@ void main() {
),
);
});
- test('parses results', () async {
- final responses = [
- {
- 'id': '1',
- 'name': 'first',
- },
- {
- 'id': '2',
- 'name': 'second',
- },
- ].map((item) => Response(
+ test('broadcasts', () async {
+ const initialName = 'initial';
+ const firstUpdateName = 'first';
+ const secondUpdateName = 'second';
+ final initialQueryResponse = Response(
data: <String, dynamic>{
- 'item': {
+ 'single': {
+ 'id': '1',
'__typename': 'Item',
- ...item,
+ 'name': initialName,
},
},
- ));
+ );
+ when(
+ link.request(any),
+ ).thenAnswer(
+ (_) => Stream.fromIterable(
+ [initialQueryResponse],
+ ),
+ );
+
+ final ObservableQuery observable = client.watchQuery(
+ WatchQueryOptions(
+ document: parseString(readSingle),
+ eagerlyFetchResults: true,
+ variables: {'id': '1'},
+ ),
+ );
+ final responses = [
+ {'id': '1', 'name': firstUpdateName, '__typename': 'Item'},
+ {'id': '1', 'name': secondUpdateName, '__typename': 'Item'},
+ ].map((item) => Response(data: <String, dynamic>{'item': item}));
when(
link.request(any),
).thenAnswer(
(_) => Stream.fromIterable(responses),
);
+ final stream = client.subscribe(
+ SubscriptionOptions(
+ document: parseString(
+ r'''
+ subscription {
+ item {
+ id
+ name
+ }
+ }
+ ''',
+ ),
+ ),
+ );
+ expect(
+ stream,
+ emitsInOrder(
+ [
+ isA<QueryResult>().having(
+ (result) => result.data!['item']['name'],
+ 'name',
+ firstUpdateName,
+ ),
+ isA<QueryResult>().having(
+ (result) => result.data!['item']['name'],
+ 'name',
+ secondUpdateName,
+ )
+ ],
+ ),
+ );
+ expect(
+ observable.stream,
+ emitsInOrder(
+ [
+ isA<QueryResult>().having(
+ (result) => result.data,
+ 'name',
+ null,
+ ),
+ isA<QueryResult>().having(
+ (result) => result.data!['single']['name'],
+ 'name',
+ initialName,
+ ),
+ isA<QueryResult>().having(
+ (result) => result.data!['single']['name'],
+ 'name',
+ firstUpdateName,
+ ),
+ isA<QueryResult>().having(
+ (result) => result.data!['single']['name'],
+ 'name',
+ secondUpdateName,
+ )
+ ],
+ ),
+ );
+ });
+ test('parses results', () async {
+ final responses = [
+ {'id': '1', 'name': 'first', '__typename': 'Item'},
+ {'id': '2', 'name': 'second', '__typename': 'Item'},
+ ].map((item) => Response(data: <String, dynamic>{'item': item}));
+ when(link.request(any))
+ .thenAnswer((_) => Stream.fromIterable(responses));
final ResultParserFn<String> parserFn =
(data) => data['item']['name'] as String;

fix: Stream broadcast