together_mobile/lib/models/websocket_model.dart

import 'dart:async';
import 'dart:convert';

import 'package:flutter/foundation.dart';
import 'package:hive_flutter/hive_flutter.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/status.dart' as status;

import '/database/box_type.dart';
import '/utils/ws_receive_callback.dart';

enum SocketStatus {
  connected,
  reconnecting,
  error,
  closed,
}

// const chunkSize = 1024 * 1024;

class WebSocketManager extends ChangeNotifier {
  late Uri wsUrl;
  late WebSocketChannel channel;
  String id = '';
  SocketStatus socketStatus = SocketStatus.closed;
  Timer? heartBeatTimer;
  Timer? serverTimer;
  Timer? reconnectTimer;
  int reconnectCount = 5;
  int reconnectTimes = 0;
  Duration heartBeatTimeout = const Duration(seconds: 30);
  Duration reconnectTimeout = const Duration(seconds: 5);
  Map<String, Timer> sendImageTimer = {};

  void connect(String userId, bool isReconnect) {
    id = userId;
    wsUrl = Uri.parse('ws://10.0.2.2:8000/ws/$id?is_reconnect=$isReconnect');
    if (isReconnect) {
      socketStatus = SocketStatus.reconnecting;
      notifyListeners();
    }
    // This doesn't block the program whether it connect the server or not
    // So heartBeat will be execute straightly
    channel = WebSocketChannel.connect(wsUrl);
    heartBeatInspect();

    channel.stream.listen(onData, onError: onError, onDone: onDone);
  }

  void disconnect() {
    channel.sink.close();
    wsUrl = Uri();
    id = '';
    socketStatus = SocketStatus.closed;
    notifyListeners();
    heartBeatTimer?.cancel();
    serverTimer?.cancel();
    reconnectTimer?.cancel();
    heartBeatTimer = null;
    serverTimer = null;
    reconnectTimer = null;
    reconnectTimes = 0;
  }

  void onData(jsonData) async {
    // If socket can receive msg, that means connection is estabilished
    socketStatus = SocketStatus.connected;
    notifyListeners();
    print('websocket connected <$channel>');
    if (reconnectTimer != null) {
      reconnectTimer!.cancel();
      reconnectTimer = null;
      reconnectTimes = 0;
    }

    heartBeatInspect();

    // Map<String, dynamic> data = json.decode(jsonData);
    Map<String, dynamic> data =
        await compute((message) => json.decode(message), jsonData);
    switch (data['event']) {
      case 'friend-chat-msg':
        await receiveFriendMsg(data, true);
      case 'apply-friend':
        receiveApplyFriend(data);
      case 'friend-added':
        receiveFriendAdded(data);
      case 'friend-deleted':
        receiveFriendDeleted(data);
      case 'chat-image':
        receiveChatImages(data);
      case 'group-chat-creation':
        receiveGroupChatCreation(data);
      case 'group-chat-msg':
        receiveGroupChatMsg(data, true);
      case 'pull-chat-image':
        receivePullChatImage(data);
      case 'chat-image-send-ok':
        receiveCheckChatImage(data);
    }
  }

  // This will be trigger while server or client close the connection
  // for example server is restarting
  void onDone() {
    print('websocket disconnected <$channel>');
    print('22222222222222222222');
    if (socketStatus == SocketStatus.closed) {
      // Client close the connection
      return;
    }
    if (socketStatus == SocketStatus.connected) {
      // Server close the connection
      socketStatus = SocketStatus.reconnecting;
      notifyListeners();
      print(111111111111111);
      reconnectTimes++;
      reconnect();
    }
    if (reconnectTimes >= reconnectCount) {
      socketStatus = SocketStatus.error;
      reconnectTimes = 0;
    } else {
      socketStatus = SocketStatus.reconnecting;
      print('3333333333333333333');
      notifyListeners();
    }
  }

  // This will be trigger while server exactly shutdown
  void onError(Object error, StackTrace st) {
    print('Websocket connect occurs error: <$error>');
    // print(st);
    if (reconnectTimes >= reconnectCount) {
      socketStatus = SocketStatus.error;
      notifyListeners();
      channel.sink.close();
      if (heartBeatTimer != null) {
        heartBeatTimer!.cancel();
        heartBeatTimer = null;
      }
    } else {
      print('${reconnectTimes}th reconnection');
      reconnect();
    }
  }

  void heartBeatInspect() {
    print('start heartbeat inspect......');

    if (heartBeatTimer != null) {
      heartBeatTimer!.cancel();
      heartBeatTimer = null;
    }

    if (serverTimer != null) {
      serverTimer!.cancel();
      serverTimer = null;
    }

    heartBeatTimer = Timer(heartBeatTimeout, () {
      channel.sink.add(json.encode({'event': 'ping'}));
      serverTimer = Timer(heartBeatTimeout, () {
        // This will trigger the onDone callback
        channel.sink.close(status.internalServerError);
        socketStatus = SocketStatus.reconnecting;
        notifyListeners();
      });
    });
  }

  void reconnect() {
    if (heartBeatTimer != null) {
      heartBeatTimer!.cancel();
      heartBeatTimer = null;
    }

    if (serverTimer != null) {
      serverTimer!.cancel();
      serverTimer = null;
    }

    if (reconnectTimer != null) {
      reconnectTimer!.cancel();
      reconnectTimer = null;
    }

    reconnectTimer = Timer(reconnectTimeout, () {
      if (reconnectTimes < reconnectCount) {
        print('websocket reconnecting......');
        reconnectTimes++;
        connect(id, true);
      } else {
        print('reconnection times exceed the max times......');
        // If it is still disconnection after reconnect 30 times, set the socket
        // status to error, means the network is bad, and stop reconnecting.
        socketStatus = SocketStatus.error;
        notifyListeners();
        channel.sink.close();
      }
    });
  }

  void addSendImageTimer(String filename, int totalChunkNum) {
    if (sendImageTimer.containsKey(filename)) {
      sendImageTimer[filename]!.cancel();
      sendImageTimer.remove(filename);
    }

    sendImageTimer[filename] = Timer(
      heartBeatTimeout,
      () async {
        late Box<AttachmentProgress> attachmentLoadingBox;
        try {
          attachmentLoadingBox =
              Hive.box<AttachmentProgress>('attachment_receive');
        } catch (_) {
          attachmentLoadingBox =
              await Hive.openBox<AttachmentProgress>('attachment_receive');
        }

        AttachmentProgress? at = attachmentLoadingBox.get(filename);

        if (at == null) {
          attachmentLoadingBox.put(
            filename,
            AttachmentProgress(
              0,
              totalChunkNum,
              0,
              false,
              true,
            ),
          );
        } else {
          at.isPause = true;
          attachmentLoadingBox.put(filename, at);
        }
      },
    );
  }

  void removeSendImageTimer(String filename) {
    if (sendImageTimer.containsKey(filename)) {
      sendImageTimer[filename]!.cancel();
    }
    sendImageTimer.remove(filename);
  }
}