huixiang_app/lib/im/SocketClient.dart

import 'dart:async';
import 'dart:convert';
import 'dart:core';
import 'dart:io';
import 'package:dio/dio.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter/material.dart';
import 'package:flutter_smart_dialog/flutter_smart_dialog.dart';
import 'package:huixiang/constant.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 {

  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(socketHost, socketPort).then((value) {
      debugPrint("socket-connect-$socketHost");
      _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<Map<String, dynamic>> receiveInsertMessage(MsgData dataResult) async {
    Uint8List dataBytes = dataResult.data;
    String content = utf8.decode(dataBytes);

    String attach = "";
    MsgType type = MsgType.values[dataResult.type.value];
    if (type == MsgType.IMAGE || type == MsgType.VIDEO || type == MsgType.AUDIO) {

      String filePath = await qiniu.downloadFile(content);

      Map<String, dynamic> result = await ImageGallerySaver.saveFile(filePath).catchError((error){});
      bool isSuccess = result["isSuccess"] != null && result["isSuccess"];
      attach = filePath;
    }

    Map<String, dynamic> 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<String, Function(Message message)> callbacks = <String, Function(Message message)>{};

  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<Message> sendMessage(String toId, String content, {String attach, int msgType = 1, replyId}) async {
    MsgType type = MsgType.values[msgType];
    Uint8List data = utf8.encode(content);
    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_${attach.split('/').last}");
      attach = newFile.path;
    }
    Map<String, dynamic> 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");

}