新增websocket客户端

lib/EasyDartModule.dart

@@ -12,7 +12,9 @@ import 'package:EasyDartModule/base/redis/redis.dart';
 import 'package:EasyDartModule/base/storage/Storage.dart';
 import 'package:EasyDartModule/base/storage/impl/MinIoStorage.dart';
 import 'package:EasyDartModule/base/webserver/WebServer.dart';
-import 'package:EasyDartModule/base/webserver/impl/ShelfWebServer.dart' if (dart.library.html) 'package:EasyDartModule/base/webserver/impl/ShelfWebServer_none.dart';
+import 'package:EasyDartModule/base/webserver/impl/ShelfWebServer.dart'
+    if (dart.library.html) 'package:EasyDartModule/base/webserver/impl/ShelfWebServer_none.dart';
+import 'package:EasyDartModule/base/websocket/WebSocket.dart';
 
 export 'package:shelf/shelf.dart';
 export 'package:mongo_dart/mongo_dart.dart';
@@ -26,14 +28,17 @@ class EasyDartModule {
   static Mqtt get mqtt => Mqtt.getInstance();
   static Storage get storage => Storage.getInstance();
   static Redis get redis => Redis.getInstance();
+  static WebSocket get websocket => WebSocket.getInstance();
 
-  static bool init(
+  static bool init({
-      {DiscoveryConfig? discoveryConfig,
+    DiscoveryConfig? discoveryConfig,
-      DataBaseConfig? dataBaseConfig,
+    DataBaseConfig? dataBaseConfig,
-      LoggerConfig? loggerConfig,
+    LoggerConfig? loggerConfig,
-      MqttConfig? mqttConfig,
+    MqttConfig? mqttConfig,
-      StorageConfig? storageConfig,
+    StorageConfig? storageConfig,
-      RedisConfig? redisConfig}) {
+    RedisConfig? redisConfig,
+    WebSocketConfig? webSocketConfig,
+  }) {
     if (discoveryConfig != null) {
       //nacos注册配置中心
       Discovery.setInstance(NacosDiscovery(discoveryConfig));
@@ -63,6 +68,10 @@ class EasyDartModule {
       //Dio组件
       TraceDio.setInstance(TraceDio(logger));
     }
+    if (webSocketConfig != null) {
+      WebSocket.setInstance(WebSocket(webSocketConfig));
+      websocket.connect();
+    }
     return true;
   }
 }

lib/base/websocket/WebSocket.dart

@@ -0,0 +1,78 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:web_socket_channel/web_socket_channel.dart';
+
+class WebSocket {
+  static late WebSocket _webSocket;
+  WebSocketConfig _config;
+  WebSocketChannel? _channel;
+  bool _reConnect = true;
+
+  static WebSocket getInstance() {
+    return _webSocket;
+  }
+
+  static void setInstance(WebSocket webSocket) {
+    _webSocket = webSocket;
+  }
+
+  WebSocket(this._config);
+  void reConnect() {
+    //判断是否启用自动重连
+    if (_config.reConnect && _reConnect) {
+      //2秒后重连
+      Future.delayed(Duration(seconds: 2), () {
+        connect();
+      });
+    }
+  }
+
+  Future<void> connect() async {
+    final wsUrl = Uri.parse(_config.host);
+    final channel = WebSocketChannel.connect(wsUrl);
+    try {
+      await channel.ready;
+    } catch (e) {
+      print(e);
+      //连接失败
+      reConnect();
+      return;
+    }
+    _channel = channel;
+    _reConnect = _config.reConnect;
+    Timer? _timer;
+    channel.stream.listen((message) {
+      // channel.sink.add('received!');
+      // channel.sink.close(goingAway);
+      _config.messgae(message);
+    }, onDone: () {
+      _timer?.cancel();
+      _timer = null;
+      //尝试重连
+      _channel = null;
+      reConnect();
+    });
+    //定期发送心跳包
+    _timer = Timer.periodic(Duration(seconds: 30), (t) {
+      sendData(jsonEncode({"ht": "ht"}));
+    });
+  }
+
+  void close() {
+    _reConnect = false;
+    _channel?.sink.close();
+  }
+
+  void sendData(String data) {
+    _channel?.sink.add(data);
+  }
+}
+
+class WebSocketConfig {
+  String host;
+  bool reConnect;
+  void Function(String) messgae;
+
+  WebSocketConfig(this.host, this.messgae, {this.reConnect = true});
+}

pubspec.yaml

@@ -20,6 +20,7 @@ dependencies:
   minio: ^3.5.7
   redis: ^4.0.0
   dart_jsonwebtoken: ^2.14.2
+  web_socket_channel: ^3.0.1
 
 dev_dependencies:
   lints: ^4.0.0