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 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 data = json.decode(jsonData); Map 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 attachmentLoadingBox; try { attachmentLoadingBox = Hive.box('attachment_receive'); } catch (_) { attachmentLoadingBox = await Hive.openBox('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); } }