TF-3157 Update web socket with background service worker

TF-3157 Stub BroadcastChannel for mobile build
This commit is contained in:
DatDang
2024-09-27 10:42:10 +07:00
committed by Dat H. Pham
parent c4d0c27603
commit 7d59982cc9
16 changed files with 414 additions and 67 deletions
@@ -1,6 +1,7 @@
import 'dart:convert';
import 'package:core/data/constants/constant.dart';
import 'package:core/utils/app_logger.dart';
import 'package:core/utils/broadcast_channel/broadcast_channel.dart';
import 'package:jmap_dart_client/jmap/account_id.dart';
import 'package:jmap_dart_client/jmap/core/capability/capability_identifier.dart';
import 'package:jmap_dart_client/jmap/core/capability/websocket_capability.dart';
@@ -8,17 +9,21 @@ import 'package:jmap_dart_client/jmap/core/session/session.dart';
import 'package:model/extensions/session_extension.dart';
import 'package:rxdart/transformers.dart';
import 'package:tmail_ui_user/features/push_notification/data/datasource/web_socket_datasource.dart';
import 'package:tmail_ui_user/features/push_notification/data/model/connect_web_socket_message.dart';
import 'package:tmail_ui_user/features/push_notification/data/network/web_socket_api.dart';
import 'package:tmail_ui_user/features/push_notification/domain/exceptions/web_socket_exceptions.dart';
import 'package:tmail_ui_user/features/push_notification/domain/model/web_socket_action.dart';
import 'package:tmail_ui_user/main/error/capability_validator.dart';
import 'package:tmail_ui_user/main/exceptions/exception_thrower.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:universal_html/html.dart';
class RemoteWebSocketDatasourceImpl implements WebSocketDatasource {
class WebSocketDatasourceImpl implements WebSocketDatasource {
final WebSocketApi _webSocketApi;
final ExceptionThrower _exceptionThrower;
const RemoteWebSocketDatasourceImpl(this._webSocketApi, this._exceptionThrower);
const WebSocketDatasourceImpl(this._webSocketApi, this._exceptionThrower);
static const String _webSocketClosed = 'webSocketClosed';
@override
Stream getWebSocketChannel(Session session, AccountId accountId) {
@@ -30,32 +35,31 @@ class RemoteWebSocketDatasourceImpl implements WebSocketDatasource {
Stream _getWebSocketChannel(
Session session,
AccountId accountId,
[int retryRemained = 3]
) async* {
) async* {
final broadcastChannel = BroadcastChannel(Constant.wsServiceWorkerBroadcastChannel);
try {
_verifyWebSocketCapabilities(session, accountId);
final webSocketTicket = await _webSocketApi.getWebSocketTicket(session, accountId);
final webSocketUri = _getWebSocketUri(session, accountId);
window.navigator.serviceWorker?.controller?.postMessage(ConnectWebSocketMessage(
webSocketAction: WebSocketAction.connect,
webSocketUrl: webSocketUri.toString(),
webSocketTicket: webSocketTicket
).toJson());
final webSocketChannel = WebSocketChannel.connect(
Uri.parse('$webSocketUri?ticket=$webSocketTicket'));
await webSocketChannel.ready;
webSocketChannel.sink.add(jsonEncode({"@type": "WebSocketPushEnable"}));
yield* webSocketChannel.stream;
yield* _webSocketListener(broadcastChannel);
} catch (e) {
logError('RemoteWebSocketDatasourceImpl::getWebSocketChannel():error: $e');
if (retryRemained > 0) {
yield* _getWebSocketChannel(session, accountId, retryRemained - 1);
} else {
rethrow;
}
rethrow;
}
}
void _verifyWebSocketCapabilities(Session session, AccountId accountId) {
if (!CapabilityIdentifier.jmapWebSocket.isSupported(session, accountId)
|| !CapabilityIdentifier.jmapWebSocketTicket.isSupported(session, accountId)
|| session.getCapabilityProperties<WebSocketCapability>(
accountId,
CapabilityIdentifier.jmapWebSocket)?.supportsPush != true
) {
throw WebSocketPushNotSupportedException();
}
@@ -73,4 +77,14 @@ class RemoteWebSocketDatasourceImpl implements WebSocketDatasource {
return webSocketUri;
}
Stream _webSocketListener(BroadcastChannel broadcastChannel) {
return broadcastChannel.onMessage.map((event) {
if (event.data == _webSocketClosed) {
throw WebSocketClosedException();
}
return event.data;
});
}
}
@@ -0,0 +1,28 @@
import 'package:equatable/equatable.dart';
import 'package:json_annotation/json_annotation.dart';
import 'package:tmail_ui_user/features/push_notification/domain/model/web_socket_action.dart';
part 'connect_web_socket_message.g.dart';
@JsonSerializable()
class ConnectWebSocketMessage with EquatableMixin {
@JsonKey(name: 'action')
final WebSocketAction webSocketAction;
@JsonKey(name: 'url')
final String webSocketUrl;
@JsonKey(name: 'ticket')
final String webSocketTicket;
ConnectWebSocketMessage({
required this.webSocketAction,
required this.webSocketUrl,
required this.webSocketTicket,
});
factory ConnectWebSocketMessage.fromJson(Map<String, dynamic> json)
=> _$ConnectWebSocketMessageFromJson(json);
Map<String, dynamic> toJson() => _$ConnectWebSocketMessageToJson(this);
@override
List<Object?> get props => [webSocketAction, webSocketUrl, webSocketTicket];
}
@@ -0,0 +1,31 @@
import 'package:json_annotation/json_annotation.dart';
part 'web_socket_echo.g.dart';
@JsonSerializable(includeIfNull: false)
class WebSocketEcho {
@JsonKey(name: '@type')
final String? type;
final String? requestId;
final List<List<dynamic>>? methodResponses;
WebSocketEcho({
this.type,
this.requestId,
this.methodResponses,
});
factory WebSocketEcho.fromJson(Map<String, dynamic> json) => _$WebSocketEchoFromJson(json);
Map<String, dynamic> toJson() => _$WebSocketEchoToJson(this);
static bool isValid(Map<String, dynamic> json) {
try {
final webSocketEcho = WebSocketEcho.fromJson(json);
final listResponses = webSocketEcho.methodResponses?.firstOrNull;
return listResponses?.contains('Core/echo') ?? false;
} catch (_) {
return false;
}
}
}
@@ -2,4 +2,6 @@ class WebSocketPushNotSupportedException implements Exception {}
class WebSocketUriUnavailableException implements Exception {}
class WebSocketTicketUnavailableException implements Exception {}
class WebSocketTicketUnavailableException implements Exception {}
class WebSocketClosedException implements Exception {}
@@ -0,0 +1 @@
enum WebSocketAction {connect, disconnect}
@@ -5,12 +5,12 @@ import 'package:jmap_dart_client/jmap/push/state_change.dart';
class InitializingWebSocketPushChannel extends LoadingState {}
class WebSocketPushStateReceived extends UIState {
final StateChange stateChange;
final StateChange? stateChange;
WebSocketPushStateReceived(this.stateChange);
@override
List<Object> get props => [stateChange];
List<Object?> get props => [stateChange];
}
class WebSocketConnectionFailed extends FeatureFailure {
@@ -7,6 +7,7 @@ import 'package:dartz/dartz.dart';
import 'package:jmap_dart_client/jmap/account_id.dart';
import 'package:jmap_dart_client/jmap/core/session/session.dart';
import 'package:jmap_dart_client/jmap/push/state_change.dart';
import 'package:tmail_ui_user/features/push_notification/data/model/web_socket_echo.dart';
import 'package:tmail_ui_user/features/push_notification/domain/repository/web_socket_repository.dart';
import 'package:tmail_ui_user/features/push_notification/domain/state/web_socket_push_state.dart';
@@ -31,9 +32,21 @@ class ConnectWebSocketInteractor {
}
Either<Failure, Success> _toStateChange(dynamic data) {
if (data is String) {
data = jsonDecode(data);
StateChange? possibleStateChange;
try {
if (data is String) {
data = jsonDecode(data);
}
possibleStateChange = StateChange.fromJson(data);
return Right(WebSocketPushStateReceived(possibleStateChange));
} catch (e) {
logError('ConnectWebSocketInteractor::_toStateChange: '
'websocket message is not StateChange: $data');
final dataIsWebSocketEcho = WebSocketEcho.isValid(data);
if (dataIsWebSocketEcho) {
return Right(WebSocketPushStateReceived(null));
}
return Left(WebSocketConnectionFailed(exception: e));
}
return Right(WebSocketPushStateReceived(StateChange.fromJson(data)));
}
}
@@ -1,7 +1,7 @@
import 'package:get/get.dart';
import 'package:tmail_ui_user/features/base/interactors_bindings.dart';
import 'package:tmail_ui_user/features/push_notification/data/datasource/web_socket_datasource.dart';
import 'package:tmail_ui_user/features/push_notification/data/datasource_impl/remote_web_socket_datasource_impl.dart';
import 'package:tmail_ui_user/features/push_notification/data/datasource_impl/web_socket_datasource_impl.dart';
import 'package:tmail_ui_user/features/push_notification/data/network/web_socket_api.dart';
import 'package:tmail_ui_user/features/push_notification/data/repository/web_socket_repository_impl.dart';
import 'package:tmail_ui_user/features/push_notification/domain/repository/web_socket_repository.dart';
@@ -11,12 +11,12 @@ import 'package:tmail_ui_user/main/exceptions/remote_exception_thrower.dart';
class WebSocketInteractorBindings extends InteractorsBindings {
@override
void bindingsDataSource() {
Get.lazyPut<WebSocketDatasource>(() => Get.find<RemoteWebSocketDatasourceImpl>());
Get.lazyPut<WebSocketDatasource>(() => Get.find<WebSocketDatasourceImpl>());
}
@override
void bindingsDataSourceImpl() {
Get.lazyPut(() => RemoteWebSocketDatasourceImpl(
Get.lazyPut(() => WebSocketDatasourceImpl(
Get.find<WebSocketApi>(),
Get.find<RemoteExceptionThrower>(),
));
@@ -1,3 +1,5 @@
import 'dart:async';
import 'package:collection/collection.dart';
import 'package:core/presentation/state/failure.dart';
import 'package:core/presentation/state/success.dart';
@@ -17,12 +19,12 @@ abstract class PushBaseController {
Session? session;
AccountId? accountId;
StreamSubscription<Either<Failure, Success>>? _stateStreamSubscription;
void consumeState(Stream<Either<Failure, Success>> newStateStream) {
newStateStream.listen(
_stateStreamSubscription = newStateStream.listen(
_handleStateStream,
onError: (error, stackTrace) {
logError('PushBaseController::consumeState():onError:error: $error | stackTrace: $stackTrace');
}
onError: handleErrorViewState,
);
}
@@ -34,6 +36,15 @@ abstract class PushBaseController {
void handleSuccessViewState(Success success);
void handleErrorViewState(Object error, StackTrace stackTrace) {
logError('PushBaseController::handleErrorViewState():error: $error | stackTrace: $stackTrace');
}
void cancelStateStreamSubscription() {
_stateStreamSubscription?.cancel();
_stateStreamSubscription = null;
}
void initialize({AccountId? accountId, Session? session}) {
this.accountId = accountId;
this.session = session;
@@ -87,22 +98,20 @@ abstract class PushBaseController {
{Session? session}
) {
final newState = jmap.State(mapTypeState[typeName.value]);
if (typeName == TypeName.emailType) {
if (isForeground) {
return SynchronizeEmailOnForegroundAction(typeName, newState, accountId, session);
} else {
return StoreEmailStateToRefreshAction(typeName, newState, accountId, userName, session);
}
} else if (typeName == TypeName.emailDelivery) {
if (!isForeground) {
return PushNotificationAction(typeName, newState, session, accountId, userName);
}
} else if (typeName == TypeName.mailboxType) {
if (isForeground) {
return SynchronizeMailboxOnForegroundAction(typeName, newState, accountId);
} else {
return StoreMailboxStateToRefreshAction(typeName, newState, accountId, userName);
}
switch (typeName) {
case TypeName.emailType:
return isForeground
? SynchronizeEmailOnForegroundAction(typeName, newState, accountId, session)
: StoreEmailStateToRefreshAction(typeName, newState, accountId, userName, session);
case TypeName.emailDelivery:
if (!isForeground) {
return PushNotificationAction(typeName, newState, session, accountId, userName);
}
break;
case TypeName.mailboxType:
return isForeground
? SynchronizeMailboxOnForegroundAction(typeName, newState, accountId)
: StoreMailboxStateToRefreshAction(typeName, newState, accountId, userName);
}
return null;
}
@@ -3,6 +3,7 @@ import 'package:core/presentation/state/success.dart';
import 'package:core/utils/app_logger.dart';
import 'package:jmap_dart_client/jmap/account_id.dart';
import 'package:jmap_dart_client/jmap/core/session/session.dart';
import 'package:tmail_ui_user/features/push_notification/domain/exceptions/web_socket_exceptions.dart';
import 'package:tmail_ui_user/features/push_notification/domain/state/web_socket_push_state.dart';
import 'package:tmail_ui_user/features/push_notification/domain/usecases/connect_web_socket_interactor.dart';
import 'package:tmail_ui_user/features/push_notification/presentation/controller/push_base_controller.dart';
@@ -20,9 +21,15 @@ class WebSocketController extends PushBaseController {
ConnectWebSocketInteractor? _connectWebSocketInteractor;
int _retryRemained = 3;
@override
void handleFailureViewState(Failure failure) {
logError('WebSocketController::handleFailureViewState():Failure $failure');
cancelStateStreamSubscription();
if (failure is WebSocketConnectionFailed) {
_handleWebSocketConnectionRetry();
}
}
@override
@@ -33,6 +40,15 @@ class WebSocketController extends PushBaseController {
}
}
@override
void handleErrorViewState(Object error, StackTrace stackTrace) {
super.handleErrorViewState(error, stackTrace);
cancelStateStreamSubscription();
if (error is WebSocketClosedException) {
_handleWebSocketConnectionRetry();
}
}
@override
void initialize({AccountId? accountId, Session? session}) {
super.initialize(accountId: accountId, session: session);
@@ -51,8 +67,10 @@ class WebSocketController extends PushBaseController {
void _handleWebSocketPushStateReceived(WebSocketPushStateReceived success) {
log('WebSocketController::_handleWebSocketPushStateReceived(): $success');
_retryRemained = 3;
if (accountId == null || session == null) return;
final stateChange = success.stateChange;
if (stateChange == null) return;
final mapTypeState = stateChange.getMapTypeState(accountId!);
mappingTypeStateToAction(
mapTypeState,
@@ -62,4 +80,11 @@ class WebSocketController extends PushBaseController {
session!.username,
session: session);
}
void _handleWebSocketConnectionRetry() {
if (_retryRemained > 0) {
_retryRemained--;
_connectWebSocket(accountId, session);
}
}
}