Livequery reconnecting controller (#316)

* Fix: reconnecting

Done as in #315 (comment) described.

* Changed type of Future

Done as described in fischerscode@805e830#r37587405.

* Revert "Changed type of Future"

This reverts commit fecee76.

* adding LiveQueryReconnectingController

This is a solution for #315 (comment)

How it works:
- Only the LiveQueryReconnectingController handles reconnecting after a connection loss.
- LiveQueryReconnectingController holds informations:
  1. connection state of the device
  2. did the user disconnected from the server
  3. is currently a connection to the server estalished
- LiveQueryReconnectingController tries to reconnect after preset timespans.

* cleanup

* remove _userDisconnected from Client

This is not needed any more.
Keeping track of this information is now done by LiveQueryReconnectingController.

* subscribe(query) modified

Wait until client is connected to live-query-server.

Fixes fischerscode@d058eb0#r37612749

Author: fischerscode

lib/src/network/parse_live_query.dart

@@ -32,9 +32,87 @@ class Subscription {
   }
 }
 
-enum LiveQueryClientEvent { CONNECTED, DISCONNECTED }
+enum LiveQueryClientEvent { CONNECTED, DISCONNECTED, USER_DISCONNECTED }
 
-class Client with WidgetsBindingObserver {
+class LiveQueryReconnectingController with WidgetsBindingObserver {
+  // -1 means "do not try to reconnect",
+  static const List<int> retryInterval = [0, 500, 1000, 2000, 5000, 10000];
+  static const String DEBUG_TAG = 'LiveQueryReconnectingController';
+
+  final Function _reconnect;
+  final Stream<LiveQueryClientEvent> _eventStream;
+  final bool debug;
+
+  int _retryState = 0;
+  bool _isOnline = false;
+  bool _isConnected = false;
+  bool _userDisconnected = false;
+
+  Timer _currentTimer;
+
+  LiveQueryReconnectingController(
+      this._reconnect, this._eventStream, this.debug) {
+    Connectivity().onConnectivityChanged.listen((ConnectivityResult state) {
+      if (!_isOnline && state != ConnectivityResult.none) _retryState = 0;
+      _isOnline = state != ConnectivityResult.none;
+      if (debug) print('$DEBUG_TAG: $state');
+      _setReconnect();
+    });
+    _eventStream.listen((LiveQueryClientEvent event) {
+      switch (event) {
+        case LiveQueryClientEvent.CONNECTED:
+          _isConnected = true;
+          _retryState = 0;
+          _userDisconnected = false;
+          break;
+        case LiveQueryClientEvent.DISCONNECTED:
+          _isConnected = false;
+          _setReconnect();
+          break;
+        case LiveQueryClientEvent.USER_DISCONNECTED:
+          _userDisconnected = true;
+          if (_currentTimer != null) {
+            _currentTimer.cancel();
+            _currentTimer = null;
+          }
+          break;
+      }
+
+      if (debug) print('$DEBUG_TAG: $event');
+    });
+    WidgetsBinding.instance.addObserver(this);
+  }
+
+  @override
+  void didChangeAppLifecycleState(AppLifecycleState state) {
+    switch (state) {
+      case AppLifecycleState.resumed:
+        _setReconnect();
+        break;
+      default:
+        break;
+    }
+  }
+
+  void _setReconnect() {
+    if (_isOnline &&
+        !_isConnected &&
+        _currentTimer == null &&
+        !_userDisconnected &&
+        retryInterval[_retryState] >= 0) {
+      _currentTimer =
+          Timer(Duration(milliseconds: retryInterval[_retryState]), () {
+        _currentTimer = null;
+        _reconnect();
+      });
+      if (debug)
+        print('$DEBUG_TAG: Retrytimer set to ${retryInterval[_retryState]}ms');
+      if (_retryState < retryInterval.length - 1) _retryState++;
+    }
+  }
+}
+
+class Client {
   factory Client() => _getInstance();
   Client._internal(
       {bool debug, ParseHTTPClient client, bool autoSendSessionId}) {
@@ -57,14 +135,9 @@ class Client with WidgetsBindingObserver {
     } else if (_liveQueryURL.contains('http')) {
       _liveQueryURL = _liveQueryURL.replaceAll('http', 'ws');
     }
-    Connectivity().onConnectivityChanged
-        .listen((ConnectivityResult connectivityResult) {
-      print('onConnectivityChanged:$connectivityResult');
-      if (connectivityResult != ConnectivityResult.none) {
-        reconnect();
-      }
-    });
-    WidgetsBinding.instance.addObserver(this);
+
+    reconnectingController = LiveQueryReconnectingController(
+        () => reconnect(userInitialized: false), getClientEventStream, _debug);
   }
   static Client get instance => _getInstance();
   static Client _instance;
@@ -85,15 +158,15 @@ class Client with WidgetsBindingObserver {
   bool _sendSessionId;
   WebSocketChannel _channel;
   String _liveQueryURL;
-  bool _userDisconnected = false;
   bool _connecting = false;
   StreamController<LiveQueryClientEvent> _clientEventStreamController;
   Stream<LiveQueryClientEvent> _clientEventStream;
+  LiveQueryReconnectingController reconnectingController;
 
   final Map<int, Subscription> _requestSubScription = <int, Subscription>{};
 
-  Future<void> reconnect() async {
-    await _connect();
+  Future<void> reconnect({bool userInitialized = false}) async {
+    await _connect(userInitialized: userInitialized);
     _connectLiveQuery();
   }
 
@@ -104,19 +177,7 @@ class Client with WidgetsBindingObserver {
     return WebSocket.connecting;
   }
 
-  @override
-  void didChangeAppLifecycleState(AppLifecycleState state) {
-    super.didChangeAppLifecycleState(state);
-    switch (state) {
-      case AppLifecycleState.resumed: 
-        reconnect();
-        break;
-      default:
-        break;
-    }
-  }
-
-  Future<dynamic> disconnect() async {
+  Future<dynamic> disconnect({bool userInitialized = false}) async {
     if (_webSocket != null && _webSocket.readyState == WebSocket.open) {
       if (_debug) {
         print('$_printConstLiveQuery: Socket closed');
@@ -134,13 +195,16 @@ class Client with WidgetsBindingObserver {
     _requestSubScription.values.toList().forEach((Subscription subcription) {
       subcription._enabled = false;
     });
-    _userDisconnected = true;
     _connecting = false;
+    if (userInitialized)
+      _clientEventStreamController.sink
+          .add(LiveQueryClientEvent.USER_DISCONNECTED);
   }
 
   Future<Subscription> subscribe(QueryBuilder query) async {
     if (_webSocket == null) {
-      await reconnect();
+      await _clientEventStream.any((LiveQueryClientEvent event) =>
+          event == LiveQueryClientEvent.CONNECTED);
     }
     final int requestId = _requestIdGenerator();
     final Subscription subscription = Subscription(query, requestId);
@@ -173,17 +237,17 @@ class Client with WidgetsBindingObserver {
     return _requestIdCount++;
   }
 
-  Future<dynamic> _connect() async {
+  Future<dynamic> _connect({bool userInitialized = false}) async {
     if (_connecting) {
       print('already connecting');
       return Future<void>.value(null);
     }
-    await disconnect();
+    await disconnect(userInitialized: userInitialized);
     _connecting = true;
-    _userDisconnected = false;
 
     try {
       _webSocket = await WebSocket.connect(_liveQueryURL);
+      _connecting = false;
       if (_webSocket != null && _webSocket.readyState == WebSocket.open) {
         if (_debug) {
           print('$_printConstLiveQuery: Socket opened');
@@ -200,18 +264,12 @@ class Client with WidgetsBindingObserver {
       }, onDone: () {
         _clientEventStreamController.sink
             .add(LiveQueryClientEvent.DISCONNECTED);
-        if (!_userDisconnected) {
-          reconnect();
-        }
         if (_debug) {
           print('$_printConstLiveQuery: Done');
         }
       }, onError: (Object error) {
         _clientEventStreamController.sink
             .add(LiveQueryClientEvent.DISCONNECTED);
-        if (!_userDisconnected) {
-          reconnect();
-        }
         if (_debug) {
           print(
               '$_printConstLiveQuery: Error: ${error.runtimeType.toString()}');
@@ -220,6 +278,7 @@ class Client with WidgetsBindingObserver {
             ParseApiRQ.liveQuery, _debug, 'IOWebSocketChannel'));
       });
     } on Exception catch (e) {
+      _connecting = false;
       _clientEventStreamController.sink.add(LiveQueryClientEvent.DISCONNECTED);
       if (_debug) {
         print('$_printConstLiveQuery: Error: ${e.toString()}');
@@ -299,7 +358,6 @@ class Client with WidgetsBindingObserver {
     final Map<String, dynamic> actionData = jsonDecode(message);
     Subscription subscription;
     if (actionData.containsKey('op') && actionData['op'] == 'connected') {
-      _connecting = false;
       print('ReSubScription:$_requestSubScription');
       _requestSubScription.values.toList().forEach((Subscription subcription) {
         _subscribeLiveQuery(subcription);