huixiang_app/lib/im/SocketClient.dart

import 'dart:async';
import 'dart:convert';
import 'dart:core';
import 'dart:core';
import 'dart:io';
import 'package:flutter/foundation.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:shared_preferences/shared_preferences.dart';

class SocketClient {

  Socket _socket;
  SharedPreferences shared;

  connect() async {
    shared = await SharedPreferences.getInstance();

    //47.93.216.24:9090 线上  192.168.10.129:9090 测试
    await Socket.connect('47.93.216.24', 9090).then((value) {
      debugPrint("socket-connect");
      _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)} ');

        Map<String, dynamic> messageMap = createMessage(userId, utf8.decode(dataResult.data), msgType: dataResult.type.value, userId: dataResult.from);
        Message message = Message.fromJson(messageMap);
        callbacks[userId]?.call(message); /// user self conversation callback
        callbacks[dataResult.from]?.call(message); /// user conversation callback

        hxDatabase.insert(messageMap);

      }, onError: (Object error, StackTrace stackTrace) {
        debugPrint("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");
      reconnectTime = 1500;
      _socket = null;
      reconnect();
    });
  }

  heartbeat() {
    Timer.periodic(const Duration(milliseconds: 30000), (timer) {
      int milliseTime = DateTime.now().millisecondsSinceEpoch;
      debugPrint("heartbeat: ${milliseTime}");
      Uint8List data = utf8.encode(jsonEncode({"heartbeat": milliseTime}));
      MsgData msgData = MsgData(from: userId, type: MsgType.TEXT, data: data);
      final proto2 = Proto(3, 1, msgData.writeToBuffer());
      if(_socket != null)
      _socket.add(proto2.toBytes());
    });
  }

  int reconnectTime = 1500;

  reconnect() {
    Future.delayed(Duration(milliseconds: reconnectTime *= 2), () {
      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;
    }
    final authReq = AuthReq()
      ..uid = userId
      ..token = token;
    final authReqBytes = authReq.writeToBuffer();
    final proto = Proto(1, 1, authReqBytes); // 假设 operation 和 seqId 为 1
    final protoBytes = proto.toBytes();
    _socket.add(protoBytes);
  }

  Future<Message> sendMessage(String toId, String content) async {
    Map<String, dynamic> message = createMessage(toId, content, userId: userId);
    int id = await hxDatabase.insert(message).catchError((error) {
      debugPrint("insertMessage: $error");
    });
    if (!checkSocket()) {
      hxDatabase.update({"id": id, "state": 3}).catchError((error) {
        debugPrint("insertMessage: $error");
      });
      message["id"] = id;
      message["state"] = 3;
      return Message.fromJson(message);
    }
    message["id"] = id;
    Uint8List data = utf8.encode(content);
    MsgData msgData = MsgData(to: toId, from: userId, type: MsgType.TEXT, data: data);
    final proto2 = Proto(5, 1, msgData.writeToBuffer());
    _socket.add(proto2.toBytes());
    debugPrint("sendMessage: ${message["id"]}");
    return Message.fromJson(message);
  }

  checkSocket() {
    if (_socket == null) {
      reconnectTime = 1500;
      reconnect();
      return false;
    }
    return true;
  }

  String get userId => shared.getString("userId");

}