diff --git a/lib/core/websockets/ws_manager.dart b/lib/core/websockets/ws_manager.dart index 9ab8fe19..2d79ad63 100644 --- a/lib/core/websockets/ws_manager.dart +++ b/lib/core/websockets/ws_manager.dart @@ -38,6 +38,13 @@ class WsManager { StreamSubscription? _eventSubscription; StreamSubscription? _connectionEstablishedSubscription; + /// Широковещательный сигнал о восстановлении соединения после обрыва (S5). + /// + /// Контроллер создаётся один раз и не закрывается в [clear]: подписчики + /// (чаты, окно чата) переживают логаут/повторный вход и продолжают получать + /// сигналы. Если контроллер всё же закрыт, он пересоздаётся в [initialize]. + StreamController _reconnectController = StreamController.broadcast(); + Stream? _directChatStream; Stream? _clientTypingStream; @@ -49,7 +56,17 @@ class WsManager { Stream? get clientTypingStream => _clientTypingStream; + /// Сигнал о реконнекте сокета: эмитится после повторной подписки на каналы. + /// + /// Подписчики используют его, чтобы догрузить пропущенные за время обрыва + /// сообщения (S5). + Stream get reconnected => _reconnectController.stream; + Future initialize() async { + if (_reconnectController.isClosed) { + _reconnectController = StreamController.broadcast(); + } + if (client == null) { try { if (!await TokenStorage.haveTokenInStorage()) return; @@ -108,10 +125,7 @@ class WsManager { _lifecycleSubscription = client?.lifecycleStream.listen((data) { if (data == PusherChannelsClientLifeCycleState.establishedConnection) { - _unsubscribe(); - _subscribe(); - _privateDirectChatChannel?.subscribe(); - _presenceDirectChatChannel?.subscribe(); + handleConnectionEstablished(); } }); _eventSubscription = client?.eventStream.listen((event) { @@ -301,6 +315,19 @@ class WsManager { _privateNotificationChannel?.unsubscribe(); } + /// Повторно подписывается на каналы и сигналит подписчикам о реконнекте. + /// + /// Вызывается при `establishedConnection` — в том числе после обрыва сети. + /// Публичный, чтобы сигнал [reconnected] покрывался тестами без реального + /// Pusher-клиента. + void handleConnectionEstablished() { + _unsubscribe(); + _subscribe(); + _privateDirectChatChannel?.subscribe(); + _presenceDirectChatChannel?.subscribe(); + _reconnectController.add(null); + } + void subscribeDirectChatChannel() { _privateDirectChatChannel?.subscribe(); } diff --git a/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart b/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart index 1bdfa20c..377f0fe9 100644 --- a/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart +++ b/lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart @@ -31,6 +31,7 @@ class ChatRoomCubit extends Cubit { StreamSubscription? typingSubscription; StreamSubscription? messagesSubscription; + StreamSubscription? _reconnectSubscription; int currentPage = 1; late int lastPage; bool isLoading = false; @@ -126,6 +127,7 @@ class ChatRoomCubit extends Cubit { final chatMessages = await chatRepository.getChatMessages(resolvedUuid, currentPage); lastPage = chatMessages.meta.lastPage; await _subscribeToChatChannel(uuid: resolvedUuid, chatId: chatRoom.id); + _listenToReconnect(); emit( ChatRoomState.loaded( chatRoom: chatRoom.copyWith(chosenPlan: chosenPlan), @@ -146,12 +148,48 @@ class ChatRoomCubit extends Cubit { log(e.toString()); typingSubscription?.cancel(); messagesSubscription?.cancel(); + _reconnectSubscription?.cancel(); + _reconnectSubscription = null; wsManager.unsubscribeDirectChatChannel(); emit(const ChatRoomState.error(errorMessage: 'Ошибка получеия данных')); return null; } } + /// Подписывается на реконнект сокета, чтобы догрузить пропущенное (S5). + void _listenToReconnect() { + _reconnectSubscription?.cancel(); + _reconnectSubscription = wsManager.reconnected.listen((_) { + unawaited(refreshCurrentChat()); + }); + } + + /// Догружает первую страницу сообщений после восстановления соединения (S5). + /// + /// Мерж по `id` сохраняет уже подгруженные старые страницы: пропущенные + /// сообщения добавляются, существующие не дублируются. Порядок — по + /// `updatedAt` по убыванию, как его ждёт UI и отдаёт backend (S1). + Future refreshCurrentChat() async { + final uuid = _activeChatUuid; + if (uuid == null) return; + + try { + final page = await chatRepository.getChatMessages(uuid, 1); + lastPage = page.meta.lastPage; + _emitLoaded((current) { + final byId = {for (final message in current.chatMessages) message.id: message}; + for (final message in page.messagesList) { + byId.putIfAbsent(message.id, () => message); + } + final merged = byId.values.toList() + ..sort((a, b) => b.updatedAt.compareTo(a.updatedAt)); + return current.copyWith(chatMessages: merged); + }); + } catch (e, st) { + await reportFrom(e, st, this); + } + } + Future updateChosenPlanState() async { try { final currentState = state; @@ -209,9 +247,11 @@ class ChatRoomCubit extends Cubit { } Future _subscribeToChatChannel({required String uuid, required String chatId}) async { + // Активный чат запоминаем до проверки клиента: он нужен и для догрузки + // пропущенного после реконнекта (S5), даже если сокет появится позже. + _activeChatUuid = uuid; final client = wsManager.client; if (client == null) return; - _activeChatUuid = uuid; await messagesSubscription?.cancel(); await typingSubscription?.cancel(); messagesSubscription = null; @@ -571,6 +611,7 @@ class ChatRoomCubit extends Cubit { _timer?.cancel(); await messagesSubscription?.cancel(); await typingSubscription?.cancel(); + await _reconnectSubscription?.cancel(); super.close(); } diff --git a/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart b/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart index d4e421fa..cecc16ab 100644 --- a/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart +++ b/lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart @@ -17,6 +17,7 @@ class ChatsCubit extends Cubit { final WsManager wsManager; StreamSubscription? newIncomingMessagesSubscription; StreamSubscription? newLastMessageSubscription; + StreamSubscription? reconnectedSubscription; ChatsCubit({ required this.chatRepository, @@ -37,6 +38,11 @@ class ChatsCubit extends Cubit { newIncomingMessagesSubscription = wsManager.newMessagesStream?.listen((event) { _handleNewMessageViaSocket(event.data); }); + + await reconnectedSubscription?.cancel(); + reconnectedSubscription = wsManager.reconnected.listen((_) { + unawaited(getChatRooms()); + }); } catch (e, st) { await reportFrom(e, st, this); } @@ -199,6 +205,7 @@ class ChatsCubit extends Cubit { @override Future close() { newIncomingMessagesSubscription?.cancel(); + reconnectedSubscription?.cancel(); return super.close(); } } diff --git a/pubspec.yaml b/pubspec.yaml index cf78db1c..403b6ed4 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: eatvio_app description: A new Flutter project. -version: 1.3.45+134 +version: 1.3.45+135 publish_to: 'none' environment: diff --git a/test/core/websockets/ws_manager_test.dart b/test/core/websockets/ws_manager_test.dart index d0a584bb..ffa7614c 100644 --- a/test/core/websockets/ws_manager_test.dart +++ b/test/core/websockets/ws_manager_test.dart @@ -252,5 +252,36 @@ void main() { expect(manager.client, isNull); }); + + test('reconnected - эмитит сигнал после восстановления соединения', () async { + final wsService = MockWsService(); + final manager = WsManager(wsService: wsService, errorReporter: const NoOpErrorReporter()); + + final received = []; + final subscription = manager.reconnected.listen(received.add); + + manager.handleConnectionEstablished(); + await Future.delayed(Duration.zero); + + expect(received, hasLength(1)); + + await subscription.cancel(); + }); + + test('reconnected - переживает clear (поток не закрывается при логауте)', () async { + final wsService = MockWsService(); + final manager = WsManager(wsService: wsService, errorReporter: const NoOpErrorReporter()); + + final received = []; + final subscription = manager.reconnected.listen(received.add); + + await manager.clear(); + manager.handleConnectionEstablished(); + await Future.delayed(Duration.zero); + + expect(received, hasLength(1)); + + await subscription.cancel(); + }); }); } diff --git a/test/features/chat/presentation/bloc/chat_room_cubit_test.dart b/test/features/chat/presentation/bloc/chat_room_cubit_test.dart index 9c1f9b5e..dc25c93b 100644 --- a/test/features/chat/presentation/bloc/chat_room_cubit_test.dart +++ b/test/features/chat/presentation/bloc/chat_room_cubit_test.dart @@ -1729,5 +1729,78 @@ void main() { expect(state.chatMessages.any((message) => message.id == 88), isTrue); expect(state.chatMessages.firstWhere((message) => message.id == 1).isRead, isTrue); }); + + test('refreshCurrentChat - подтягивает сообщение, добавленное в БД в обход сокета', () async { + const uuid = 'room-uuid'; + final reconnectController = StreamController.broadcast(); + + when(mockWsManager.reconnected).thenAnswer((_) => reconnectController.stream); + when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom); + when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages); + when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan')); + when(mockWsManager.client).thenReturn(null); + when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value()); + when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null); + when(mockWsManager.directChatStream).thenReturn(null); + when(mockWsManager.clientTypingStream).thenReturn(null); + + await chatRoomCubit.initChat(uuid); + await Future.delayed(const Duration(milliseconds: 100)); + + final missedMessage = ChatMessageModel( + id: 99, + data: 'Missed during outage', + mediaData: null, + authorId: 2, + updatedAt: DateTime.now(), + isRead: false, + isCurrentUserAuthor: false, + roomId: 'chat-id', + ); + + when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer( + (_) async => ChatMessagesModel( + messagesList: [missedMessage, ...testChatMessages.messagesList], + meta: const ChatMetaModel(lastPage: 3), + ), + ); + + reconnectController.add(null); + await Future.delayed(const Duration(milliseconds: 100)); + + final state = chatRoomCubit.state as ChatRoomStateLoaded; + expect(state.chatMessages.any((message) => message.id == 99), isTrue); + expect(state.chatMessages.any((message) => message.id == 1), isTrue); + expect(chatRoomCubit.lastPage, 3); + + await reconnectController.close(); + }); + + test('refreshCurrentChat - идемпотентен: повторный реконнект не двоит сообщения', () async { + const uuid = 'room-uuid'; + final reconnectController = StreamController.broadcast(); + + when(mockWsManager.reconnected).thenAnswer((_) => reconnectController.stream); + when(mockChatRepository.getChatRoom(uuid)).thenAnswer((_) async => testChatRoom); + when(mockChatRepository.getChatMessages(uuid, 1)).thenAnswer((_) async => testChatMessages); + when(mockChatRepository.getChosenPlan(id: anyNamed('id'))).thenAnswer((_) async => throw Exception('No plan')); + when(mockWsManager.client).thenReturn(null); + when(mockWsManager.initPrivateDirectChatChannel(any)).thenAnswer((_) async => Future.value()); + when(mockWsManager.subscribeDirectChatChannel()).thenReturn(null); + when(mockWsManager.directChatStream).thenReturn(null); + when(mockWsManager.clientTypingStream).thenReturn(null); + + await chatRoomCubit.initChat(uuid); + await Future.delayed(const Duration(milliseconds: 100)); + + reconnectController.add(null); + reconnectController.add(null); + await Future.delayed(const Duration(milliseconds: 100)); + + final state = chatRoomCubit.state as ChatRoomStateLoaded; + expect(state.chatMessages.where((message) => message.id == 1).length, 1); + + await reconnectController.close(); + }); }); } diff --git a/test/features/chat/presentation/bloc/chats_cubit_test.dart b/test/features/chat/presentation/bloc/chats_cubit_test.dart index a3569618..f1919b22 100644 --- a/test/features/chat/presentation/bloc/chats_cubit_test.dart +++ b/test/features/chat/presentation/bloc/chats_cubit_test.dart @@ -578,6 +578,23 @@ void main() { expect(updatedRoom.unreadMessagesCount, 0); expect(updatedState.chatsWithUnreadMessagesCount, 0); }); + + test('connectToWebSocket - обновляет список чатов после реконнекта', () async { + final reconnectController = StreamController.broadcast(); + + when(mockWsManager.reconnected).thenAnswer((_) => reconnectController.stream); + when(mockWsManager.newMessagesStream).thenReturn(null); + when(mockChatRepository.getChatRooms(query: null)).thenAnswer((_) async => testChatRoomsPagination); + + await chatsCubit.connectToWebSocket(); + + reconnectController.add(null); + await Future.delayed(const Duration(milliseconds: 100)); + + verify(mockChatRepository.getChatRooms(query: null)).called(1); + + await reconnectController.close(); + }); }); }