From 12e6b9c500bfee4c5375a13deac3960473129bb5 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Jun 2021 17:56:38 +0530 Subject: [PATCH] unify errors, refactor retry_queue Signed-off-by: Sahil Kumar --- packages/stream_chat/lib/src/api/channel.dart | 42 +-- .../stream_chat/lib/src/api/retry_policy.dart | 35 +-- .../stream_chat/lib/src/api/retry_queue.dart | 251 ++++++++++-------- packages/stream_chat/lib/src/client.dart | 34 ++- .../http/interceptor/auth_interceptor.dart | 4 +- .../src/core/http/stream_chat_dio_error.dart | 2 +- .../lib/src/core/http/stream_http_client.dart | 11 +- .../lib/src/core/models/event.dart | 2 +- packages/stream_chat/lib/src/core/utils.dart | 9 + .../lib/src/errors/chat_error_code.dart | 229 ++++++++++------ .../lib/src/errors/stream_chat_error.dart | 106 ++++++-- packages/stream_chat/lib/src/exceptions.dart | 53 ---- .../stream_chat/lib/src/ws/socket_error.dart | 25 -- .../stream_chat/lib/src/ws/websocket.dart | 55 ++-- packages/stream_chat/lib/stream_chat.dart | 3 +- .../stream_chat/test/src/client_test.dart | 1 - .../lib/src/message_actions_modal.dart | 3 +- .../lib/src/db/moor_chat_database.dart | 2 +- 18 files changed, 486 insertions(+), 381 deletions(-) delete mode 100644 packages/stream_chat/lib/src/exceptions.dart delete mode 100644 packages/stream_chat/lib/src/ws/socket_error.dart diff --git a/packages/stream_chat/lib/src/api/channel.dart b/packages/stream_chat/lib/src/api/channel.dart index 3fd38e41..3fb63c91 100644 --- a/packages/stream_chat/lib/src/api/channel.dart +++ b/packages/stream_chat/lib/src/api/channel.dart @@ -7,6 +7,8 @@ import 'package:dio/dio.dart'; import 'package:logging/logging.dart'; import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/api/retry_queue.dart'; +import 'package:stream_chat/src/core/utils.dart'; +import 'package:stream_chat/src/errors/stream_chat_error.dart'; import 'package:stream_chat/src/event_type.dart'; import 'package:stream_chat/src/extensions/rate_limit.dart'; import 'package:stream_chat/src/core/models/attachment_file.dart'; @@ -233,12 +235,14 @@ class Channel { }) { final cancelToken = _cancelableAttachmentUploadRequest[attachmentId]; if (cancelToken == null) { - throw Exception( - "Upload request for this Attachment hasn't started yet or else " + throw const StreamChatError( + "Upload request for this Attachment hasn't started yet or maybe " 'Already completed', ); } - if (cancelToken.isCancelled) throw Exception('Already cancelled'); + if (cancelToken.isCancelled) { + throw const StreamChatError('Upload request already cancelled'); + } cancelToken.cancel(reason); } @@ -255,7 +259,7 @@ class Channel { ); if (message == null) { - throw Exception('Error, Message not found'); + throw const StreamChatError('Error, Message not found'); } final attachments = message.attachments.where((it) { @@ -396,9 +400,9 @@ class Channel { final response = await _client.sendMessage(message, id!, type); state!.addMessage(response.message); return response; - } catch (error) { - if (error is DioError && error.type != DioErrorType.response) { - state!.retryQueue?.add([message]); + } catch (e) { + if (e is StreamChatNetworkError && e.isRetriable) { + state!._retryQueue.add([message]); } rethrow; } @@ -453,9 +457,9 @@ class Channel { state?.addMessage(m); return response; - } catch (error) { - if (error is DioError && error.type != DioErrorType.response) { - state?.retryQueue?.add([message]); + } catch (e) { + if (e is StreamChatNetworkError && e.isRetriable) { + state!._retryQueue.add([message]); } rethrow; } @@ -494,9 +498,9 @@ class Channel { state?.addMessage(message.copyWith(status: MessageSendingStatus.sent)); return response; - } catch (error) { - if (error is DioError && error.type != DioErrorType.response) { - state?.retryQueue?.add([message]); + } catch (e) { + if (e is StreamChatNetworkError && e.isRetriable) { + state!._retryQueue.add([message]); } rethrow; } @@ -1188,9 +1192,11 @@ class ChannelClientState { _channel._client.chatPersistenceClient ?.updateChannelState(state)) .debounced(const Duration(seconds: 1)) { - retryQueue = RetryQueue( + _retryQueue = RetryQueue( channel: _channel, - logger: Logger('RETRY QUEUE ${_channel.cid}'), + logger: _channel.client.detachedLogger( + '⟳ (${generateHash([_channel.cid])})', + ), ); _checkExpiredAttachmentMessages(channelState); @@ -1338,7 +1344,7 @@ class ChannelClientState { BehaviorSubject.seeded(true); /// The retry queue associated to this channel - RetryQueue? retryQueue; + late final RetryQueue _retryQueue; /// Retry failed message Future retryFailedMessages() async { @@ -1357,7 +1363,7 @@ class ChannelClientState { ) .toList(); - retryQueue!.add(failedMessages); + _retryQueue.add(failedMessages); } void _listenReactionDeleted() { @@ -1832,7 +1838,7 @@ class ChannelClientState { void dispose() { _debouncedUpdatePersistenceChannelState.cancel(); _unreadCountController.close(); - retryQueue!.dispose(); + _retryQueue.dispose(); _subscriptions.forEach((s) => s.cancel()); _channelStateController.close(); _isUpToDateController.close(); diff --git a/packages/stream_chat/lib/src/api/retry_policy.dart b/packages/stream_chat/lib/src/api/retry_policy.dart index f0f2c67b..14933fed 100644 --- a/packages/stream_chat/lib/src/api/retry_policy.dart +++ b/packages/stream_chat/lib/src/api/retry_policy.dart @@ -1,5 +1,5 @@ import 'package:stream_chat/src/client.dart'; -import 'package:stream_chat/src/exceptions.dart'; +import 'package:stream_chat/src/errors/stream_chat_error.dart'; /// The retry options class RetryPolicy { @@ -7,32 +7,25 @@ class RetryPolicy { RetryPolicy({ required this.shouldRetry, required this.retryTimeout, - this.attempt = 0, + this.maxRetryAttempts = 6, }); - /// The number of attempts tried so far - int attempt = 0; + /// Hard limit on maximum retry attempts before giving up, defaults to 6 + /// Resets once connection recovers. + final int maxRetryAttempts; /// This function evaluates if we should retry the failure - final bool Function(StreamChatClient client, int attempt, ApiError? apiError) - shouldRetry; + final bool Function( + StreamChatClient client, + int attempt, + StreamChatError? error, + ) shouldRetry; /// In the case that we want to retry a failed request the retryTimeout /// method is called to determine the timeout final Duration Function( - StreamChatClient client, int attempt, ApiError? apiError) retryTimeout; - - /// Creates a copy of [RetryPolicy] with specified attributes overridden. - RetryPolicy copyWith({ - bool Function(StreamChatClient client, int attempt, ApiError? apiError)? - shouldRetry, - Duration Function(StreamChatClient client, int attempt, ApiError? apiError)? - retryTimeout, - int? attempt, - }) => - RetryPolicy( - retryTimeout: retryTimeout ?? this.retryTimeout, - shouldRetry: shouldRetry ?? this.shouldRetry, - attempt: attempt ?? this.attempt, - ); + StreamChatClient client, + int attempt, + StreamChatError? error, + ) retryTimeout; } diff --git a/packages/stream_chat/lib/src/api/retry_queue.dart b/packages/stream_chat/lib/src/api/retry_queue.dart index e6d883a9..9d0b8502 100644 --- a/packages/stream_chat/lib/src/api/retry_queue.dart +++ b/packages/stream_chat/lib/src/api/retry_queue.dart @@ -2,10 +2,11 @@ import 'dart:async'; import 'package:collection/collection.dart'; import 'package:logging/logging.dart'; +import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/api/channel.dart'; import 'package:stream_chat/src/api/retry_policy.dart'; +import 'package:stream_chat/src/errors/stream_chat_error.dart'; import 'package:stream_chat/src/event_type.dart'; -import 'package:stream_chat/src/exceptions.dart'; import 'package:stream_chat/src/core/models/message.dart'; import 'package:stream_chat/stream_chat.dart'; @@ -15,116 +16,155 @@ class RetryQueue { RetryQueue({ required this.channel, this.logger, - }) { - _retryPolicy = channel.client.retryPolicy; - + }) : client = channel.client { + _retryPolicy = client.retryPolicy; _listenConnectionRecovered(); - _listenFailedEvents(); } /// The channel of this queue final Channel channel; + /// The client associated with this [channel] + final StreamChatClient client; + /// The logger associated to this queue final Logger? logger; - final _subscriptions = []; + late final RetryPolicy _retryPolicy; + + final _compositeSubscription = CompositeSubscription(); + + final _messageQueue = HeapPriorityQueue(_byDate); + bool _isRetrying = false; void _listenConnectionRecovered() { - _subscriptions - .add(channel.client.on(EventType.connectionRecovered).listen((event) { - if (!_isRetrying && event.online!) { + client.on(EventType.connectionRecovered).listen((event) { + if (event.online == true) { _startRetrying(); } - })); + }).addTo(_compositeSubscription); } - final HeapPriorityQueue _messageQueue = HeapPriorityQueue(_byDate); - bool _isRetrying = false; - RetryPolicy? _retryPolicy; + void _listenFailedEvents() { + channel.on().where((event) => event.message != null).listen((event) { + final message = event.message!; + final containsMessage = _messageQueue.containsMessage(message); + if (!containsMessage) return; + if (message.status == MessageSendingStatus.sent) { + logger?.info('Removing sent message from queue : ${message.id}'); + _messageQueue.removeMessage(message); + return; + } else { + if ([ + MessageSendingStatus.failed_update, + MessageSendingStatus.failed, + MessageSendingStatus.failed_delete, + ].contains(message.status)) { + logger?.info('Adding failed message from event : ${event.type}'); + add([message]); + } + } + }).addTo(_compositeSubscription); + } /// Add a list of messages void add(List messages) { - logger?.info('added ${messages.length} messages'); + if (messages.isEmpty) return; + if (_messageQueue.containsAllMessage(messages)) return; + + logger?.info('Adding ${messages.length} messages'); final messageList = _messageQueue.toList(); - - _messageQueue.addAll(messages - .where((element) => !messageList.any((m) => m.id == element.id))); - - if (_messageQueue.isNotEmpty && !_isRetrying) { - _startRetrying(); - } + // we should not add message if already available in the queue + _messageQueue.addAll(messages.where( + (it) => !messageList.any((m) => m.id == it.id), + )); + _startRetrying(); } Future _startRetrying() async { - logger?.info('start retrying'); + if (_isRetrying) return; _isRetrying = true; - final retryPolicy = _retryPolicy!.copyWith(attempt: 0); + logger?.info('Started retrying failed messages'); while (_messageQueue.isNotEmpty) { + logger?.info('${_messageQueue.length} messages remaining in the queue'); final message = _messageQueue.first; - try { - logger?.info('retry attempt ${retryPolicy.attempt}'); - await _sendMessage(message); - logger?.info('message sent - removing it from the queue'); - _messageQueue.remove(message); - logger?.info('now ${_messageQueue.length} messages in the queue'); - retryPolicy.attempt = 0; - } catch (error) { - ApiError? apiError; - if (error is DioError) { - if (error.type == DioErrorType.response) { - _messageQueue.remove(message); - return; - } - apiError = ApiError( - error.response?.data, - error.response?.statusCode, - ); - } else if (error is ApiError) { - apiError = error; - if (apiError.status?.toString().startsWith('4') == true) { - _messageQueue.remove(message); - return; - } - } - - if (!retryPolicy.shouldRetry( - channel.client, - retryPolicy.attempt, - apiError, - )) { - _messageQueue.toList().forEach(_sendFailedEvent); - _isRetrying = false; - return; - } - - retryPolicy.attempt++; - - final timeout = retryPolicy.retryTimeout( - channel.client, - retryPolicy.attempt, - apiError, - ); - await Future.delayed(timeout); - } + await _runAndRetry(message); } _isRetrying = false; } - void _sendFailedEvent(Message? message) { - final newStatus = message!.status == MessageSendingStatus.sending - ? MessageSendingStatus.failed - : (message.status == MessageSendingStatus.updating - ? MessageSendingStatus.failed_update - : MessageSendingStatus.failed_delete); - channel.state!.addMessage(message.copyWith( - status: newStatus, - )); + Future _runAndRetry(Message message) async { + var attempt = 1; + + final maxAttempt = _retryPolicy.maxRetryAttempts; + // early return in case maxAttempt is less than 0 + if (attempt > maxAttempt) return; + + // ignore: literal_only_boolean_expressions + while (true) { + try { + logger?.info('Message (${message.id}) retry attempt $attempt'); + await _retryMessage(message); + logger?.info('Message (${message.id}) sent successfully'); + _messageQueue.removeMessage(message); + break; + } on StreamChatError catch (e) { + // retry logic + final maxAttempt = _retryPolicy.maxRetryAttempts; + if (attempt < maxAttempt) { + final shouldRetry = _retryPolicy.shouldRetry(client, attempt, e); + if (shouldRetry) { + final timeout = _retryPolicy.retryTimeout(client, attempt, e); + // temporary failure, continue + logger?.info( + 'API call failed (attempt $attempt), ' + 'retrying in ${timeout.inSeconds} seconds. Error was $e', + ); + await Future.delayed(timeout); + attempt += 1; + } else { + logger?.info( + 'API call failed (attempt $attempt). ' + 'Giving up for now, will retry when connection recovers. ' + 'Error was $e', + ); + _sendFailedEvent(message); + break; + } + } else { + logger?.info( + 'API call failed (attempt $attempt). ' + 'Exceeds maxRetryAttempt : $maxAttempt ' + 'Giving up for now, will retry when connection recovers. ' + 'Error was $e', + ); + _sendFailedEvent(message); + break; + } + } catch (e) { + logger?.info( + 'API call failed due to unknown error (attempt $attempt). ' + 'Giving up for now, will retry when connection recovers. ' + 'Error was $e', + ); + _sendFailedEvent(message); + break; + } + } } - Future _sendMessage(Message message) async { + void _sendFailedEvent(Message message) { + final newStatus = message.status == MessageSendingStatus.sending + ? MessageSendingStatus.failed + : message.status == MessageSendingStatus.updating + ? MessageSendingStatus.failed_update + : MessageSendingStatus.failed_delete; + channel.state?.addMessage(message.copyWith(status: newStatus)); + } + + Future _retryMessage(Message message) async { if (message.status == MessageSendingStatus.failed_update || message.status == MessageSendingStatus.updating) { await channel.updateMessage(message); @@ -137,39 +177,10 @@ class RetryQueue { } } - void _listenFailedEvents() { - _subscriptions.add(channel.on().listen((event) { - final messageList = _messageQueue.toList(); - if (event.message != null) { - final messageIndex = - messageList.indexWhere((m) => m.id == event.message!.id); - if (messageIndex == -1 && - [ - MessageSendingStatus.failed_update, - MessageSendingStatus.failed, - MessageSendingStatus.failed_delete, - ].contains(event.message!.status)) { - logger?.info('add message from events'); - final m = event.message; - - if (m != null) { - add([m]); - } - } else if (messageIndex != -1 && - [ - MessageSendingStatus.sent, - null, - ].contains(event.message!.status)) { - _messageQueue.remove(messageList[messageIndex]); - } - } - })); - } - /// Call this method to dispose this object void dispose() { _messageQueue.clear(); - _subscriptions.forEach((s) => s.cancel()); + _compositeSubscription.dispose(); } static int _byDate(Message m1, Message m2) { @@ -201,3 +212,27 @@ class RetryQueue { } } } + +extension _MessageHeapPriorityQueue on HeapPriorityQueue { + void removeMessage(Message message) { + final list = toUnorderedList(); + final index = list.indexWhere((it) => it.id == message.id); + if (index == -1) return; + final element = list[index]; + remove(element); + } + + bool containsMessage(Message message) { + final list = toUnorderedList(); + final index = list.indexWhere((it) => it.id == message.id); + if (index == -1) return false; + return true; + } + + bool containsAllMessage(List messages) { + if (isEmpty) return false; + final list = toUnorderedList(); + final messageIds = messages.map((it) => it.id); + return list.every((it) => messageIds.contains(it.id)); + } +} diff --git a/packages/stream_chat/lib/src/client.dart b/packages/stream_chat/lib/src/client.dart index cd986b2e..e7e05b69 100644 --- a/packages/stream_chat/lib/src/client.dart +++ b/packages/stream_chat/lib/src/client.dart @@ -7,6 +7,8 @@ import 'package:dio/dio.dart'; import 'package:logging/logging.dart'; import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/api/channel.dart'; +import 'package:stream_chat/src/core/utils.dart'; +import 'package:stream_chat/src/errors/stream_chat_error.dart'; import 'package:stream_chat/src/location.dart'; import 'package:stream_chat/src/ws/connection_status.dart'; import 'package:stream_chat/src/core/api/attachment_file_uploader.dart'; @@ -92,7 +94,7 @@ class StreamChatClient { tokenManager: _tokenManager, connectionIdManager: _connectionIdManager, attachmentFileUploader: attachmentFileUploader, - logger: _detachedLogger('πŸ•ΈοΈ'), + logger: detachedLogger('πŸ•ΈοΈ'), ); _ws = WebSocket( @@ -100,13 +102,13 @@ class StreamChatClient { baseUrl: options.baseUrl, tokenManager: _tokenManager, handler: handleEvent, - logger: _detachedLogger('πŸ”Œ'), + logger: detachedLogger('πŸ”Œ'), ); _retryPolicy = retryPolicy ?? RetryPolicy( - retryTimeout: (_, int attempt, __) => Duration(seconds: 1 * attempt), - shouldRetry: (_, int attempt, __) => attempt < 5, + shouldRetry: (_, attempt, __) => attempt < 5, + retryTimeout: (_, attempt, __) => Duration(seconds: attempt), ); state = ClientState(this); @@ -137,7 +139,7 @@ class StreamChatClient { /// Whether the chat persistence is available or not bool get persistenceEnabled => _chatPersistenceClient != null; - RetryPolicy? _retryPolicy; + late final RetryPolicy _retryPolicy; // sync state of the channels present inside state, defaults to false bool _synced = false; @@ -146,7 +148,7 @@ class StreamChatClient { DateTime? _lastSyncedAt; /// The retry policy options getter - RetryPolicy? get retryPolicy => _retryPolicy; + RetryPolicy get retryPolicy => _retryPolicy; /// By default the Chat client will write all messages with level Warn or /// Error to stdout. @@ -217,7 +219,8 @@ class StreamChatClient { if (record.stackTrace != null) print(record.stackTrace); }; - Logger _detachedLogger( + /// + Logger detachedLogger( String name, ) => Logger.detached(name) @@ -303,7 +306,7 @@ class StreamChatClient { TokenProvider? provider, }) async { if (_ws.connectionCompleter?.isCompleted == false) { - throw Exception( + throw const StreamChatError( 'User already getting connected, try calling `disconnectUser` ' 'before trying to connect again', ); @@ -346,11 +349,11 @@ class StreamChatClient { logger.info('Opening web-socket connection for ${user.id}'); if (wsConnectionStatus == ConnectionStatus.connecting) { - throw Exception('Connection already in progress for ${user.id}'); + throw StreamChatError('Connection already in progress for ${user.id}'); } if (wsConnectionStatus == ConnectionStatus.connected) { - throw Exception('Connection already connected for ${user.id}'); + throw StreamChatError('Connection already connected for ${user.id}'); } _wsConnectionStatus = ConnectionStatus.connecting; @@ -486,13 +489,6 @@ class StreamChatClient { } } - String _generateHash(List objects) { - final payload = json.encode(objects); - final payloadBytes = utf8.encode(payload); - final payloadB64 = base64.encode(payloadBytes); - return payloadB64; - } - final _queryChannelsStreams = >>{}; /// Requests channels with a given query. @@ -512,7 +508,7 @@ class StreamChatClient { watch = false; } - final hash = _generateHash([ + final hash = generateHash([ filter, sort, state, @@ -575,7 +571,7 @@ class StreamChatClient { await _ws.connectionCompleter?.future; } if (wsConnectionStatus != ConnectionStatus.connected) { - throw Exception( + throw const StreamChatError( 'You cannot use queryChannels without an active connection. ' 'Please call `connectUser` to connect the client.', ); diff --git a/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart b/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart index a75b2d2c..9dafc6c9 100644 --- a/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart +++ b/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart @@ -27,12 +27,12 @@ class AuthInterceptor extends Interceptor { try { token = await _tokenManager.loadToken(); } catch (_) { - final error = StreamChatError(ChatErrorCode.undefinedToken); + final error = StreamChatNetworkError(ChatErrorCode.undefinedToken); final dioError = StreamChatDioError( error: error, requestOptions: options, ); - return handler.reject(dioError); + return handler.reject(dioError, true); } final params = {'user_id': token.userId}; final headers = { diff --git a/packages/stream_chat/lib/src/core/http/stream_chat_dio_error.dart b/packages/stream_chat/lib/src/core/http/stream_chat_dio_error.dart index dd2c2535..f0e61111 100644 --- a/packages/stream_chat/lib/src/core/http/stream_chat_dio_error.dart +++ b/packages/stream_chat/lib/src/core/http/stream_chat_dio_error.dart @@ -17,5 +17,5 @@ class StreamChatDioError extends DioError { ); @override - final StreamChatError error; + final StreamChatNetworkError error; } diff --git a/packages/stream_chat/lib/src/core/http/stream_http_client.dart b/packages/stream_chat/lib/src/core/http/stream_http_client.dart index d814d705..0daeaacd 100644 --- a/packages/stream_chat/lib/src/core/http/stream_http_client.dart +++ b/packages/stream_chat/lib/src/core/http/stream_http_client.dart @@ -100,14 +100,15 @@ class StreamHttpClient { void close({bool force = false}) => httpClient.close(force: force); StreamChatNetworkError _parseError(DioError err) { + StreamChatNetworkError error; // locally thrown dio error if (err is StreamChatDioError) { - final code = err.error.code; - final message = err.error.message; - return StreamChatNetworkError.raw(code: code, message: message); + error = err.error; + } else { + // real network request dio error + error = StreamChatNetworkError.fromDioError(err); } - // real network request dio error - return StreamChatNetworkError.fromDioError(err); + return error..stackTrace = err.stackTrace; } /// Handy method to make http GET request with error parsing. diff --git a/packages/stream_chat/lib/src/core/models/event.dart b/packages/stream_chat/lib/src/core/models/event.dart index d0d17aa0..6ed1d414 100644 --- a/packages/stream_chat/lib/src/core/models/event.dart +++ b/packages/stream_chat/lib/src/core/models/event.dart @@ -29,7 +29,7 @@ class Event { this.parentId, this.extraData = const {}, this.isLocal = true, - }) : createdAt = createdAt ?? DateTime.now(); + }) : createdAt = createdAt?.toUtc() ?? DateTime.now().toUtc(); /// Create a new instance from a json factory Event.fromJson(Map json) => diff --git a/packages/stream_chat/lib/src/core/utils.dart b/packages/stream_chat/lib/src/core/utils.dart index 9366e34e..c2d77927 100644 --- a/packages/stream_chat/lib/src/core/utils.dart +++ b/packages/stream_chat/lib/src/core/utils.dart @@ -1,3 +1,4 @@ +import 'dart:convert'; import 'dart:math' as math; // This alphabet uses `A-Za-z0-9_-` symbols. The genetic algorithm helped @@ -14,3 +15,11 @@ String randomId({int size = 21}) { } return id; } + +/// Creates a hash string from the passed [objects] +String generateHash(List objects) { + final payload = json.encode(objects); + final payloadBytes = utf8.encode(payload); + final payloadB64 = base64.encode(payloadBytes); + return payloadB64; +} diff --git a/packages/stream_chat/lib/src/errors/chat_error_code.dart b/packages/stream_chat/lib/src/errors/chat_error_code.dart index 3dbb600c..6391428d 100644 --- a/packages/stream_chat/lib/src/errors/chat_error_code.dart +++ b/packages/stream_chat/lib/src/errors/chat_error_code.dart @@ -1,91 +1,166 @@ -/// -enum ChatErrorCode { - // client error codes - networkFailed, - parserError, - socketClosed, - socketFailure, - cantParseConnectionEvent, - cantParseEvent, - invalidToken, - undefinedToken, - unableToParseSocketEvent, - noErrorBody, +// ignore_for_file: lines_longer_than_80_chars - // server error codes +import 'package:collection/collection.dart'; + +/// Complete list of errors that are returned by the API +/// together with the description and API code. +enum ChatErrorCode { + // Client errors + + /// Unauthenticated, token not defined + undefinedToken, + + // Bad Request + + /// Wrong data/parameter is sent to the API + inputError, + + /// Duplicate username is sent while enforce_unique_usernames is enabled + duplicateUsername, + + /// Message is too long + messageTooLong, + + /// Event is not supported + eventNotSupported, + + /// The feature is currently disabled + /// on the dashboard (i.e. Reactions & Replies) + channelFeatureNotSupported, + + /// Multiple Levels Reply is not supported + /// the API only supports 1 level deep reply threads + multipleNestling, + + /// Custom Command handler returned an error + customCommandEndpointCall, + + /// App config does not have custom_action_handler_url + customCommandEndpointMissing, + + // Unauthorised + + /// Unauthenticated, problem with authentication authenticationError, + + /// Unauthenticated, token expired tokenExpired, + + /// Unauthenticated, token date incorrect + tokenBeforeIssuedAt, + + /// Unauthenticated, token not valid yet tokenNotValid, - tokenDateIncorrect, - tokenSignatureIncorrect, - apiKeyNotFound, + + /// Unauthenticated, token signature invalid + tokenSignatureInvalid, + + /// Access Key invalid + accessKeyError, + + // Forbidden + + /// Unauthorised / forbidden to make request + notAllowed, + + /// App suspended + appSuspended, + + /// User tried to post a message during the cooldown period + cooldownError, + + // Miscellaneous + + /// Resource not found + doesNotExist, + + /// Request timed out + requestTimeout, + + /// Payload too big + payloadTooBig, + + /// Too many requests in a certain time frame + rateLimitError, + + /// Request headers are too large + maximumHeaderSizeExceeded, + + /// Something goes wrong in the system + internalSystemError, + + /// No access to requested channels + noAccessToChannels } +const _errorCodeWithDescription = { + ChatErrorCode.undefinedToken: + MapEntry(1000, 'Unauthorised, token not defined'), + ChatErrorCode.inputError: + MapEntry(4, 'Wrong data/parameter is sent to the API'), + ChatErrorCode.duplicateUsername: MapEntry(6, + 'Duplicate username is sent while enforce_unique_usernames is enabled'), + ChatErrorCode.messageTooLong: MapEntry(20, 'Message is too long'), + ChatErrorCode.eventNotSupported: MapEntry(18, 'Event is not supported'), + ChatErrorCode.channelFeatureNotSupported: MapEntry(19, + 'The feature is currently disabled on the dashboard (i.e. Reactions & Replies)'), + ChatErrorCode.multipleNestling: MapEntry(21, + 'Multiple Levels Reply is not supported - the API only supports 1 level deep reply threads'), + ChatErrorCode.customCommandEndpointCall: + MapEntry(45, 'Custom Command handler returned an error'), + ChatErrorCode.customCommandEndpointMissing: + MapEntry(44, 'App config does not have custom_action_handler_url'), + ChatErrorCode.authenticationError: + MapEntry(5, 'Unauthenticated, problem with authentication'), + ChatErrorCode.tokenExpired: MapEntry(40, 'Unauthenticated, token expired'), + ChatErrorCode.tokenBeforeIssuedAt: + MapEntry(42, 'Unauthenticated, token date incorrect'), + ChatErrorCode.tokenNotValid: + MapEntry(41, 'Unauthenticated, token not valid yet'), + ChatErrorCode.tokenSignatureInvalid: + MapEntry(43, 'Unauthenticated, token signature invalid'), + ChatErrorCode.accessKeyError: MapEntry(2, 'Access Key invalid'), + ChatErrorCode.notAllowed: + MapEntry(17, 'Unauthorised / forbidden to make request'), + ChatErrorCode.appSuspended: MapEntry(99, 'App suspended'), + ChatErrorCode.cooldownError: + MapEntry(60, 'User tried to post a message during the cooldown period'), + ChatErrorCode.doesNotExist: MapEntry(16, 'Resource not found'), + ChatErrorCode.requestTimeout: MapEntry(23, 'Request timed out'), + ChatErrorCode.payloadTooBig: MapEntry(22, 'Payload too big'), + ChatErrorCode.rateLimitError: + MapEntry(9, 'Too many requests in a certain time frame'), + ChatErrorCode.maximumHeaderSizeExceeded: + MapEntry(24, 'Request headers are too large'), + ChatErrorCode.internalSystemError: + MapEntry(-1, 'Something goes wrong in the system'), + ChatErrorCode.noAccessToChannels: + MapEntry(70, 'No access to requested channels'), +}; + +const _authenticationErrors = [ + ChatErrorCode.undefinedToken, + ChatErrorCode.authenticationError, + ChatErrorCode.tokenExpired, + ChatErrorCode.tokenBeforeIssuedAt, + ChatErrorCode.tokenNotValid, + ChatErrorCode.tokenSignatureInvalid, + ChatErrorCode.accessKeyError, + ChatErrorCode.noAccessToChannels, +]; + +/// +ChatErrorCode? chatErrorCodeFromCode(int code) => _errorCodeWithDescription.keys + .firstWhereOrNull((key) => _errorCodeWithDescription[key]!.key == code); + /// extension ChatErrorCodeX on ChatErrorCode { /// - String get message => { - // client error message - ChatErrorCode.networkFailed: 'Response is failed. See cause', - ChatErrorCode.parserError: 'Unable to parse error', - ChatErrorCode.socketClosed: 'Server closed connection', - ChatErrorCode.socketFailure: 'See stack trace in logs. ' - 'Intercept error in error handler of setUser', - ChatErrorCode.cantParseConnectionEvent: - 'Unable to parse connection event', - ChatErrorCode.cantParseEvent: 'Unable to parse event', - ChatErrorCode.invalidToken: 'Invalid token', - ChatErrorCode.undefinedToken: - 'No defined token. Check if client.setUser was called and finished', - ChatErrorCode.unableToParseSocketEvent: - 'Socket event payload either invalid or null', - ChatErrorCode.noErrorBody: 'No error body. See http status code', - - // server error message - ChatErrorCode.authenticationError: - 'Unauthenticated, problem with authentication', - ChatErrorCode.tokenExpired: 'Token expired, new one must be requested.', - ChatErrorCode.tokenNotValid: 'Unauthenticated, token not valid yet', - ChatErrorCode.tokenDateIncorrect: - 'Unauthenticated, token date incorrect', - ChatErrorCode.tokenSignatureIncorrect: - 'Unauthenticated, token signature invalid', - ChatErrorCode.apiKeyNotFound: - "Api key is not found, verify it if it's correct or was created.", - }[this]!; + String get message => _errorCodeWithDescription[this]!.value; /// - int get code => { - // client error codes - ChatErrorCode.networkFailed: 1000, - ChatErrorCode.parserError: 1001, - ChatErrorCode.socketClosed: 1002, - ChatErrorCode.socketFailure: 1003, - ChatErrorCode.cantParseConnectionEvent: 1004, - ChatErrorCode.cantParseEvent: 1005, - ChatErrorCode.invalidToken: 1006, - ChatErrorCode.undefinedToken: 1007, - ChatErrorCode.unableToParseSocketEvent: 1008, - ChatErrorCode.noErrorBody: 1009, - - // server error codes - ChatErrorCode.authenticationError: 5, - ChatErrorCode.tokenExpired: 40, - ChatErrorCode.tokenNotValid: 41, - ChatErrorCode.tokenDateIncorrect: 42, - ChatErrorCode.tokenSignatureIncorrect: 43, - ChatErrorCode.apiKeyNotFound: 2, - }[this]!; + int get code => _errorCodeWithDescription[this]!.key; /// - Set get authenticationErrors => { - ChatErrorCode.authenticationError, - ChatErrorCode.tokenExpired, - ChatErrorCode.tokenNotValid, - ChatErrorCode.tokenDateIncorrect, - ChatErrorCode.tokenSignatureIncorrect, - }.map((it) => it.code).toSet(); - - /// - bool isAuthenticationError(int code) => authenticationErrors.contains(code); + bool get isAuthenticationError => _authenticationErrors.contains(this); } diff --git a/packages/stream_chat/lib/src/errors/stream_chat_error.dart b/packages/stream_chat/lib/src/errors/stream_chat_error.dart index 280ab6c7..a057783a 100644 --- a/packages/stream_chat/lib/src/errors/stream_chat_error.dart +++ b/packages/stream_chat/lib/src/errors/stream_chat_error.dart @@ -3,45 +3,74 @@ import 'package:stream_chat/src/errors/chat_error_code.dart'; import 'package:stream_chat/stream_chat.dart'; /// -class StreamChatError extends Equatable implements Exception { +class StreamChatError with EquatableMixin implements Exception { /// - StreamChatError(ChatErrorCode errorCode) - : code = errorCode.code, - message = errorCode.message; - - /// - const StreamChatError.raw(this.code, this.message); - - /// Error code - final int code; + const StreamChatError(this.message); /// Error message final String message; @override - List get props => [code, message]; + List get props => [message]; @override - String toString() => 'StreamChatError(code: $code, message: $message)'; + String toString() => 'StreamChatError(message: $message)'; +} + +/// +class StreamWebSocketError extends StreamChatError { + /// + const StreamWebSocketError( + String message, { + this.data, + }) : super(message); + + /// + factory StreamWebSocketError.fromStreamError(Map error) { + final data = ErrorResponse.fromJson(error); + final message = data.message ?? ''; + return StreamWebSocketError(message, data: data); + } + + /// + int? get code => data?.code; + + /// + ChatErrorCode? get errorCode { + final code = this.code; + if (code == null) return null; + return chatErrorCodeFromCode(code); + } + + /// Response body. please refer to [ErrorResponse]. + final ErrorResponse? data; + + @override + String toString() { + var params = 'message: $message'; + if (data != null) params += ', data: $data'; + return 'WebSocketError($params)'; + } } /// class StreamChatNetworkError extends StreamChatError { /// - StreamChatNetworkError({ - required ChatErrorCode errorCode, + StreamChatNetworkError( + ChatErrorCode errorCode, { int? statusCode, this.data, - }) : statusCode = statusCode ?? data?.statusCode, - super(errorCode); + }) : code = errorCode.code, + statusCode = statusCode ?? data?.statusCode, + super(errorCode.message); /// - const StreamChatNetworkError.raw({ - required int code, + StreamChatNetworkError.raw({ + required this.code, required String message, this.statusCode, this.data, - }) : super.raw(code, message); + }) : super(message); /// factory StreamChatNetworkError.fromDioError(DioError error) { @@ -53,25 +82,46 @@ class StreamChatNetworkError extends StreamChatError { } return StreamChatNetworkError.raw( code: errorResponse?.code ?? -1, - message: errorResponse?.message ?? response?.statusMessage ?? '', + message: + errorResponse?.message ?? response?.statusMessage ?? error.message, statusCode: errorResponse?.statusCode ?? response?.statusCode, data: errorResponse, - ); + )..stackTrace = error.stackTrace; } + /// Error code + final int code; + /// HTTP status code final int? statusCode; /// Response body. please refer to [ErrorResponse]. final ErrorResponse? data; - @override - List get props => [...super.props, statusCode]; + StackTrace? _stackTrace; + + /// + set stackTrace(StackTrace? stack) => _stackTrace = stack; + + /// + ChatErrorCode? get errorCode => chatErrorCodeFromCode(code); + + /// + bool get isRetriable => data == null; @override - String toString() => 'StreamChatNetworkError(' - 'code: $code, ' - 'message: $message, ' - 'statusCode: $statusCode, ' - 'data: $data)'; + List get props => [...super.props, code, statusCode]; + + @override + String toString({bool printStackTrace = false}) { + var params = 'code: $code, message: $message'; + if (statusCode != null) params += ', statusCode: $statusCode'; + if (data != null) params += ', data: $data'; + var msg = 'StreamChatNetworkError($params)'; + + if (printStackTrace && _stackTrace != null) { + msg += '\n$_stackTrace'; + } + return msg; + } } diff --git a/packages/stream_chat/lib/src/exceptions.dart b/packages/stream_chat/lib/src/exceptions.dart deleted file mode 100644 index 714bb386..00000000 --- a/packages/stream_chat/lib/src/exceptions.dart +++ /dev/null @@ -1,53 +0,0 @@ -import 'dart:convert'; - -/// Exception related to api calls -class ApiError extends Error { - /// Creates a new ApiError instance using the response body and status code - ApiError(this.body, this.status) : jsonData = _decode(body) { - if (jsonData != null && jsonData!.containsKey('code')) { - _code = jsonData!['code']; - } - } - - /// Raw body of the response - final String? body; - - /// Json parsed body - final Map? jsonData; - - /// Http status code of the response - final int? status; - - /// Stream specific error code - int? get code => _code; - int? _code; - - static Map? _decode(String? body) { - try { - if (body == null) { - return null; - } - return json.decode(body); - } on FormatException { - return null; - } - } - - @override - bool operator ==(Object other) => - identical(this, other) || - other is ApiError && - runtimeType == other.runtimeType && - body == other.body && - jsonData == other.jsonData && - status == other.status && - _code == other._code; - - @override - int get hashCode => - body.hashCode ^ jsonData.hashCode ^ status.hashCode ^ _code.hashCode; - - @override - String toString() => 'ApiError{body: $body, jsonData: $jsonData, ' - 'status: $status, code: $_code}'; -} diff --git a/packages/stream_chat/lib/src/ws/socket_error.dart b/packages/stream_chat/lib/src/ws/socket_error.dart deleted file mode 100644 index 00b052b9..00000000 --- a/packages/stream_chat/lib/src/ws/socket_error.dart +++ /dev/null @@ -1,25 +0,0 @@ -/// -class SocketError { - /// - const SocketError({ - this.code = -1, - this.statusCode = -1, - this.message = '', - }); - - /// - factory SocketError.fromJson(Map json) => SocketError( - code: json['code'], - statusCode: json['StatusCode'], - message: json['message'], - ); - - /// - final int code; - - /// - final int statusCode; - - /// - final String message; -} diff --git a/packages/stream_chat/lib/src/ws/websocket.dart b/packages/stream_chat/lib/src/ws/websocket.dart index 6ebb7a7b..6c36fd0b 100644 --- a/packages/stream_chat/lib/src/ws/websocket.dart +++ b/packages/stream_chat/lib/src/ws/websocket.dart @@ -5,13 +5,14 @@ import 'dart:math' as math; import 'package:logging/logging.dart'; import 'package:meta/meta.dart'; import 'package:rxdart/rxdart.dart'; +import 'package:stream_chat/src/errors/chat_error_code.dart'; +import 'package:stream_chat/src/errors/stream_chat_error.dart'; import 'package:stream_chat/src/ws/connection_status.dart'; import 'package:stream_chat/src/core/http/token.dart'; import 'package:stream_chat/src/core/http/token_manager.dart'; import 'package:stream_chat/src/core/models/event.dart'; import 'package:stream_chat/src/core/models/user.dart'; import 'package:stream_chat/src/event_type.dart'; -import 'package:stream_chat/src/ws/socket_error.dart'; import 'package:stream_chat/src/ws/timer_helper.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; import 'package:web_socket_channel/status.dart' as status; @@ -27,8 +28,6 @@ typedef WebSocketChannelProvider = WebSocketChannel Function( Iterable? protocols, }); -const _tokenExpiredErrorCode = 40; - /// A WebSocket connection that reconnects upon failure. class WebSocket with TimerHelper { /// Creates a new websocket @@ -174,7 +173,7 @@ class WebSocket with TimerHelper { /// Connect the WS using the parameters passed in the constructor Future connect(User user) async { if (_connectRequestInProgress) { - throw Exception(''' + throw const StreamWebSocketError(''' You've called connect twice, can only attempt 1 connection at the time, '''); @@ -195,7 +194,7 @@ class WebSocket with TimerHelper { int _reconnectAttempt = 0; bool _reconnectRequestInProgress = false; - Future _reconnect({bool refreshToken = false}) async { + void _reconnect({bool refreshToken = false}) async { _logger?.info('Retrying connection : $_reconnectAttempt'); if (_reconnectRequestInProgress) return; _reconnectRequestInProgress = true; @@ -270,10 +269,35 @@ class WebSocket with TimerHelper { _connectionStatus = ConnectionStatus.connected; } + void _handleStreamError(Map errorResponse) { + // resetting connect, reconnect request flag + _resetRequestFlags(); + + final error = StreamWebSocketError.fromStreamError(errorResponse); + final isTokenExpired = error.errorCode == ChatErrorCode.tokenExpired; + if (isTokenExpired && !tokenManager.isStatic) { + _logger?.warning('Connection failed, token expired'); + return _reconnect(refreshToken: true); + } + + _logger?.severe('Connection failed', error); + + final completer = connectionCompleter; + // complete with error if not yet completed + if (completer != null && !completer.isCompleted) { + // complete the connection with error + completer.completeError(error); + // disconnect the web-socket connection + return disconnect(); + } + + return _reconnect(); + } + void _onDataReceived(dynamic data) { - final jsonData = json.decode(data); - final error = jsonData['error']; - if (error != null) return _onConnectionError(error); + final jsonData = json.decode(data) as Map; + final error = jsonData['error'] as Map?; + if (error != null) return _handleStreamError(error); // resetting connect, reconnect request flag _resetRequestFlags(resetAttempts: true); @@ -300,25 +324,18 @@ class WebSocket with TimerHelper { } void _onConnectionError(error, [stacktrace]) { - _logger?.severe('Error occurred', error, stacktrace); + _logger?.warning('Error occurred', error, stacktrace); // resetting connect, reconnect request flag _resetRequestFlags(); - var refreshToken = false; - try { - final socketError = SocketError.fromJson(json.decode(error)); - refreshToken = socketError.code == _tokenExpiredErrorCode; - } catch (_) {} - - // refresh token in case it is expired - _reconnect(refreshToken: refreshToken); + _reconnect(); } bool _manuallyClosed = false; void _onConnectionClosed() { - _logger?.info('Connection closed : $connectionId'); + _logger?.warning('Connection closed : $connectionId'); // resetting connect, reconnect request flag _resetRequestFlags(); @@ -381,7 +398,7 @@ class WebSocket with TimerHelper { if (connectionStatus == ConnectionStatus.disconnected) return; _connectionStatus = ConnectionStatus.disconnected; - _logger?.info('Disconnecting $connectionId'); + _logger?.info('Disconnecting web-socket connection'); // resetting user _user = null; diff --git a/packages/stream_chat/lib/stream_chat.dart b/packages/stream_chat/lib/stream_chat.dart index 991e721b..66dace1d 100644 --- a/packages/stream_chat/lib/stream_chat.dart +++ b/packages/stream_chat/lib/stream_chat.dart @@ -32,8 +32,9 @@ export './src/core/models/reaction.dart'; export './src/core/models/read.dart'; export './src/core/models/user.dart'; export './src/db/chat_persistence_client.dart'; +export './src/errors/chat_error_code.dart'; +export './src/errors/stream_chat_error.dart'; export './src/event_type.dart'; -export './src/exceptions.dart'; export './src/extensions/rate_limit.dart'; export './src/extensions/string_extension.dart'; export './src/location.dart'; diff --git a/packages/stream_chat/test/src/client_test.dart b/packages/stream_chat/test/src/client_test.dart index 8da187a4..13506f66 100644 --- a/packages/stream_chat/test/src/client_test.dart +++ b/packages/stream_chat/test/src/client_test.dart @@ -8,7 +8,6 @@ import 'package:logging/logging.dart'; import 'package:mocktail/mocktail.dart'; import 'package:stream_chat/src/core/api/requests.dart'; import 'package:stream_chat/src/client.dart'; -import 'package:stream_chat/src/exceptions.dart'; import 'package:stream_chat/src/core/models/channel_model.dart'; import 'package:stream_chat/src/core/models/filter.dart'; import 'package:stream_chat/src/core/models/message.dart'; diff --git a/packages/stream_chat_flutter/lib/src/message_actions_modal.dart b/packages/stream_chat_flutter/lib/src/message_actions_modal.dart index 5336ed48..e40cfc21 100644 --- a/packages/stream_chat_flutter/lib/src/message_actions_modal.dart +++ b/packages/stream_chat_flutter/lib/src/message_actions_modal.dart @@ -341,7 +341,8 @@ class _MessageActionsModalState extends State { okText: 'OK', ); } catch (err) { - if (err is ApiError && json.decode(err.body ?? '{}')['code'] == 4) { + if (err is StreamChatNetworkError && + err.errorCode == ChatErrorCode.inputError) { await showInfoDialog( context, icon: StreamSvgIcon.flag( diff --git a/packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart b/packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart index efc8610b..447533e0 100644 --- a/packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart +++ b/packages/stream_chat_persistence/lib/src/db/moor_chat_database.dart @@ -51,7 +51,7 @@ class MoorChatDatabase extends _$MoorChatDatabase { // you should bump this number whenever you change or add a table definition. @override - int get schemaVersion => 3; + int get schemaVersion => 4; @override MigrationStrategy get migration => MigrationStrategy(