flutter websockets autoreconnect - 如何实现

Question

我正在努力如何websockets在颤振中实现自动重新连接。我使用web_socket_channel，但是，该插件只是包装了dart.io WebSocket，因此任何基于WebSocket类的解决方案也适用于我。

我已经想通了，如何捕获套接字断开连接，请参阅下面的代码片段：

    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');
    }

我想IOWebSocketChannel.connect从inside 调用onDone，但是，这会导致一种无限循环 - 因为我必须再次关闭_channel先前的调用connect，这又会onDone再次调用等等。

任何帮助将不胜感激！

Answer 1

使用 package:web_socket_channel (IOWebSocketChannel) 没有任何方法可以实现套接字连接的重新连接。但是您可以使用 WebSocket 类来实现可重新连接的连接。

您可以实现 WebSocket 通道，然后使用 StreamController 类广播消息。工作示例：

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();
  }
}

因为通知控制器返回一个单例实例，那么服务器和设备之间总会有一个 Socket 连接。并且通过StreamController的broadcast方法，我们可以在多个消费者之间共享Websocket发送的数据

var _streamController = new NotificationController().streamController;

_streamController.stream.listen(pushNotifications);

Answer 2

大多数时候，当我们创建一个 时WebSocketChannel，我们将使用它stream来接收消息和sink发送消息。

重新连接的想法是当错误发生或套接字关闭时，我们将创建一个新WebSocketChannel实例并将其分配给全局共享变量。但困难的是其他地方使用它的stream&sink将会无效。

为了克服这一问题，我们将创建一个固定的stream&sink来转发和传输与新实例等效的消息WebSocketChannel。

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

这就是我所做的：

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

然后要启动您的 websocket，只需初始调用 reconnect() 即可。基本上，它的作用是在调用 onDone 回调时重新创建 WebSocket，这会在连接被销毁时发生。所以，连接被破坏了——好吧，让我们自动重新连接。我还没有找到一种方法可以在不重新创建 _channel 的情况下执行此操作。就像理想情况下，会有一个 _channel.connect() 可以重新连接到现有的 URL，或者某种自动重新连接功能，但这似乎并不存在。

哦，这里有一些更好的东西，可以在远程服务器关闭时消除丑陋的重新连接异常回溯，并添加 4 秒的重新连接延迟。在这种情况下，cancelOnError 参数会在出现任何错误时触发套接字关闭。

  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);
  }