import 'dart:async'; import 'dart:convert'; import 'dart:core'; import 'dart:core'; import 'dart:io'; import 'package:flutter/foundation.dart'; import 'package:flutter/material.dart'; import 'package:flutter_smart_dialog/flutter_smart_dialog.dart'; import 'package:huixiang/im/Proto.dart'; import 'package:huixiang/im/database/message.dart'; import 'package:huixiang/im/out/auth.pb.dart'; import 'package:huixiang/im/out/message.pb.dart'; import 'package:huixiang/main.dart'; import 'package:image_gallery_saver/image_gallery_saver.dart'; import 'package:path_provider/path_provider.dart'; import 'package:shared_preferences/shared_preferences.dart'; class SocketClient { //47.93.216.24:9090 线上 192.168.10.129:9090 测试 final String ip = kDebugMode ? '192.168.10.129' : '47.93.216.24'; final num port = 9090; Socket _socket; SharedPreferences _shared; Timer timer; bool get heartbeatActive => timer != null && timer.isActive; connect() async { _shared = await SharedPreferences.getInstance(); if (_socket != null) { reconnect(); return; } showDebugToast("socket-connect .... "); await Socket.connect(ip, port).then((value) { debugPrint("socket-connect-$ip"); _socket = value; _socket.listen((data) { print(data); print("socket-listen"); Proto proto = Proto.fromBytes(data); MsgData dataResult = MsgData.fromBuffer(proto.body); print('收到来自:${dataResult.from},消息内容: ${utf8.decode(dataResult.data)} '); receiveInsertMessage(dataResult).then((messageMap) { if (callbacks[dataResult.from] != null) { messageMap["state"] = 1; } hxDatabase.insert(messageMap).then((value) { messageMap["id"] = value; Message message = Message.fromJson(messageMap); if (callbacks[dataResult.from] != null) { callbacks[dataResult.from].call(message); /// user conversation callback } callbacks[userId]?.call(message); /// user self conversation list callback }); }); }, onError: (Object error, StackTrace stackTrace) { debugPrint("socket-listen-error: $error, stackTrace: $stackTrace"); showDebugToast("socket-listen-error: $error, stackTrace: $stackTrace"); reconnect(); }, onDone: () { debugPrint("socket-listen-down: down"); }); authRequest(_shared.getString("token")); heartbeat(); }).catchError((error) { debugPrint("socket-connect-error: $error"); showDebugToast("socket-connect-error: $error"); _socket = null; reconnect(); }); } Future> receiveInsertMessage(MsgData dataResult) async { String content = ""; String attach = ""; Uint8List dataBytes = dataResult.data; MsgType type = MsgType.values[dataResult.type.value]; if (type == MsgType.TEXT) { content = utf8.decode(dataBytes); } else if (type == MsgType.IMAGE || type == MsgType.VIDEO) { Map result = await ImageGallerySaver.saveImage( dataBytes, isReturnImagePathOfIOS: true, ); bool isSuccess = result["isSuccess"] != null && result["isSuccess"]; if (isSuccess) { attach = result["filePath"]; } } else if (type == MsgType.AUDIO) { Directory dir = await getTemporaryDirectory(); File file = File("${dir.path}/hx_${DateTime.now().millisecondsSinceEpoch}.wav"); try { IOSink ioSink = file.openWrite(); ioSink.write(dataBytes); ioSink.close(); attach = file.path; } catch (e) { debugPrint("${file.path} write fail"); } } else { content = "暂不支持的消息格式"; } Map messageMap = createMessage(userId, content, attach: attach, msgType: type.value, fromId: dataResult.from); return messageMap; } showDebugToast(text) { if (kDebugMode) { SmartDialog.showToast(text, alignment: Alignment.center); } } Proto heartbeatData() { DateTime dateTime = DateTime.now(); int millisecondsTime = dateTime.millisecondsSinceEpoch; Uint8List data = utf8.encode(jsonEncode({"heartbeat": millisecondsTime})); MsgData msgData = MsgData(from: userId, type: MsgType.TEXT, data: data); final proto2 = Proto(3, 1, msgData.writeToBuffer()); debugPrint("heartbeat: ${dateTime.toString()}"); return proto2; } heartbeat() { cancelTimer(); timer = Timer.periodic(const Duration(milliseconds: 30000), (timer) { if(!checkSocket()) { timer.cancel(); return; } sendHeartBeatAndCheckSocket(); }); } /// send heartBeat and check socket is connected /// send error: reconnect, /// else check Timer.periodic isActive , /// if not active: Timer.periodic send heartBeat sendHeartBeatAndCheckSocket() { final proto2 = heartbeatData(); try { _socket.add(proto2.toBytes()); if (!socketClient.heartbeatActive) { heartbeat(); debugPrint("socket-periodic-send-heart-beat"); } } catch (e) { debugPrint("socket-send-error: ${e.toString()}"); showDebugToast("socket-send-error: ${e.toString()}"); reconnect(); } } cancelTimer() { if (timer != null && timer.isActive) { timer.cancel(); timer = null; } } reconnect() { dispose(); connect(); } Map callbacks = {}; addCallback(String userId, callback) { callbacks[userId] = callback; } removeCallback(String userId) { callbacks.remove(userId); } dispose() { if (_socket != null) { _socket.close(); _socket = null; } } authRequest(String token) { if (!checkSocket()) { return; } debugPrint("socket-authRequest: request"); final authReq = AuthReq() ..uid = userId ..token = token; final authReqBytes = authReq.writeToBuffer(); final proto = Proto(1, 1, authReqBytes); // 假设 operation 和 seqId 为 1 final protoBytes = proto.toBytes(); try { _socket.add(protoBytes); } catch (e) { debugPrint("socket-authRequest: $e"); Future.delayed(const Duration(milliseconds: 1000), () { authRequest(token); }); } } Future sendMessage(String toId, String content, {String attach, int msgType, replyId}) async { MsgType type = MsgType.values[msgType]; Uint8List data; if (type == MsgType.TEXT) { data = utf8.encode(content); } else if (type == MsgType.IMAGE || type == MsgType.VIDEO || type == MsgType.AUDIO) { File file = File(attach); Directory dir = await getTemporaryDirectory(); File newFile = await file.copy("${dir.path}/hx_${DateTime.now().millisecondsSinceEpoch}.${file.path.split(".")[1]}"); data = await file.readAsBytes(); attach = newFile.path; } else { data = utf8.encode(content); } Map message = createMessage(toId, content, fromId: userId, attach: attach, msgType: msgType, replyId: replyId); message["state"] = 1; int id = await hxDatabase.insert(message).catchError((error) { debugPrint("insertMessage: $error"); }); if (!checkSocket()) { hxDatabase.update({"id": id, "state": 3}).catchError((error) { debugPrint("insertMessage: ${error.toString()}"); }); message["id"] = id; message["state"] = 3; return Message.fromJson(message); } message["id"] = id; MsgData msgData = MsgData(to: toId, from: userId, type: type, data: data); final proto2 = Proto(5, 1, msgData.writeToBuffer()); try { _socket.add(proto2.toBytes()); } catch (e) { hxDatabase.update({"id": id, "state": 3}).catchError((error) { debugPrint("insertMessage: ${error.toString()}"); }); debugPrint("socket-send-error: ${e.toString()}"); showDebugToast("socket-send-error: ${e.toString()}"); reconnect(); } return Message.fromJson(message); } checkSocket() { if (_socket == null) { reconnect(); return false; } return true; } String get userId => _shared.getString("userId"); }