flutter websockets autoreconnect - 如何实现

Question

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

我已经想通了，如何捕捉socket断开，看下面的sn-p代码：

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

我想从onDone 内部调用IOWebSocketChannel.connect，但是，这会导致一种无限循环 - 因为我必须在再次调用connect 之前关闭_channel，这轮到它调用@ 987654330@再次等等。

任何帮助将不胜感激！

Answer 1

这是我的工作：

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

Answer 2

使用 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的广播方式，我们可以在多个消费者之间共享Websocket发送的数据

var _streamController = new NotificationController().streamController;

_streamController.stream.listen(pushNotifications);

Answer 3

大多数时候，当我们创建一个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);
  }
}