Refactored SocketClient, and removed GraphQLSocket class.
MODIFY
example/lib/main.dart
example/lib/main.dart
@@ -7,13 +7,13 @@ import './queries/readRepositories.dart' as queries;
const String YOUR_PERSONAL_ACCESS_TOKEN = '<YOUR_PERSONAL_ACCESS_TOKEN_HERE>';
void main() async {
- final websocketLink = WebSocketLink(SocketClient(GraphQLSocket(
+ final websocketLink = WebSocketLink(SocketClient(
'ws://192.168.88.79:8080/ws/graphql',
headers: <String, String>{
'Authorization': 'Bearer $YOUR_PERSONAL_ACCESS_TOKEN',
},
- config: GraphQLSocketConfig(autoReconnect: true, inactivityTimeout: Duration(seconds: 15)),
- )));
+ config: SocketClientConfig(autoReconnect: true, inactivityTimeout: Duration(seconds: 15)),
+ ));
runApp(MyApp(websocketLink));
}

MODIFY
lib/src/socket_client.dart
lib/src/socket_client.dart
import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
import 'package:graphql_flutter/graphql_flutter.dart';
+import 'package:graphql_flutter/src/websocket/messages.dart';
import 'package:uuid/uuid.dart';
SocketClient socketClient;
class SocketClientConfig {
- /// The duration after which a query or mutation should timeout.
- final Duration queryOrMutationTimeout;
-
- const SocketClientConfig({this.queryOrMutationTimeout = const Duration(seconds: 30)});
+ /// Whether to reconnect to the server after detecting connection loss.
+ final bool autoReconnect;
+
+ /// The duration after which the connection is considered unstable, because no keep alive message
+ /// was received from the server. The connection to the server will be closed. If [autoReconnect] is
+ /// set to true, we try to reconnect to the server.
+ ///
+ /// If null, the keep alive messages will be ignored.
+ final Duration inactivityTimeout;
+
+ /// The duration after a connection loss that needs to pass before trying to reconnect to the socket.
+ /// This field only takes effect when [autoReconnect] is set to true.
+ ///
+ /// If null, the reconnection will occur immediately, although not recommended.
+ final Duration delayBetweenReconnectionAttempts;
+
+ const SocketClientConfig(
+ {this.autoReconnect = true,
+ this.inactivityTimeout = const Duration(seconds: 30),
+ this.delayBetweenReconnectionAttempts = const Duration(seconds: 5)});
}
+enum SocketConnectionState { NOT_CONNECTED, CONNECTING, CONNECTED }
+
+/// Wraps a standard web socket instance to marshal and un-marshal the server /
+/// client payloads into dart object representation.
+///
+/// This class also deals with reconnection, handles timeout and keep alive messages.
class SocketClient {
final Uuid _uuid = Uuid();
- final GraphQLSocket _socket;
+ 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 = StreamController<SocketConnectionState>.broadcast();
+ final _messageController = StreamController<GraphQLSocketMessage>.broadcast();
+
+ WebSocket _socket;
+
+ StreamSubscription _keepAliveSubscription;
+ StreamSubscription _messageSubscription;
+ StreamSubscription _connectionStateSubscription;
+
+ 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();
+
+ _connectionStateSubscription = connectionState.listen((SocketConnectionState state) {
+ print('WebSocket connection state changed to: $state');
+ if (state == SocketConnectionState.CONNECTED) _write(InitOperation(initPayload));
+ });
+ }
+
+ Future<void> _connect({Duration delayUntilConnectionAttempt}) async {
+ if (delayUntilConnectionAttempt != null) {
+ print('Scheduling to connect in ${delayUntilConnectionAttempt.inSeconds} seconds...');
+ await Future<void>.delayed(delayUntilConnectionAttempt);
+ }
- SocketClient(this._socket, {this.config = const SocketClientConfig(), Map<String, String> initPayload}) {
- _socket.connectionAck.listen(print);
- _socket.connectionError.listen(print);
- _socket.unknownData.listen(print);
+ if (_socket != null) print('Reconnecting to socket...');
+ _connectionStateController.add(SocketConnectionState.CONNECTING);
- _socket.connectionState.listen((GraphQLSocketConnectionState state) {
- print('WebSocket connection state changed to: $state');
- if (state == GraphQLSocketConnectionState.CONNECTED) _socket.write(InitOperation(initPayload));
+ try {
+ _socket = await WebSocket.connect(url, protocols: protocols, headers: headers, compression: compression);
+ _connectionStateController.add(SocketConnectionState.CONNECTED);
+
+ final messageStream = _socket.asBroadcastStream().map<GraphQLSocketMessage>(_parseSocketMessage);
+ _messageController.addStream(messageStream);
+
+ if (config.inactivityTimeout != null) {
+ _keepAliveSubscription = _connectionKeepAlive.timeout(config.inactivityTimeout, onTimeout: (event) {
+ event.close();
+ _socket.close(WebSocketStatus.goingAway);
+ }).listen(null);
+ }
+
+ _messageSubscription = messageStream.listen(
+ (dynamic data) {
+ print("data: $data");
+ },
+ onDone: () {
+ print('done');
+ onConnectionLost();
+ },
+ cancelOnError: true,
+ onError: (dynamic e) {
+ print("error: $e");
});
+ } catch (e) {
+ onConnectionLost();
+ }
+ }
+
+ void onConnectionLost() {
+ _keepAliveSubscription?.cancel();
+ _messageSubscription?.cancel();
+ _connectionStateController.add(SocketConnectionState.NOT_CONNECTED);
+
+ if (config.autoReconnect) {
+ _connect(delayUntilConnectionAttempt: config.delayBetweenReconnectionAttempts);
}
+ }
+
+ /// Closes the underlying socket if connected, and stops reconnection attempts.
+ /// After calling this method, this [GraphQLSocket] instance must be considered
+ /// obsolete, and should not be used. Instead, create a new instance of this class.
+ void dispose() {
+ _socket?.close();
+ _keepAliveSubscription?.cancel();
+ _messageSubscription?.cancel();
+ _connectionStateSubscription?.cancel();
+ }
+
+ static GraphQLSocketMessage _parseSocketMessage(dynamic message) {
+ final Map<String, dynamic> map = json.decode(message);
+ final String type = map['type'] ?? 'unknown';
+ final dynamic payload = map['payload'] ?? <String, dynamic>{};
+ final String id = map['id'] ?? 'none';
+
+ switch (type) {
+ case MessageTypes.GQL_CONNECTION_ACK:
+ return ConnectionAck();
+ case MessageTypes.GQL_CONNECTION_ERROR:
+ return ConnectionError(payload);
+ case MessageTypes.GQL_CONNECTION_KEEP_ALIVE:
+ return ConnectionKeepAlive();
+ case MessageTypes.GQL_DATA:
+ final dynamic data = payload['data'];
+ final dynamic errors = payload['errors'];
+ return SubscriptionData(id, data, errors);
+ case MessageTypes.GQL_ERROR:
+ return SubscriptionError(id, payload);
+ case MessageTypes.GQL_COMPLETE:
+ return SubscriptionComplete(id);
+ default:
+ return UnknownData(map);
+ }
+ }
+
+ void _write(final GraphQLSocketMessage message) {
+ _socket.add(
+ json.encode(
+ message,
+ toEncodable: (dynamic m) => m.toJson(),
+ ),
+ );
+ }
+
+ Stream<SocketConnectionState> get connectionState => _connectionStateController.stream.distinct();
+
+ Stream<ConnectionAck> get _connectionAck =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionAck).cast<ConnectionAck>();
+
+ Stream<ConnectionError> get _connectionError =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionError).cast<ConnectionError>();
+
+ Stream<ConnectionKeepAlive> get _connectionKeepAlive =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionKeepAlive).cast<ConnectionKeepAlive>();
+
+ Stream<UnknownData> get _unknownData =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is UnknownData).cast<UnknownData>();
+
+ Stream<SubscriptionData> get _subscriptionData =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionData).cast<SubscriptionData>();
+
+ Stream<SubscriptionError> get _subscriptionError =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionError).cast<SubscriptionError>();
+
+ Stream<SubscriptionComplete> get _subscriptionComplete =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionComplete).cast<SubscriptionComplete>();
Stream<SubscriptionData> subscribe(final SubscriptionRequest payload) {
final String id = _uuid.v4();
final StreamController<SubscriptionData> response = StreamController<SubscriptionData>();
- final Stream<SubscriptionComplete> complete = _socket.subscriptionComplete.where((SubscriptionComplete message) => message.id == id).take(1);
+ final Stream<SubscriptionComplete> complete = _subscriptionComplete.where((SubscriptionComplete message) => message.id == id).take(1);
final Stream<SubscriptionData> data =
- _socket.subscriptionData.where((SubscriptionData message) => message.id == id).takeWhile((_) => !response.isClosed);
+ _subscriptionData.where((SubscriptionData message) => message.id == id).takeWhile((_) => !response.isClosed);
final Stream<SubscriptionError> error =
- _socket.subscriptionError.where((SubscriptionError message) => message.id == id).takeWhile((_) => !response.isClosed);
+ _subscriptionError.where((SubscriptionError message) => message.id == id).takeWhile((_) => !response.isClosed);
complete.listen((_) => response.close());
data.listen((SubscriptionData message) => response.add(message));
error.listen((SubscriptionError message) => response.addError(message));
- _socket.connectionState
- .where((GraphQLSocketConnectionState state) => state == GraphQLSocketConnectionState.CONNECTED)
+ connectionState
+ .where((SocketConnectionState state) => state == SocketConnectionState.CONNECTED)
.takeWhile((_) => !response.isClosed)
- .listen((_) => _socket.write(StartOperation(id, payload)));
+ .listen((_) => _write(StartOperation(id, payload)));
// response.onListen = () => );
response.onCancel = () {
- _socket.write(StopOperation(id));
+ _write(StopOperation(id));
};
return response.stream;

MODIFY
lib/src/websocket/socket.dart
lib/src/websocket/socket.dart
-import 'dart:async';
-import 'dart:convert';
-import 'dart:io';
-import 'package:graphql_flutter/src/websocket/messages.dart';
-
-class GraphQLSocketConfig {
- /// Whether to reconnect to the server after detecting connection loss.
- final bool autoReconnect;
-
- /// The duration after which the connection is considered unstable, because no keep alive message
- /// was received from the server. The connection to the server will be closed. If [autoReconnect] is
- /// set to true, we try to reconnect to the server.
- ///
- /// If null, the keep alive messages will be ignored.
- final Duration inactivityTimeout;
-
- /// The duration after a connection loss that needs to pass before trying to reconnect to the socket.
- /// This field only takes effect when [autoReconnect] is set to true.
- ///
- /// If null, the reconnection will occur immediately, although not recommended.
- final Duration delayBetweenReconnectionAttempts;
-
- const GraphQLSocketConfig(
- {this.autoReconnect = true,
- this.inactivityTimeout = const Duration(seconds: 30),
- this.delayBetweenReconnectionAttempts = const Duration(seconds: 5)});
-}
-
-enum GraphQLSocketConnectionState { NOT_CONNECTED, CONNECTING, CONNECTED }
-
-/// Wraps a standard web socket instance to marshal and un-marshal the server /
-/// client payloads into dart object representation.
-///
-/// This class also deals with reconnection, handles timeout and keep alive messages.
-class GraphQLSocket {
- final String url;
- final Iterable<String> protocols;
- final Map<String, dynamic> headers;
- final CompressionOptions compression;
- final GraphQLSocketConfig config;
- final _connectionStateController = StreamController<GraphQLSocketConnectionState>.broadcast();
- final _messageController = StreamController<GraphQLSocketMessage>.broadcast();
- WebSocket _socket;
-
- StreamSubscription _keepAliveSubscription;
-
- GraphQLSocket(this.url,
- {this.protocols = const <String>[
- 'graphql-ws',
- ],
- this.headers = const <String, String>{
- 'content-type': 'application/json',
- },
- this.compression = CompressionOptions.compressionDefault,
- this.config = const GraphQLSocketConfig()}) {
- _connect();
- }
-
- Future<void> _connect({Duration delayUntilConnectionAttempt}) async {
- if (delayUntilConnectionAttempt != null) {
- print('Scheduling to connect in ${delayUntilConnectionAttempt.inSeconds} seconds...');
- await Future<void>.delayed(delayUntilConnectionAttempt);
- }
-
- if (_socket != null) print('Reconnecting to socket...');
- _connectionStateController.add(GraphQLSocketConnectionState.CONNECTING);
-
- try {
- _socket = await WebSocket.connect(url, protocols: protocols, headers: headers, compression: compression);
- _connectionStateController.add(GraphQLSocketConnectionState.CONNECTED);
-
- final messageStream = _socket.asBroadcastStream().map<GraphQLSocketMessage>(_parseSocketMessage);
- _messageController.addStream(messageStream);
-
- if (config.inactivityTimeout != null) {
- _keepAliveSubscription = _connectionKeepAlive.timeout(config.inactivityTimeout, onTimeout: (event) {
- event.close();
- _socket.close(WebSocketStatus.goingAway);
- }).listen(null);
- }
-
- messageStream.listen(
- (dynamic data) {
- print("data: $data");
- },
- onDone: () {
- print('done');
- onConnectionLost();
- },
- cancelOnError: true,
- onError: (dynamic e) {
- print("error: $e");
- });
- } catch (e) {
- onConnectionLost();
- }
- }
-
- void onConnectionLost() {
- _keepAliveSubscription?.cancel();
- _connectionStateController.add(GraphQLSocketConnectionState.NOT_CONNECTED);
-
- if (config.autoReconnect) {
- _connect(delayUntilConnectionAttempt: config.delayBetweenReconnectionAttempts);
- }
- }
-
- GraphQLSocketMessage _parseSocketMessage(dynamic message) {
- final Map<String, dynamic> map = json.decode(message);
- final String type = map['type'] ?? 'unknown';
- final dynamic payload = map['payload'] ?? <String, dynamic>{};
- final String id = map['id'] ?? 'none';
-
- switch (type) {
- case MessageTypes.GQL_CONNECTION_ACK:
- return ConnectionAck();
- case MessageTypes.GQL_CONNECTION_ERROR:
- return ConnectionError(payload);
- case MessageTypes.GQL_CONNECTION_KEEP_ALIVE:
- return ConnectionKeepAlive();
- case MessageTypes.GQL_DATA:
- final dynamic data = payload['data'];
- final dynamic errors = payload['errors'];
- return SubscriptionData(id, data, errors);
- case MessageTypes.GQL_ERROR:
- return SubscriptionError(id, payload);
- case MessageTypes.GQL_COMPLETE:
- return SubscriptionComplete(id);
- default:
- return UnknownData(map);
- }
- }
-
- void write(final GraphQLSocketMessage message) {
- _socket.add(
- json.encode(
- message,
- toEncodable: (dynamic m) => m.toJson(),
- ),
- );
- }
-
- Stream<GraphQLSocketConnectionState> get connectionState => _connectionStateController.stream.distinct();
-
- Stream<ConnectionAck> get connectionAck =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionAck).cast<ConnectionAck>();
-
- Stream<ConnectionError> get connectionError =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionError).cast<ConnectionError>();
-
- Stream<ConnectionKeepAlive> get _connectionKeepAlive =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is ConnectionKeepAlive).cast<ConnectionKeepAlive>();
-
- Stream<UnknownData> get unknownData =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is UnknownData).cast<UnknownData>();
-
- Stream<SubscriptionData> get subscriptionData =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionData).cast<SubscriptionData>();
-
- Stream<SubscriptionError> get subscriptionError =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionError).cast<SubscriptionError>();
-
- Stream<SubscriptionComplete> get subscriptionComplete =>
- _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionComplete).cast<SubscriptionComplete>();
-}

Refactored SocketClient, and removed GraphQLSocket class.