fis_lib_jsonrpc/lib/notifications/connection.dart

import 'dart:async';

import 'package:fis_common/event/event_type.dart';
import 'package:fis_common/logger/logger.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/status.dart' as statuses;

import 'core/interface/connection.dart';

class WsConnectionException implements Exception {
  final dynamic message;

  WsConnectionException(this.message);
}

/// WebScoket连接
class WsConnection implements IConnection {
  /// 错误断开重试次数
  static const C_ERR_RETRY_LIMIT = 5;

  /// 连接请求超时时间，10s
  static const C_CONNECT_REQ_TIMEOUT = 10 * 1000;

  ConnectionStatus _status = ConnectionStatus.connectedClosed;

  int _errorRetryCount = 0;

  bool _keepAlive = false;

  bool _hasError = false;

  /// 主机地址
  final String host;

  WebSocketChannel? _channel;

  Timer? _timer;

  WsConnection(
    this.host, {
    this.exceptionOccurred,
    this.statusChanged,
  }) {
    messageReceived = FEventHandler<dynamic>();
  }

  /// 连接状态
  ConnectionStatus get status => _status;

  /// 是否已连接
  bool get isConnected =>
      _channel != null && status == ConnectionStatus.connected;

  bool get isKeepAlive => _keepAlive;

  @override
  FEventHandler<Exception>? exceptionOccurred;

  @override
  late final FEventHandler<dynamic> messageReceived;

  @override
  FEventHandler<ConnectionStatus>? statusChanged;

  @override
  Future<bool> connect() async {
    logger.i("WsConnection is trying connect...");
    _hasError = false;
    _keepAlive = true;
    final uri = Uri.parse(host);
    _channel = WebSocketChannel.connect(uri);
    if (_channel == null) {
      logger.i("The ws connect failed.");
      return false;
    }

    final completer = Completer<bool>();
    Future.delayed(const Duration(milliseconds: 500), () {
      if (completer.isCompleted || _hasError) return;

      _updateStatus(ConnectionStatus.connected);
      completer.complete(true);

      logger.i("The ws connection is connected.");
      _startCheckConnection();
    });

    _channel!.stream.listen(
      (data) {
        messageReceived.emit(this, data);
      },
      onError: (error) {
        _hasError = true;
        _handleCancelOnError(error);
        try {
          if (!completer.isCompleted) {
            completer.complete(false);
          }
        } catch (e) {}
      },
      cancelOnError: true,
      onDone: () {},
    );

    return completer.future;
  }

  @override
  Future<bool> close() async {
    logger.i("WsConnection is closing...");
    _keepAlive = false;
    _stopCheckConnection();

    if (_channel != null) {
      if (_channel!.closeCode == null) {
        await _channel!.sink.close(statuses.normalClosure);
      }
      _channel = null;
    }
    _updateStatus(ConnectionStatus.connectedClosed);
    logger.i("The ws connection is disconnected.");
    return true;
  }

  void _updateStatus(ConnectionStatus value) {
    if (value != _status) {
      _status = value;
      statusChanged?.emit(this, value);
    }
  }

  void _startCheckConnection() {
    _timer?.cancel();

    _timer = Timer.periodic(
      const Duration(milliseconds: 500),
      (timer) {
        if (isConnected) {
          if (_channel!.closeCode != null) {
            _onDisconnected();
          }
        }
      },
    );
  }

  void _stopCheckConnection() {
    _timer?.cancel();
    _timer = null;
  }

  void _onDisconnected() {
    _updateStatus(ConnectionStatus.connectedClosed);
    _channel = null;
    logger.i("WsConnection done, the ws connection is disconnected.");

    _tryReconnect();
  }

  /// 处理因错误导致的连接中断
  void _handleCancelOnError(error) async {
    exceptionOccurred?.emit(this, WsConnectionException(error));

    _updateStatus(ConnectionStatus.connectedClosed);
    logger.e("WsConnection error, the ws connection is disconnected.", error);

    _tryReconnect();
  }

  Future<void> _tryReconnect() async {
    _stopCheckConnection();

    if (_keepAlive) {
      // 延迟1000ms后尝试再次连接
      await Future.delayed(const Duration(milliseconds: 1000));
      logger.i(
          "WsConnection is now keeping alive and trying to reconnect for $_errorRetryCount time.");
      connect();
      _errorRetryCount++;
    }
  }
}