together_mobile/lib/models/websocket_model.dart

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

import 'package:flutter/material.dart';
import 'package:hive_flutter/hive_flutter.dart';
import 'package:together_mobile/database/box_type.dart';

import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/status.dart' as status;

class WebSocketManager extends ChangeNotifier {
  late Uri wsUrl;
  late WebSocketChannel channel;
  String id = '';
  SocketStatus? socketStatus;
  Timer? heartBeatTimer;
  Timer? serverTimer;
  Timer? reconnectTimer;
  int reconnectCount = 30;
  int reconnectTimes = 0;
  Duration timeout = const Duration(seconds: 4);

  void connect(String userId) {
    id = userId;
    wsUrl = Uri.parse('ws://10.0.2.2:8000/ws/$id');
    channel = WebSocketChannel.connect(wsUrl);
    socketStatus = SocketStatus.connected;
    print('websocket connected <$channel>');
    heartBeatInspect();
    if (reconnectTimer != null) {
      reconnectTimer!.cancel();
      reconnectTimer = null;
      reconnectTimes = 0;
    }

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

  void onData(jsonData) {
    heartBeatInspect();
    Map<String, dynamic> data = json.decode(jsonData);
    switch (data['event']) {
      case 'one-to-one-chat':
        receiveFriendMsg(data);
    }
  }

  void onDone() {
    print('websocket disconnected <$channel>');
    reconnect();
  }

  void onError(Object error) {}

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

    if (serverTimer != null) {
      serverTimer!.cancel();
      serverTimer = null;
    }
    print('start heartbeat inspect......');
    heartBeatTimer = Timer(timeout, () {
      channel.sink.add(json.encode({'event': 'ping'}));
      serverTimer = Timer(timeout, () {
        channel.sink.close(status.internalServerError);
        socketStatus = SocketStatus.closed;
      });
    });
  }

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

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

    print('websocket reconnecting......');

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

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

    reconnectTimer = Timer.periodic(timeout, (timer) {
      if (reconnectTimes < reconnectCount) {
        connect(id);
        reconnectTimes++;
      } 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;
        channel.sink.close();
        timer.cancel();
      }
    });
  }
}

enum SocketStatus {
  connected,
  closed,
  error,
}

void receiveFriendMsg(Map<String, dynamic> msg) async {
  String senderId = msg['senderId'] as String;
  late Box<MessageT> messageTBox;
  try {
    messageTBox = Hive.box('message_$senderId');
  } catch (e) {
    messageTBox = await Hive.openBox('message_$senderId');
  }

  Box<ChatSetting> chatSettingBox = Hive.box<ChatSetting>('chat_setting');

  ChatSetting? chatSetting = chatSettingBox.get(senderId);

  if (chatSetting == null) {
    chatSettingBox.put(senderId, ChatSetting(senderId, false, true, false));
  } else {
    chatSetting.isOpen = true;
    chatSettingBox.put(senderId, chatSetting);
  }

  messageTBox.add(
    MessageT(
      senderId,
      msg['text'],
      msg['type'],
      DateTime.parse(msg['dateTime']),
      msg['isShowTime'],
      msg['attachments'],
    ),
  );
}