misha/eatvio_app
2
1
Fork 0
forked from eatvio/eatvio_app
Code Pull Requests Activity

Merge pull request 'fix(chat): ws manager' (#1060) from apoloz/eatvio_app:main into main

Browse Source 
Reviewed-on: eatvio/eatvio_app#1060
Reviewed-by: misha <misha@noreply.localhost>
This commit is contained in:
Анатолий Полоз
2026-09-25 07:31:28 +00:00
parent 8ec6d5b799 a00aae36cf
commit ae4f9d2e9b
7 changed files with 202 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
lib/core/websockets/ws_manager.dart
View File

@@ -38,6 +38,13 @@ class WsManager {
StreamSubscription? _eventSubscription;
StreamSubscription? _connectionEstablishedSubscription;
/// Широковещательный сигнал о восстановлении соединения после обрыва (S5).
///
/// Контроллер создаётся один раз и не закрывается в [clear]: подписчики
/// (чаты, окно чата) переживают логаут/повторный вход и продолжают получать
/// сигналы. Если контроллер всё же закрыт, он пересоздаётся в [initialize].
StreamController<void> _reconnectController = StreamController<void>.broadcast();
Stream<ChannelReadEvent>? _directChatStream;
Stream<ChannelReadEvent>? _clientTypingStream;
@@ -49,7 +56,17 @@ class WsManager {
Stream<ChannelReadEvent>? get clientTypingStream => _clientTypingStream;
/// Сигнал о реконнекте сокета: эмитится после повторной подписки на каналы.
///
/// Подписчики используют его, чтобы догрузить пропущенные за время обрыва
/// сообщения (S5).
Stream<void> get reconnected => _reconnectController.stream;
Future<void> initialize() async {
if (_reconnectController.isClosed) {
_reconnectController = StreamController<void>.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();
}

43
lib/features/chat/presentation/bloc/chat_room_cubit/chat_room_cubit.dart
View File

@@ -31,6 +31,7 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
StreamSubscription<ChannelReadEvent>? typingSubscription;
StreamSubscription<ChannelReadEvent>? messagesSubscription;
StreamSubscription<void>? _reconnectSubscription;
int currentPage = 1;
late int lastPage;
bool isLoading = false;
@@ -126,6 +127,7 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
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<ChatRoomState> {
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<void> 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<void> updateChosenPlanState() async {
try {
final currentState = state;
@@ -209,9 +247,11 @@ class ChatRoomCubit extends Cubit<ChatRoomState> {
}
Future<void> _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<ChatRoomState> {
_timer?.cancel();
await messagesSubscription?.cancel();
await typingSubscription?.cancel();
await _reconnectSubscription?.cancel();
super.close();
}

7
lib/features/chat/presentation/bloc/chats_cubit/chats_cubit.dart
View File

@@ -17,6 +17,7 @@ class ChatsCubit extends Cubit<ChatsState> {
final WsManager wsManager;
StreamSubscription? newIncomingMessagesSubscription;
StreamSubscription? newLastMessageSubscription;
StreamSubscription<void>? reconnectedSubscription;
ChatsCubit({
required this.chatRepository,
@@ -37,6 +38,11 @@ class ChatsCubit extends Cubit<ChatsState> {
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<ChatsState> {
@override
Future<void> close() {
newIncomingMessagesSubscription?.cancel();
reconnectedSubscription?.cancel();
return super.close();
}
}

2
pubspec.yaml
View File

@@ -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:

31
test/core/websockets/ws_manager_test.dart
View File

@@ -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 = <void>[];
final subscription = manager.reconnected.listen(received.add);
manager.handleConnectionEstablished();
await Future<void>.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 = <void>[];
final subscription = manager.reconnected.listen(received.add);
await manager.clear();
manager.handleConnectionEstablished();
await Future<void>.delayed(Duration.zero);
expect(received, hasLength(1));
await subscription.cancel();
});
});
}

73
test/features/chat/presentation/bloc/chat_room_cubit_test.dart
View File

@@ -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<void>.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<void>.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();
});
});
}

17
test/features/chat/presentation/bloc/chats_cubit_test.dart
View File

@@ -578,6 +578,23 @@ void main() {
expect(updatedRoom.unreadMessagesCount, 0);
expect(updatedState.chatsWithUnreadMessagesCount, 0);
});
test('connectToWebSocket - обновляет список чатов после реконнекта', () async {
final reconnectController = StreamController<void>.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();
});
});
}