o2oa-flutter/lib/common/utils/webscoket_util.dart

import 'dart:async';
 
 
import 'package:get/get.dart';
import 'package:local_notifier/local_notifier.dart';
import 'package:web_socket_channel/io.dart';
import 'package:web_socket_channel/status.dart' as status;

import '../models/im/index.dart';
import '../values/index.dart';
import 'index.dart';

enum O2WebsocketStatus {
  SocketStatusConnected, // 已连接
  SocketStatusFailed, // 失败
  SocketStatusClosed, // 连接关闭
}

class O2WebsocketClient {
  static final O2WebsocketClient _instance = O2WebsocketClient._internal();

  factory O2WebsocketClient() {
    return _instance;
  }

  
  O2WebsocketClient._internal() {
    //初始化
  }

  final heartBeatKey = 'heartbeat';
  IOWebSocketChannel? _channel;
  var _status = O2WebsocketStatus.SocketStatusConnected;
  Timer? _heartBeat; // 心跳定时器
  final int _heartTimes = 3000; // 心跳间隔(毫秒)
  final int _reconnectCount = 5; // 重连次数，默认5次
  int _reconnectTimes = 0; // 重连计数器
  Timer? _reconnectTimer; // 重连定时器
  var _isReconnect = true;

  final eventBus = EventBus();
  final channelUtil = O2FlutterMethodChannelUtils();


  ///
  /// 启动websocket
  /// 
  void open({reopen = false}) {
    String? url = O2ApiManager.instance.websocketUrl();
    OLogger.i("启动websocket ， url： $url");
    if (url == null || url.isEmpty) {
      OLogger.e('没有获取到websocket地址，无法连接！');
      return;
    }
    try {
      _closeSocket(false);
      _channel = IOWebSocketChannel.connect(Uri.parse(url));
      _status = O2WebsocketStatus.SocketStatusConnected;
      // 连接成功，重置重连计数器
      if (!reopen) {
        _reconnectTimes = 0;
      }
      if (_reconnectTimer != null) {
        _reconnectTimer?.cancel();
        _reconnectTimer = null;
      }
      _initHeartBeat();
      _channel?.stream.listen(_webSocketOnMessage, onDone: _webSocketOnDone, onError: _webSocketOnError);
    }catch(e) {
      OLogger.e(e);
    }
  }

  ///
  /// 关闭websokect
  /// 
  void stop() {
    _closeSocket(false);
  }

  ///
  /// 发送websocket消息
  /// 
  void sendMsg(String msg) {
    if (_channel != null) {
      switch(_status) {
        case O2WebsocketStatus.SocketStatusClosed:
        OLogger.i(" websokect 连接已关闭！");
        break;
        case O2WebsocketStatus.SocketStatusFailed:
        OLogger.i(" websokect 连接失败！");
        break;
        case O2WebsocketStatus.SocketStatusConnected:
        _channel?.sink.add(msg);
        break;
        default:
        break;
      }
    }
  }

  


  /// WebSocket接收消息回调
  _webSocketOnMessage(event) {
    if (!(event is String && event == heartBeatKey)) {
      OLogger.i("接收到 websokect 消息，$event");
      Map<String, dynamic> msg = O2Utils.parseStringToJson(event);
      if (msg["type"] != null) {
        // 聊天消息
        if (O2.o2MessageTypeIMCreate == msg["type"]) {
          _imCreateMessageNotify(msg);
        } else if (O2.o2MessageTypeIMRevoke == msg["type"]) {
          _imRevokeMessageNotify(msg);
        } else if (O2.o2MeesageTypeIMConversationUpdate == msg['type']) {
          _imConversationUpdateNotify(msg);
        } else if (O2.o2MeesageTypeIMConversationUpdate == msg['type']) {
          _imConversationDeleteNotify(msg);
        }
        
      }
    }
  }


  _imConversationDeleteNotify(Map<String, dynamic> msg) {
    final body = msg["body"];
    if (body != null) {
      IMConversationInfo conv = IMConversationInfo.fromJson(body);
      eventBus.emit(EventBus.websocketImConversationDeleteMsg, conv);
    }

  }
  _imConversationUpdateNotify(Map<String, dynamic> msg) {
    final body = msg["body"];
    if (body != null) {
      IMConversationInfo conv = IMConversationInfo.fromJson(body);
      eventBus.emit(EventBus.websocketImConversationUpdateMsg, conv);
    }
  }

  /// 刷新界面
  _imRevokeMessageNotify(Map<String, dynamic> msg) {
    var body = msg["body"];
    if (body == null) {
      return;
    }
    IMMessage message = IMMessage.fromJson(body);
    eventBus.emit(EventBus.websocketRevokeImMsg, message);
  }
  ///
  /// 聊天消息 
  /// 
  /// 通知界面刷新消息
  /// pc端发送消息通知
  _imCreateMessageNotify(Map<String, dynamic> msg) {
    var body = msg["body"];
    if (body == null) {
      return;
    }
    IMMessage message = IMMessage.fromJson(body);
    eventBus.emit(EventBus.websocketCreateImMsg, message);
    // macos 上发送消息通知
    if (GetPlatform.isMacOS ) {
      channelUtil.sendIMMsgToPCNotify(message.id ,msg["title"] ?? '', message.toBody()?.conversationBodyString() ?? '');
    } else if (GetPlatform.isWindows) {
      LocalNotification notification = LocalNotification(
        title: 'appName'.tr,
        body: '${msg["title"] ?? ''}: ${message.toBody()?.conversationBodyString() ?? ''}',
      );
      notification.show();
      // channelUtil.sendIMMsgToPCNotify(message.id ,msg["title"] ?? '', message.toBody()?.conversationBodyString() ?? '');
    }
  }
 
  /// WebSocket关闭连接回调
  _webSocketOnDone() {
    OLogger.d("onDone , websocket done ！！！");
    if (_isReconnect) {
      _reconnect();
    }
  }

  /// WebSocket连接错误回调
  _webSocketOnError(err) {
    OLogger.e("onError, websocket error", err);
    _status = O2WebsocketStatus.SocketStatusFailed;
    _closeSocket(true);
  }


  /// 初始化心跳
  void _initHeartBeat() {
    _destroyHeartBeat();
    _heartBeat = Timer.periodic(Duration(milliseconds: _heartTimes), (timer) {
      _sentHeart();
    });
  }

  /// 心跳
  void _sentHeart() {
    sendMsg('heartbeat');
  }

  /// 重连机制
  void _reconnect() {
    if (_reconnectTimes < _reconnectCount) {
      _reconnectTimes++;
      _reconnectTimer = Timer.periodic(Duration(milliseconds: _heartTimes), (timer) {
        open(reopen: true);
      });
    } else {
      if (_reconnectTimer != null) {
        OLogger.i('重连次数超过最大次数');
        _reconnectTimer?.cancel();
        _reconnectTimer = null;
      }
      return;
    }
  }

  /// 销毁心跳
  void _destroyHeartBeat() {
    if (_heartBeat != null) {
      _heartBeat?.cancel();
      _heartBeat = null;
    }
  }

   /// 关闭WebSocket
  void _closeSocket(bool isReconnect) {
    if (_channel != null && _status != O2WebsocketStatus.SocketStatusClosed) {
      OLogger.i('WebSocket连接关闭');
      _isReconnect = isReconnect;
      _channel?.sink.close(status.goingAway);
      _destroyHeartBeat();
      _status = O2WebsocketStatus.SocketStatusClosed;
    }
  }

}