Preparing socket connection auto reconnect and keep alive.
MODIFY
example/ios/Runner.xcodeproj/project.pbxproj
example/ios/Runner.xcodeproj/project.pbxproj
/* Begin PBXBuildFile section */
1498D2341E8E89220040F4C2 /* GeneratedPluginRegistrant.m in Sources */ = {isa = PBXBuildFile; fileRef = 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */; };
- 2D5378261FAA1A9400D5DBA9 /* flutter_assets in Resources */ = {isa = PBXBuildFile; fileRef = 2D5378251FAA1A9400D5DBA9 /* flutter_assets */; };
3B3967161E833CAA004F5970 /* AppFrameworkInfo.plist in Resources */ = {isa = PBXBuildFile; fileRef = 3B3967151E833CAA004F5970 /* AppFrameworkInfo.plist */; };
3B80C3941E831B6300D905FE /* App.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 3B80C3931E831B6300D905FE /* App.framework */; };
3B80C3951E831B6300D905FE /* App.framework in Embed Frameworks */ = {isa = PBXBuildFile; fileRef = 3B80C3931E831B6300D905FE /* App.framework */; settings = {ATTRIBUTES = (CodeSignOnCopy, RemoveHeadersOnCopy, ); }; };
/* Begin PBXFileReference section */
1498D2321E8E86230040F4C2 /* GeneratedPluginRegistrant.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = GeneratedPluginRegistrant.h; sourceTree = "<group>"; };
1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; path = GeneratedPluginRegistrant.m; sourceTree = "<group>"; };
- 2D5378251FAA1A9400D5DBA9 /* flutter_assets */ = {isa = PBXFileReference; lastKnownFileType = folder; name = flutter_assets; path = Flutter/flutter_assets; sourceTree = SOURCE_ROOT; };
3B3967151E833CAA004F5970 /* AppFrameworkInfo.plist */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text.plist.xml; name = AppFrameworkInfo.plist; path = Flutter/AppFrameworkInfo.plist; sourceTree = "<group>"; };
3B80C3931E831B6300D905FE /* App.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = App.framework; path = Flutter/App.framework; sourceTree = "<group>"; };
4A9CD12FDAA2BE6D4AEA7ACE /* libPods-Runner.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-Runner.a"; sourceTree = BUILT_PRODUCTS_DIR; };
9740EEB11CF90186004384FC /* Flutter */ = {
isa = PBXGroup;
children = (
- 2D5378251FAA1A9400D5DBA9 /* flutter_assets */,
3B80C3931E831B6300D905FE /* App.framework */,
3B3967151E833CAA004F5970 /* AppFrameworkInfo.plist */,
9740EEBA1CF902C7004384FC /* Flutter.framework */,
3B3967161E833CAA004F5970 /* AppFrameworkInfo.plist in Resources */,
9740EEB41CF90195004384FC /* Debug.xcconfig in Resources */,
97C146FE1CF9000F007C117D /* Assets.xcassets in Resources */,
- 2D5378261FAA1A9400D5DBA9 /* flutter_assets in Resources */,
97C146FC1CF9000F007C117D /* Main.storyboard in Resources */,
);
runOnlyForDeploymentPostprocessing = 0;

MODIFY
example/lib/main.dart
example/lib/main.dart
import 'package:flutter/material.dart';
-
import 'package:graphql_flutter/graphql_flutter.dart';
import './mutations/addStar.dart' as mutations;
@@ -7,9 +6,22 @@ import './queries/readRepositories.dart' as queries;
const String YOUR_PERSONAL_ACCESS_TOKEN = '<YOUR_PERSONAL_ACCESS_TOKEN_HERE>';
-void main() => runApp(MyApp());
+void main() async {
+ final websocketLink = WebSocketLink(SocketClient(GraphQLSocket(
+ '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)),
+ )));
+
+ runApp(MyApp(websocketLink));
+}
class MyApp extends StatelessWidget {
+ final WebSocketLink wsLink;
+ MyApp(this.wsLink);
+
@override
Widget build(BuildContext context) {
final HttpLink link = HttpLink(
@@ -22,7 +34,7 @@ class MyApp extends StatelessWidget {
final ValueNotifier<GraphQLClient> client = ValueNotifier<GraphQLClient>(
GraphQLClient(
cache: InMemoryCache(),
- link: link,
+ link: wsLink,
),
);
@@ -87,8 +99,7 @@ class _MyHomePageState extends State<MyHomePage> {
variables: <String, dynamic>{
'nRepositories': nRepositories,
},
- pollInterval: 4,
- // you can optionally override some http options through the contexts
+ pollInterval: 4, // you can optionally override some http options through the contexts
context: <String, dynamic>{
'headers': <String, String>{
'Authorization': 'Bearer $YOUR_PERSONAL_ACCESS_TOKEN',
@@ -107,15 +118,13 @@ 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) {
- final Map<String, dynamic> repository =
- repositories[index];
+ final Map<String, dynamic> repository = repositories[index];
return Mutation(
options: MutationOptions(
@@ -125,11 +134,8 @@ class _MyHomePageState extends State<MyHomePage> {
RunMutation addStar,
QueryResult addStarResult,
) {
- if (addStarResult.data != null &&
- addStarResult.data.isNotEmpty) {
- repository['viewerHasStarred'] =
- addStarResult.data['addStar']['starrable']
- ['viewerHasStarred'];
+ if (addStarResult.data != null && addStarResult.data.isNotEmpty) {
+ repository['viewerHasStarred'] = addStarResult.data['addStar']['starrable']['viewerHasStarred'];
}
return ListTile(

MODIFY
graphql.iml
graphql.iml
<excludeFolder url="file://$MODULE_DIR$/example/.dart_tool" />
<excludeFolder url="file://$MODULE_DIR$/example/.pub" />
<excludeFolder url="file://$MODULE_DIR$/example/build" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/.dart_tool" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/.pub" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/build" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/example/.dart_tool" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/example/.pub" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/.symlinks/plugins/path_provider/example/build" />
+ <excludeFolder url="file://$MODULE_DIR$/example/ios/Flutter/App.framework/flutter_assets/packages" />
</content>
<orderEntry type="jdk" jdkName="Android API 25 Platform" jdkType="Android SDK" />
<orderEntry type="sourceFolder" forTests="false" />

MODIFY
lib/graphql_flutter.dart
lib/graphql_flutter.dart
library graphql_flutter;
-export 'package:graphql_flutter/src/graphql_client.dart';
-export 'package:graphql_flutter/src/socket_client.dart';
-
+export 'package:graphql_flutter/src/cache/in_memory.dart';
+export 'package:graphql_flutter/src/cache/normalized_in_memory.dart';
+export 'package:graphql_flutter/src/core/graphql_error.dart';
export 'package:graphql_flutter/src/core/query_options.dart';
export 'package:graphql_flutter/src/core/query_result.dart';
-export 'package:graphql_flutter/src/core/graphql_error.dart';
-
-export 'package:graphql_flutter/src/link/link.dart';
+export 'package:graphql_flutter/src/graphql_client.dart';
export 'package:graphql_flutter/src/link/http/link_http.dart';
-
-export 'package:graphql_flutter/src/cache/in_memory.dart';
-export 'package:graphql_flutter/src/cache/normalized_in_memory.dart';
-
+export 'package:graphql_flutter/src/link/link.dart';
+export 'package:graphql_flutter/src/link/web_socket/link_web_socket.dart';
+export 'package:graphql_flutter/src/socket_client.dart';
export 'package:graphql_flutter/src/websocket/messages.dart';
export 'package:graphql_flutter/src/websocket/socket.dart';
-
-export 'package:graphql_flutter/src/widgets/graphql_provider.dart';
-export 'package:graphql_flutter/src/widgets/graphql_consumer.dart';
export 'package:graphql_flutter/src/widgets/cache_provider.dart';
-export 'package:graphql_flutter/src/widgets/query.dart';
+export 'package:graphql_flutter/src/widgets/graphql_consumer.dart';
+export 'package:graphql_flutter/src/widgets/graphql_provider.dart';
export 'package:graphql_flutter/src/widgets/mutation.dart';
+export 'package:graphql_flutter/src/widgets/query.dart';
export 'package:graphql_flutter/src/widgets/subscription.dart';

MODIFY
lib/src/link/web_socket/link_web_socket.dart
lib/src/link/web_socket/link_web_socket.dart
+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/websocket/messages.dart';
class WebSocketLink extends Link {
- RequestHandler subscriptionClient;
-
+ final SocketClient socketClient;
+ WebSocketLink(this.socketClient)
+ : super(
+ request: (
+ Operation operation, [
+ NextLink forward,
+ ]) {
+ return socketClient.subscribe(SubscriptionRequest(operation.operationName, operation.document, operation.variables)).map((result) {
+ return FetchResult(data: result.data, errors: result.errors, context: operation.getContext(), extensions: operation.extensions);
+ });
+ },
+ );
// TODO: implement https://github.com/apollographql/apollo-link/blob/master/packages/apollo-link-ws/src/webSocketLink.ts
}

MODIFY
lib/src/socket_client.dart
lib/src/socket_client.dart
import 'dart:async';
-import 'dart:io';
-
-import 'package:uuid/uuid.dart';
import 'package:graphql_flutter/graphql_flutter.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)});
+}
+
class SocketClient {
final Uuid _uuid = Uuid();
final GraphQLSocket _socket;
- static Map<String, String> _initPayload;
+ final SocketClientConfig config;
- SocketClient(this._socket) {
+ SocketClient(this._socket, {this.config = const SocketClientConfig(), Map<String, String> initPayload}) {
_socket.connectionAck.listen(print);
_socket.connectionError.listen(print);
_socket.unknownData.listen(print);
- _socket.write(InitOperation(_initPayload));
- }
- static Future<SocketClient> connect(
- final String endPoint, {
- final List<String> protocols = const <String>[
- 'graphql-ws',
- ],
- final Map<String, String> headers = const <String, String>{
- 'content-type': 'application/json',
- },
- final Map<String, String> initPayload,
- }) async {
- _initPayload = initPayload;
-
- return SocketClient(
- GraphQLSocket(
- await WebSocket.connect(
- endPoint,
- protocols: protocols,
- headers: headers,
- ),
- ),
- );
+ _socket.connectionState.listen((GraphQLSocketConnectionState state) {
+ print('WebSocket connection state changed to: $state');
+ if (state == GraphQLSocketConnectionState.CONNECTED) _socket.write(InitOperation(initPayload));
+ });
}
Stream<SubscriptionData> subscribe(final SubscriptionRequest payload) {
final String id = _uuid.v4();
- final StreamController<SubscriptionData> response =
- StreamController<SubscriptionData>();
+ final StreamController<SubscriptionData> response = StreamController<SubscriptionData>();
- final Stream<SubscriptionComplete> complete = _socket.subscriptionComplete
- .where((SubscriptionComplete message) => message.id == id)
- .take(1);
+ final Stream<SubscriptionComplete> complete = _socket.subscriptionComplete.where((SubscriptionComplete message) => message.id == id).take(1);
- final Stream<SubscriptionData> data = _socket.subscriptionData
- .where((SubscriptionData message) => message.id == id)
- .takeWhile((_) => !response.isClosed);
+ final Stream<SubscriptionData> data =
+ _socket.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);
+ final Stream<SubscriptionError> error =
+ _socket.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));
- response.onListen = () => _socket.write(StartOperation(id, payload));
- response.onCancel = () => _socket.write(StopOperation(id));
+ _socket.connectionState
+ .where((GraphQLSocketConnectionState state) => state == GraphQLSocketConnectionState.CONNECTED)
+ .takeWhile((_) => !response.isClosed)
+ .listen((_) => _socket.write(StartOperation(id, payload)));
+
+ // response.onListen = () => );
+ response.onCancel = () {
+ _socket.write(StopOperation(id));
+ };
return response.stream;
}

MODIFY
lib/src/websocket/messages.dart
lib/src/websocket/messages.dart
@@ -15,6 +15,7 @@ class MessageTypes {
// server connections
static const String GQL_CONNECTION_ACK = 'connection_ack';
static const String GQL_CONNECTION_ERROR = 'connection_error';
+ static const String GQL_CONNECTION_KEEP_ALIVE = 'ka';
// client operations
static const String GQL_START = 'start';
@@ -141,6 +142,16 @@ class ConnectionError extends GraphQLSocketMessage {
};
}
+/// The server will send this connection keep alive message periodically.
+class ConnectionKeepAlive extends GraphQLSocketMessage {
+ ConnectionKeepAlive() : super(MessageTypes.GQL_CONNECTION_KEEP_ALIVE);
+
+ @override
+ dynamic toJson() => <String, dynamic>{
+ 'type': type,
+ };
+}
+
/// Data sent from the server to the client with subscription data or error
/// payload. The user should check the errors result before processing the
/// data value. These error are from the query resolvers.
@@ -149,8 +160,7 @@ class SubscriptionData extends GraphQLSocketMessage {
final dynamic data;
final dynamic errors;
- SubscriptionData(this.id, this.data, this.errors)
- : super(MessageTypes.GQL_DATA);
+ SubscriptionData(this.id, this.data, this.errors) : super(MessageTypes.GQL_DATA);
@override
dynamic toJson() => <String, dynamic>{

MODIFY
lib/src/websocket/socket.dart
lib/src/websocket/socket.dart
@@ -2,48 +2,134 @@ import 'dart:async';
import 'dart:convert';
import 'dart:io';
-import '../../graphql_flutter.dart';
+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 StreamController<GraphQLSocketMessage> _subject =
- StreamController<GraphQLSocketMessage>.broadcast();
+ 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;
- final WebSocket _socket;
+ StreamSubscription _keepAliveSubscription;
- GraphQLSocket(this._socket) {
- _socket
- .map<Map<String, dynamic>>((dynamic message) => json.decode(message))
- .listen(
- (Map<String, dynamic> message) {
- final String type = message['type'] ?? 'unknown';
- final dynamic payload = message['payload'] ?? <String, dynamic>{};
- final String id = message['id'] ?? 'none';
+ 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:
- _subject.add(ConnectionAck());
- break;
+ return ConnectionAck();
case MessageTypes.GQL_CONNECTION_ERROR:
- _subject.add(ConnectionError(payload));
- break;
+ 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'];
- _subject.add(SubscriptionData(id, data, errors));
- break;
+ return SubscriptionData(id, data, errors);
case MessageTypes.GQL_ERROR:
- _subject.add(SubscriptionError(id, payload));
- break;
+ return SubscriptionError(id, payload);
case MessageTypes.GQL_COMPLETE:
- _subject.add(SubscriptionComplete(id));
- break;
+ return SubscriptionComplete(id);
default:
- _subject.add(UnknownData(message));
+ return UnknownData(map);
}
- },
- );
}
void write(final GraphQLSocketMessage message) {
@@ -55,27 +141,26 @@ class GraphQLSocket {
);
}
- Stream<ConnectionAck> get connectionAck => _subject.stream
- .where((GraphQLSocketMessage message) => message is ConnectionAck)
- .cast<ConnectionAck>();
+ 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<ConnectionError> get connectionError => _subject.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 => _subject.stream
- .where((GraphQLSocketMessage message) => message is UnknownData)
- .cast<UnknownData>();
+ Stream<UnknownData> get unknownData =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is UnknownData).cast<UnknownData>();
- Stream<SubscriptionData> get subscriptionData => _subject.stream
- .where((GraphQLSocketMessage message) => message is SubscriptionData)
- .cast<SubscriptionData>();
+ Stream<SubscriptionData> get subscriptionData =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionData).cast<SubscriptionData>();
- Stream<SubscriptionError> get subscriptionError => _subject.stream
- .where((GraphQLSocketMessage message) => message is SubscriptionError)
- .cast<SubscriptionError>();
+ Stream<SubscriptionError> get subscriptionError =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionError).cast<SubscriptionError>();
- Stream<SubscriptionComplete> get subscriptionComplete => _subject.stream
- .where((GraphQLSocketMessage message) => message is SubscriptionComplete)
- .cast<SubscriptionComplete>();
+ Stream<SubscriptionComplete> get subscriptionComplete =>
+ _messageController.stream.where((GraphQLSocketMessage message) => message is SubscriptionComplete).cast<SubscriptionComplete>();
}

Preparing socket connection auto reconnect and keep alive.