flutter websockets autoreconnect - how to implement

Question

I am struggling how to implement websockets autoreconnect in flutter. I use web_socket_channel, however, the plugin just wraps dart.io WebSocket, hence any solution based on WebSocket class will work for me as well.

I already figured out, how to catch the socket disconnection, see the code snippet below:

    try {
      _channel = IOWebSocketChannel.connect(
        wsUrl,
      );

      ///
      /// Start listening to new notifications / messages
      ///
      _channel.stream.listen(
        _onMessageFromServer,
        onDone: () {
          debugPrint('ws channel closed');
        },
        onError: (error) {
          debugPrint('ws error $error');
        },
      );
    } catch (e) {
      ///
      /// General error handling
      /// TODO handle connection failure
      ///
      debugPrint('Connection exception $e');
    }

I was thinking to call IOWebSocketChannel.connect from within onDone, however, this leads to a kind of infinite loop - since I have to close the _channel prior calling connect again, this on its turn calls onDone again and so on.

Any help would be greatly appreciated!

Answer 1

With the package:web_socket_channel (IOWebSocketChannel) there is not any way in order to implement reconnection for the socket connections. But you can use WebSocket class in order to implement a reconnectable connection.

You can implement the WebSocket channel and then broadcast messages with StreamController class. Working example:

import 'dart:async';
import 'dart:io';

class NotificationController {

  static final NotificationController _singleton = new NotificationController._internal();

  StreamController<String> streamController = new StreamController.broadcast(sync: true);

  String wsUrl = 'ws://YOUR_WEBSERVICE_URL';

  WebSocket channel;

  factory NotificationController() {
    return _singleton;
  }

  NotificationController._internal() {
    initWebSocketConnection();
  }

  initWebSocketConnection() async {
    print("conecting...");
    this.channel = await connectWs();
    print("socket connection initializied");
    this.channel.done.then((dynamic _) => _onDisconnected());
    broadcastNotifications();
  }

  broadcastNotifications() {
    this.channel.listen((streamData) {
      streamController.add(streamData);
    }, onDone: () {
      print("conecting aborted");
      initWebSocketConnection();
    }, onError: (e) {
      print('Server error: $e');
      initWebSocketConnection();
    });
  }

  connectWs() async{
    try {
      return await WebSocket.connect(wsUrl);
    } catch  (e) {
      print("Error! can not connect WS connectWs " + e.toString());
      await Future.delayed(Duration(milliseconds: 10000));
      return await connectWs();
    }

  }

  void _onDisconnected() {
    initWebSocketConnection();
  }
}

Because the notification controller returns a singleton instance, then there will be always one Socket connection between the server and device. And with the broadcast method of StreamController, we can share the data sent by Websocket between multiple consumers

var _streamController = new NotificationController().streamController;

_streamController.stream.listen(pushNotifications);

Answer 2

Most of the time, when we create a WebSocketChannel, we will use its stream to receive messages and sink to send messages.

The idea to reconnect is when the error happens or the socket is closed, we will create a new WebSocketChannel instance and assign it into a global shared var. But the hard thing is other places where use its stream & sink will be invalid.

To overcome this, we will create a fixed stream & sink to forward & transfer messages with the equivalent one of the new WebSocketChannel instance.

class AutoReconnectWebSocket {
  final Uri _endpoint;
  final int delay;
  final StreamController<dynamic> _recipientCtrl = StreamController<dynamic>();
  final StreamController<dynamic> _sentCtrl = StreamController<dynamic>();

  WebSocketChannel? webSocketChannel;

  get stream => _recipientCtrl.stream;

  get sink => _sentCtrl.sink;

  AutoReconnectWebSocket(this._endpoint, {this.delay = 5}) {
    _sentCtrl.stream.listen((event) {
      webSocketChannel!.sink.add(event);
    });
    _connect();
  }

  void _connect() {
    webSocketChannel = WebSocketChannel.connect(_endpoint);
    webSocketChannel!.stream.listen((event) {
      _recipientCtrl.add(event);
    }, onError: (e) async {
      _recipientCtrl.addError(e);
      await Future.delayed(Duration(seconds: delay));
      _connect();
    }, onDone: () async {
      await Future.delayed(Duration(seconds: delay));
      _connect();
    }, cancelOnError: true);
  }
}

Answer 3

Here's what I do:

void reconnect() {
    setState(() {
      _channel = IOWebSocketChannel.connect(wsUrl);
    });
    _channel.stream.listen((data) => processMessage(data), onDone: reconnect);
  }

Then to start up your websocket, just do an initial call to reconnect(). Basically what this does is re-create your WebSocket when the onDone callback is called, which happens when the connection is destroyed. So, connection destroyed -- ok, let's reconnect automatically. I haven't found a way to do this without re-creating _channel. Like, ideally, there would be a _channel.connect() that would reconnect to the existing URL, or some kind of auto-reconnect feature, but that doesn't seem to exist.

Oh, here's something a bit better that will get rid of ugly reconnect exception tracebacks if the remote server is down, and add a reconnect delay of 4 seconds. In this case, the cancelOnError argument triggers socket closure on any error.

  wserror(err) async {
    print(new DateTime.now().toString() + " Connection error: $err");
    await reconnect();
  }

 reconnect() async {
    if (_channel != null) {
      // add in a reconnect delay
      await Future.delayed(Duration(seconds: 4));
    }
    setState(() {
      print(new DateTime.now().toString() + " Starting connection attempt...");
      _channel = IOWebSocketChannel.connect(wsUrl);
      print(new DateTime.now().toString() + " Connection attempt completed.");
    });
    _channel.stream.listen((data) => processMessage(data), onDone: reconnect, onError: wserror, cancelOnError: true);
  }

Answer 4

I recommend you to use this multiplatform websocket package https://pub.dev/packages/websocket_universal . There you can configure SocketConnectionOptions with [timeoutConnectionMs] , [failedReconnectionAttemptsLimit] and even [maxReconnectionAttemptsPerMinute]. And that package allows you to listen to webSocket events

Answer 5

This is how I'm doing it:

class MyAppState extends ChangeNotifier {
  WebSocketChannel? channel;
  bool webSocketConnected = false;
  int webSocketReconnectAttempts = 0;

  MyAppState() {
    connect();
  }

  void onMessage(message) {
    webSocketConnected = true;
    webSocketReconnectAttempts = 0;
    notifyListeners();
  }

  void onDone() async {
    var delay = 1 + 1 * webSocketReconnectAttempts;
    if (delay > 10) {
      delay = 10;
    }
    print(
        "Done, reconnecting in $delay seconds, attempt $webSocketReconnectAttempts ");
    webSocketConnected = false;
    channel = null;
    await Future.delayed(Duration(seconds: delay));
    connect();
  }

  void onError(error) {
    print(error);
    if (error is WebSocketChannelException) {
      webSocketReconnectAttempts += 1;
    }
  }

  void connect() {
    try {
      channel = WebSocketChannel.connect(
        Uri.parse('ws://localhost:8000/ws/generate/'),
      );
      channel!.stream.listen(onMessage, onDone: onDone, onError: onError);
    } catch (e) {
      print(e);
    }
  }
}

I'm not claiming this to be the best or even the most efficient, but it's reliable for me so far. I'm still learning Flutter myself, so I hope it helps.

If you want a full example with django channels as the host you can see that code here: https://github.com/BillSchumacher/HardDiffusion

The flutter code is in the hard_diffusion directory.

If you reconnect onError without having it close on error, I think that could be a problem.

I'm sending the client a message when I accept their connection as well.