Remove global socketClient variable, Subscription widget refactor to use WebsocketLink, update example app.
MODIFY
example/lib/main.dart
example/lib/main.dart
@@ -8,11 +8,6 @@ const String YOUR_PERSONAL_ACCESS_TOKEN = '<YOUR_PERSONAL_ACCESS_TOKEN>';
void main() => runApp(MyApp());
-class MyApp extends StatelessWidget {
- MyApp();
-
- @override
- Widget build(BuildContext context) {
final HttpLink httpLink = HttpLink(
uri: 'https://api.github.com/graphql',
);
@@ -21,15 +16,19 @@ class MyApp extends StatelessWidget {
getToken: () async => 'Bearer $YOUR_PERSONAL_ACCESS_TOKEN',
);
- final Link link = authLink.concat(httpLink);
-
- final WebSocketLink websocketLink = WebSocketLink(SocketClient(
- 'ws://api.github.com/graphql',
- headers: <String, String>{
- 'Authorization': 'Bearer $YOUR_PERSONAL_ACCESS_TOKEN',
- },
+// to handle subscriptions
+final WebSocketLink websocketLink = WebSocketLink(
+ url: 'ws://localhost:8080/ws/graphql',
config: SocketClientConfig(autoReconnect: true, inactivityTimeout: Duration(seconds: 15)),
- ));
+);
+
+// the order is important!
+final Link link = authLink.concat(httpLink).concat(websocketLink);
+
+class MyApp extends StatelessWidget {
+ @override
+ Widget build(BuildContext context) {
+ // websocketLink.connectOrReconnect();
final ValueNotifier<GraphQLClient> client = ValueNotifier<GraphQLClient>(
GraphQLClient(
@@ -115,18 +114,31 @@ class _MyHomePageState extends State<MyHomePage> {
}
// result.data can be either a [List<dynamic>] or a [Map<String, dynamic>]
- final List<dynamic> repositories =
- result.data['viewer']['repositories']['nodes'];
+ final List<dynamic> repositories = result.data['viewer']['repositories']['nodes'];
return Expanded(
child: ListView.builder(
itemCount: repositories.length,
- itemBuilder: (BuildContext context, int index) =>
- StarrableRepository(repository: repositories[index]),
+ itemBuilder: (BuildContext context, int index) => StarrableRepository(repository: repositories[index]),
),
);
},
),
+ Subscription<Map<String, dynamic>>(
+ 'test',
+ queries.testSubscription,
+ builder: ({loading, payload, dynamic error}) => loading ? Text('Loading...') : Text(payload.toString()),
+ ),
+ Subscription<Map<String, dynamic>>(
+ 'test',
+ queries.testSubscription,
+ builder: ({loading, payload, dynamic error}) => loading ? Text('Loading2...') : Text(payload.toString()),
+ ),
+ Subscription<Map<String, dynamic>>(
+ 'test',
+ queries.testSubscription,
+ builder: ({loading, payload, dynamic error}) => loading ? Text('Loading3...') : Text(payload.toString()),
+ )
],
),
),
@@ -198,9 +210,7 @@ class StarrableRepositoryState extends State<StarrableRepository> {
if (result.hasErrors) {
print(result.errors);
} else {
- final Map<String, Object> updated =
- Map<String, Object>.from(widget.repository)
- ..addAll(extractRepositoryData(result.data));
+ final Map<String, Object> updated = Map<String, Object>.from(widget.repository)..addAll(extractRepositoryData(result.data));
cache.write(typenameDataIdFromObject(updated), updated);
}
@@ -211,9 +221,7 @@ class StarrableRepositoryState extends State<StarrableRepository> {
builder: (BuildContext context) {
return AlertDialog(
title: Text(
- extractRepositoryData(result.data)['viewerHasStarred']
- ? 'Thanks for your star!'
- : 'Sorry you changed your mind!',
+ extractRepositoryData(result.data)['viewerHasStarred'] ? 'Thanks for your star!' : 'Sorry you changed your mind!',
),
actions: <Widget>[
SimpleDialogOption(

MODIFY
example/lib/queries/readRepositories.dart
example/lib/queries/readRepositories.dart
@@ -12,3 +12,12 @@ const String readRepositories = r'''
}
}
''';
+
+const String testSubscription = r'''
+ subscription test {
+ deviceChanged(id: 2) {
+ id
+ name
+ }
+ }
+''';

MODIFY
lib/src/core/query_options.dart
lib/src/core/query_options.dart
+import 'package:graphql_flutter/src/utilities/helpers.dart';
import 'package:meta/meta.dart';
/// [FetchPolicy] determines where the client may return a result from. The options are:
@@ -69,7 +70,7 @@ class QueryOptions extends BaseOptions {
);
/// The time interval (in milliseconds) on which this query should be
- /// refetched from the server.
+ /// re-fetched from the server.
int pollInterval;
}
@@ -143,33 +144,6 @@ class WatchQueryOptions extends QueryOptions {
}
// compare variables last, because maps take more time
- return _areDifferentVariables(a.variables, b.variables);
- }
-
- bool _areDifferentVariables(
- Map<String, dynamic> a,
- Map<String, dynamic> b,
- ) {
- if (a == null && b == null) {
- return false;
- }
-
- if (a == null || b == null) {
- return true;
- }
-
- if (a.length != b.length) {
- return true;
- }
-
- bool areDifferent = false;
-
- a.forEach((String key, dynamic value) {
- if ((!b.containsKey(key)) || b[key] != value) {
- areDifferent = true;
- }
- });
-
- return areDifferent;
+ return areDifferentVariables(a.variables, b.variables);
}
}

MODIFY
lib/src/graphql_client.dart
lib/src/graphql_client.dart
import 'dart:async';
-import 'package:meta/meta.dart';
-
-import 'package:graphql_flutter/src/core/query_manager.dart';
-import 'package:graphql_flutter/src/core/query_result.dart';
+import 'package:graphql_flutter/src/cache/cache.dart';
import 'package:graphql_flutter/src/core/observable_query.dart';
+import 'package:graphql_flutter/src/core/query_manager.dart';
import 'package:graphql_flutter/src/core/query_options.dart';
-
+import 'package:graphql_flutter/src/core/query_result.dart';
+import 'package:graphql_flutter/src/link/fetch_result.dart';
import 'package:graphql_flutter/src/link/link.dart';
-import 'package:graphql_flutter/src/cache/cache.dart';
+import 'package:graphql_flutter/src/link/operation.dart';
+import 'package:meta/meta.dart';
/// The link is a [Link] over which GraphQL documents will be resolved into a [FetchResult].
/// The cache is the initial [Cache] to use in the data store.
@@ -50,10 +50,9 @@ class GraphQLClient {
return queryManager.mutate(options);
}
- /// This subscribes to a GraphQL subscription according to the options specified and returns an
+ /// This subscribes to a GraphQL subscription according to the options specified and returns a
/// [Stream] which either emits received data or an error.
- Stream<dynamic> subscribe(dynamic options) {
- // TODO: merge the subscription client with the new client
- return const Stream<dynamic>.empty();
+ Stream<FetchResult> subscribe(Operation operation) {
+ return execute(link: link, operation: operation);
}
}

MODIFY
lib/src/link/http/link_http.dart
lib/src/link/http/link_http.dart
import 'dart:async';
import 'dart:convert';
-import 'package:meta/meta.dart';
-import 'package:http/http.dart';
-import 'package:http_parser/http_parser.dart';
-
-import 'package:graphql_flutter/src/link/link.dart';
-import 'package:graphql_flutter/src/link/operation.dart';
import 'package:graphql_flutter/src/link/fetch_result.dart';
-import 'package:graphql_flutter/src/link/http/http_config.dart';
import 'package:graphql_flutter/src/link/http/fallback_http_config.dart';
+import 'package:graphql_flutter/src/link/http/http_config.dart';
+import 'package:graphql_flutter/src/link/link.dart';
+import 'package:graphql_flutter/src/link/operation.dart';
+import 'package:http/http.dart';
+import 'package:http_parser/http_parser.dart';
+import 'package:meta/meta.dart';
+import 'package:rxdart/rxdart.dart';
class HttpLink extends Link {
HttpLink({
@@ -24,6 +24,11 @@ class HttpLink extends Link {
Operation operation, [
NextLink forward,
]) {
+ if (operation.isSubscription) {
+ if (forward == null) return Observable.error(Exception('This link does not support subscriptions.'));
+ return forward(operation);
+ }
+
final Client fetcher = fetch ?? Client();
final HttpConfig linkConfig = HttpConfig(
@@ -50,8 +55,7 @@ class HttpLink extends Link {
);
}
- final HttpOptionsAndBody httpOptionsAndBody =
- _selectHttpOptionsAndBody(
+ final HttpOptionsAndBody httpOptionsAndBody = _selectHttpOptionsAndBody(
operation,
fallbackHttpConfig,
linkConfig,
@@ -221,8 +225,7 @@ FetchResult _parseResponse(Response response) {
/// The default fallback encoding is set to UTF-8 according to the IETF RFC4627 standard
/// which specifies the application/json media type:
/// "JSON text SHALL be encoded in Unicode. The default encoding is UTF-8."
-Encoding _determineEncodingFromResponse(Response response,
- [Encoding fallback = utf8]) {
+Encoding _determineEncodingFromResponse(Response response, [Encoding fallback = utf8]) {
final String contentType = response.headers['content-type'];
if (contentType == null) {

MODIFY
lib/src/link/link.dart
lib/src/link/link.dart
import 'dart:async';
-import 'package:graphql_flutter/src/link/operation.dart';
import 'package:graphql_flutter/src/link/fetch_result.dart';
+import 'package:graphql_flutter/src/link/operation.dart';
typedef NextLink = Stream<FetchResult> Function(
Operation operation,
@@ -27,20 +27,11 @@ Link _concat(
}
class Link {
- Link({
- this.request,
- });
+ Link({this.request});
- final RequestHandler request;
+ RequestHandler request;
- Link concat(Link next) {
- return _concat(this, next);
- }
+ Link concat(Link next) => _concat(this, next);
}
-Stream<FetchResult> execute({
- Link link,
- Operation operation,
-}) {
- return link.request(operation);
-}
+Stream<FetchResult> execute({Link link, Operation operation}) => link.request(operation);

MODIFY
lib/src/link/operation.dart
lib/src/link/operation.dart
@@ -15,7 +15,7 @@ class Operation {
final Map<String, dynamic> _context = <String, dynamic>{};
- /// Sets the context of an opration by merging the new context with the old one.
+ /// Sets the context of an operation by merging the new context with the old one.
void setContext(Map<String, dynamic> next) {
_context.addAll(next);
}
@@ -27,6 +27,8 @@ class Operation {
return result;
}
+ bool get isSubscription => document.contains(RegExp(r'.*?subscription ' + operationName));
+
String toKey() {
/// XXX we're assuming here that variables will be serialized in the same order.
/// that might not always be true

MODIFY
lib/src/link/web_socket/link_web_socket.dart
lib/src/link/web_socket/link_web_socket.dart
@@ -2,18 +2,63 @@ import 'package:graphql_flutter/src/link/fetch_result.dart';
import 'package:graphql_flutter/src/link/link.dart';
import 'package:graphql_flutter/src/link/operation.dart';
import 'package:graphql_flutter/src/socket_client.dart';
+import 'package:graphql_flutter/src/utilities/helpers.dart';
import 'package:graphql_flutter/src/websocket/messages.dart';
+import 'package:meta/meta.dart';
+/// A websocket [Link] implementation to support the websocket transport.
+/// It supports subscriptions, query and mutation operations as well.
+///
+/// This link is aware of [AuthLink], so the headers specified there are automatically
+/// applied when connecting to the socket server, unless explicitly overridden by the [headers]
+/// parameter.
+///
+/// There's an option called [reconnectOnHeaderChange] that makes it possible to reconnect to the server when the
+/// headers have changed. For example, if the user logs in with another user account and the `Authorization` header changes.
+/// This could be desired because of the nature of websocket connections: headers can only be specified upon connecting.
+///
+/// NOTE: the actual socket connection will only get established after an [Operation] is handled by this [WebSocketLink].
+/// If you'd like to connect to the socket server instantly, call the [connectOrReconnect] method after creating this [WebSocketLink] instance.
class WebSocketLink extends Link {
- final SocketClient socketClient;
- WebSocketLink(this.socketClient)
- : super(
- request: (
- Operation operation, [
- NextLink forward,
- ]) {
- return socketClient.subscribe(SubscriptionRequest(operation.operationName, operation.document, operation.variables), false).map(
- (result) => FetchResult(data: result.data, errors: result.errors, context: operation.getContext(), extensions: operation.extensions));
- },
- );
+ /// Creates a new [WebSocketLink] instance with the specified config.
+ WebSocketLink({@required this.url, this.headers, this.reconnectOnHeaderChange = true, this.config = const SocketClientConfig()}) : super() {
+ request = _doOperation;
+ }
+
+ final String url;
+ final Map<String, dynamic> headers;
+ final bool reconnectOnHeaderChange;
+ final SocketClientConfig config;
+
+ // cannot be final because we're changing the instance upon a header change.
+ SocketClient _socketClient;
+
+ Stream<FetchResult> _doOperation(Operation operation, [NextLink forward]) {
+ final concatHeaders = <String, dynamic>{};
+ final Map<String, dynamic> context = operation.getContext();
+ if (context != null && context.containsKey('headers')) concatHeaders.addAll(context['headers']);
+ if (headers != null) concatHeaders.addAll(headers);
+
+ if (_socketClient == null || (reconnectOnHeaderChange && areDifferentVariables(_socketClient.headers, concatHeaders))) {
+ if (_socketClient != null) print('Creating a new socket client, because the headers have changed..');
+ connectOrReconnect(headers: concatHeaders);
+ }
+
+ return _socketClient
+ .subscribe(SubscriptionRequest(operation), true)
+ .map((result) => FetchResult(data: result.data, errors: result.errors, context: operation.getContext(), extensions: operation.extensions));
+ }
+
+ /// Connects or reconnects to the server with the specified headers.
+ void connectOrReconnect({Map<String, dynamic> headers}) {
+ _socketClient?.dispose();
+ _socketClient = SocketClient(url, headers: headers ?? this.headers, config: config);
+ }
+
+ /// Disposes the underlying socket client explicitly. Only use this, if you want to disconnect from
+ /// the current server in favour of another one. If that's the case, create a new [WebSocketLink] instance.
+ Future<void> dispose() async {
+ await _socketClient?.dispose();
+ _socketClient = null;
+ }
}

MODIFY
lib/src/socket_client.dart
lib/src/socket_client.dart
@@ -7,9 +7,15 @@ import 'package:graphql_flutter/src/websocket/messages.dart';
import 'package:rxdart/subjects.dart';
import 'package:uuid/uuid.dart';
-SocketClient socketClient;
-
class SocketClientConfig {
+ const SocketClientConfig(
+ {this.autoReconnect = true,
+ this.queryAndMutationTimeout = const Duration(seconds: 10),
+ this.inactivityTimeout = const Duration(seconds: 30),
+ this.delayBetweenReconnectionAttempts = const Duration(seconds: 5),
+ this.compression = CompressionOptions.compressionDefault,
+ this.initPayload});
+
/// Whether to reconnect to the server after detecting connection loss.
final bool autoReconnect;
@@ -30,11 +36,10 @@ class SocketClientConfig {
// If null, no timeout is applied, although not recommended.
final Duration queryAndMutationTimeout;
- const SocketClientConfig(
- {this.autoReconnect = true,
- this.queryAndMutationTimeout = const Duration(seconds: 10),
- this.inactivityTimeout = const Duration(seconds: 30),
- this.delayBetweenReconnectionAttempts = const Duration(seconds: 5)});
+ final CompressionOptions compression;
+
+ /// The initial payload that will be sent to the server upon connection. Can be null.
+ final Map<String, String> initPayload;
}
enum SocketConnectionState { NOT_CONNECTED, CONNECTING, CONNECTED }
@@ -48,13 +53,24 @@ enum SocketConnectionState { NOT_CONNECTED, CONNECTING, CONNECTED }
/// lifting of socket state management. Once you're done with the socket connection, make sure
/// you call the [dispose] method to release all allocated resources.
class SocketClient {
+ SocketClient(
+ this.url, {
+ this.protocols = const <String>[
+ 'graphql-ws',
+ ],
+ this.headers = const <String, String>{
+ 'content-type': 'application/json',
+ },
+ this.config = const SocketClientConfig(),
+ }) {
+ _connect();
+ }
+
final Uuid _uuid = Uuid();
final String url;
final SocketClientConfig config;
final Iterable<String> protocols;
- final Map<String, String> initPayload;
final Map<String, dynamic> headers;
- final CompressionOptions compression;
final _connectionStateController = BehaviorSubject<SocketConnectionState>();
Timer _reconnectTimer;
@@ -64,34 +80,20 @@ class SocketClient {
StreamSubscription<ConnectionKeepAlive> _keepAliveSubscription;
StreamSubscription<GraphQLSocketMessage> _messageSubscription;
- SocketClient(this.url,
- {this.protocols = const <String>[
- 'graphql-ws',
- ],
- this.headers = const <String, String>{
- 'content-type': 'application/json',
- },
- this.compression = CompressionOptions.compressionDefault,
- this.config = const SocketClientConfig(),
- this.initPayload}) {
- _connect();
- }
-
/// Connects to the server.
///
/// If this instance is disposed, this method does nothing.
Future<void> _connect() async {
if (_connectionStateController.isClosed) return;
- if (_socket != null) print('Reconnecting to socket...');
_connectionStateController.value = SocketConnectionState.CONNECTING;
print('Connecting to websocket: $url...');
try {
- _socket = await WebSocket.connect(url, protocols: protocols, headers: headers, compression: compression);
+ _socket = await WebSocket.connect(url, protocols: protocols, headers: headers, compression: config.compression);
_connectionStateController.value = SocketConnectionState.CONNECTED;
print('Connected to websocket.');
- _write(InitOperation(initPayload));
+ _write(InitOperation(config.initPayload));
_messageStream = _socket.asBroadcastStream().map<GraphQLSocketMessage>(_parseSocketMessage);
@@ -106,7 +108,7 @@ class SocketClient {
_messageSubscription = _messageStream.listen(
(dynamic data) {
- print("data: $data");
+ print('data: $data');
},
onDone: () {
print('done');
@@ -114,7 +116,7 @@ class SocketClient {
},
cancelOnError: true,
onError: (dynamic e) {
- print("error: $e");
+ print('error: $e');
});
} catch (e) {
onConnectionLost();
@@ -152,6 +154,7 @@ class SocketClient {
/// Use this method if you'd like to disconnect from the specified server permanently,
/// and you'd like to connect to another server instead of the current one.
Future<void> dispose() async {
+ print('Disposing socket client..');
_reconnectTimer?.cancel();
await _socket?.close();
await _keepAliveSubscription?.cancel();
@@ -204,12 +207,28 @@ class SocketClient {
/// If the request is a subscription, obviously no timeout is applied.
///
/// In case of socket disconnection, the returned stream will be closed.
- Stream<SubscriptionData> subscribe(final SubscriptionRequest payload, final bool isSubscription) {
+ Stream<SubscriptionData> subscribe(final SubscriptionRequest payload, final bool waitForConnection) {
final String id = _uuid.v4();
final StreamController<SubscriptionData> response = StreamController<SubscriptionData>();
+ StreamSubscription<SocketConnectionState> sub;
+ final addTimeout = !payload.operation.isSubscription && config.queryAndMutationTimeout != null;
response.onListen = () {
- if (_connectionStateController.value == SocketConnectionState.CONNECTED && _socket != null) {
+ final waitForConnectedStateWithoutTimeout = _connectionStateController
+ .startWith(waitForConnection ? null : SocketConnectionState.CONNECTED)
+ .where((state) => state == SocketConnectionState.CONNECTED)
+ .take(1);
+
+ final waitForConnectedState = addTimeout
+ ? waitForConnectedStateWithoutTimeout.timeout(config.queryAndMutationTimeout, onTimeout: (e) {
+ print('Connection timed out.');
+ response.addError(TimeoutException('Connection timed out.'));
+ e.close();
+ response.close();
+ })
+ : waitForConnectedStateWithoutTimeout;
+
+ sub = waitForConnectedState.listen((_) {
final dataErrorComplete = _messageStream.where((GraphQLSocketMessage message) {
if (message is SubscriptionData) return message.id == id;
if (message is SubscriptionError) return message.id == id;
@@ -217,7 +236,6 @@ class SocketClient {
return false;
}).takeWhile((_) => !response.isClosed);
- final addTimeout = !isSubscription && config.queryAndMutationTimeout != null;
final subscriptionComplete = addTimeout
? dataErrorComplete
.where((GraphQLSocketMessage message) => message is SubscriptionComplete)
@@ -242,11 +260,11 @@ class SocketClient {
.listen((GraphQLSocketMessage message) => response.addError(message));
_write(StartOperation(id, payload));
- } else {
- response.addError(Exception('Not connected to the server.'));
- }
+ });
};
+
response.onCancel = () {
+ sub?.cancel();
if (_connectionStateController.value == SocketConnectionState.CONNECTED && _socket != null) _write(StopOperation(id));
};

MODIFY
lib/src/utilities/helpers.dart
lib/src/utilities/helpers.dart
bool notNull(Object any) {
return any != null;
}
+
+bool areDifferentVariables(
+ Map<String, dynamic> a,
+ Map<String, dynamic> b,
+) {
+ if (a == null && b == null) {
+ return false;
+ }
+
+ if (a == null || b == null) {
+ return true;
+ }
+
+ if (a.length != b.length) {
+ return true;
+ }
+
+ bool areDifferent = false;
+
+ a.forEach((String key, dynamic value) {
+ if ((!b.containsKey(key)) || b[key] != value) {
+ areDifferent = true;
+ }
+ });
+
+ return areDifferent;
+}

MODIFY
lib/src/websocket/messages.dart
lib/src/websocket/messages.dart
import 'dart:convert';
+import 'package:graphql_flutter/src/link/operation.dart';
+
/// These messages represent the structures used for Client-server communication
/// in a GraphQL web-socket subscription. Each message is represented in a JSON
/// format where the data type is denoted by the `type` field.
@@ -70,17 +72,14 @@ class InitOperation extends GraphQLSocketMessage {
/// defined in the query provided. Additional variables can be provided
/// and sent to the server for processing.
class SubscriptionRequest extends JsonSerializable {
- SubscriptionRequest(this.operationName, this.query, this.variables);
-
- final String operationName;
- final String query;
- final dynamic variables;
+ SubscriptionRequest(this.operation);
+ final Operation operation;
@override
dynamic toJson() => <String, dynamic>{
- 'operationName': operationName,
- 'query': query,
- 'variables': variables,
+ 'operationName': operation.operationName,
+ 'query': operation.document,
+ 'variables': operation.variables,
};
}
@@ -156,8 +155,7 @@ class ConnectionKeepAlive extends GraphQLSocketMessage {
/// payload. The user should check the errors result before processing the
/// data value. These error are from the query resolvers.
class SubscriptionData extends GraphQLSocketMessage {
- SubscriptionData(this.id, this.data, this.errors)
- : super(MessageTypes.GQL_DATA);
+ SubscriptionData(this.id, this.data, this.errors) : super(MessageTypes.GQL_DATA);
final String id;
final dynamic data;

MODIFY
lib/src/widgets/subscription.dart
lib/src/widgets/subscription.dart
import 'dart:async';
import 'package:flutter/widgets.dart';
+import 'package:graphql_flutter/src/graphql_client.dart';
+import 'package:graphql_flutter/src/link/fetch_result.dart';
+import 'package:graphql_flutter/src/link/operation.dart';
+import 'package:graphql_flutter/src/utilities/helpers.dart';
+import 'package:graphql_flutter/src/widgets/graphql_provider.dart';
-import '../socket_client.dart';
import '../websocket/messages.dart';
typedef OnSubscriptionCompleted = void Function();
-typedef SubscriptionBuilder = Widget Function({
+typedef SubscriptionBuilder<T> = Widget Function({
final bool loading,
- final dynamic payload,
+ final T payload,
final dynamic error,
});
-class Subscription extends StatefulWidget {
+class Subscription<T> extends StatefulWidget {
const Subscription(
this.operationName,
this.query, {
@@ -27,33 +31,29 @@ class Subscription extends StatefulWidget {
final String operationName;
final String query;
final dynamic variables;
- final SubscriptionBuilder builder;
+ final SubscriptionBuilder<T> builder;
final OnSubscriptionCompleted onCompleted;
- final dynamic initial;
+ final T initial;
@override
- _SubscriptionState createState() => _SubscriptionState();
+ _SubscriptionState<T> createState() => _SubscriptionState<T>();
}
-class _SubscriptionState extends State<Subscription> {
+class _SubscriptionState<T> extends State<Subscription<T>> {
bool _loading = true;
- dynamic _data;
+ T _data;
dynamic _error;
+ StreamSubscription<FetchResult> _subscription;
- bool _alive = true;
+ void _initSubscription() {
+ final GraphQLClient client = GraphQLProvider.of(context).value;
+ assert(client != null);
+ final Operation operation = Operation(document: widget.query, variables: widget.variables, operationName: widget.operationName);
- @override
- void initState() {
- super.initState();
-
- final Stream<SubscriptionData> stream = socketClient.subscribe(SubscriptionRequest(widget.operationName, widget.query, widget.variables), true);
-
- stream.takeWhile((SubscriptionData message) => _alive).listen(
- _onData,
- onError: _onError,
- onDone: _onDone,
- );
+ final Stream<FetchResult> stream = client.subscribe(operation);
+ if (_subscription == null) {
+ // Set the initial value for the first time.
if (widget.initial != null) {
setState(() {
_loading = true;
@@ -63,13 +63,38 @@ class _SubscriptionState extends State<Subscription> {
}
}
+ _subscription?.cancel();
+ _subscription = stream.listen(
+ _onData,
+ onError: _onError,
+ onDone: _onDone,
+ );
+ }
+
+ @override
+ void didChangeDependencies() {
+ super.didChangeDependencies();
+ _initSubscription();
+ }
+
+ @override
+ void didUpdateWidget(Subscription<T> oldWidget) {
+ super.didUpdateWidget(oldWidget);
+
+ if (widget.query != oldWidget.query ||
+ widget.operationName != oldWidget.operationName ||
+ areDifferentVariables(widget.variables, oldWidget.variables)) {
+ _initSubscription();
+ }
+ }
+
@override
void dispose() {
- _alive = false;
+ _subscription?.cancel();
super.dispose();
}
- void _onData(final SubscriptionData message) {
+ void _onData(final FetchResult message) {
setState(() {
_loading = false;
_data = message.data;

MODIFY
pubspec.yaml
pubspec.yaml
@@ -23,4 +23,4 @@ dev_dependencies:
test: ^1.3.0
environment:
- sdk: ">=2.0.0 <3.0.0"
+ sdk: ">=2.1.0 <3.0.0"

Remove global socketClient variable, Subscription widget refactor to use WebsocketLink, update example app.