refactor: deprecate MessageSendingStatus in favor of MessageState.
Signed-off-by: xsahil03x <[email protected]>
This commit is contained in:
Sahil Kumar
@@ -582,7 +582,7 @@ class Channel {
|
||||
// Eg. Updating the message while the previous call is in progress.
|
||||
_messageAttachmentsUploadCompleter
|
||||
.remove(message.id)
|
||||
?.completeError('Message Cancelled');
|
||||
?.completeError(const StreamChatError('Message deleted'));
|
||||
|
||||
final quotedMessage = state!.messages.firstWhereOrNull(
|
||||
(m) => m.id == message.quotedMessageId,
|
||||
@@ -592,7 +592,7 @@ class Channel {
|
||||
localCreatedAt: DateTime.now(),
|
||||
user: _client.state.currentUser,
|
||||
quotedMessage: quotedMessage,
|
||||
status: MessageSendingStatus.sending,
|
||||
state: MessageState.sending,
|
||||
attachments: message.attachments.map(
|
||||
(it) {
|
||||
if (it.uploadState.isSuccess) return it;
|
||||
@@ -630,15 +630,24 @@ class Channel {
|
||||
),
|
||||
);
|
||||
|
||||
final sentMessage = response.message.syncWith(message);
|
||||
final sentMessage = response.message.syncWith(message).copyWith(
|
||||
// Update the message state to updated.
|
||||
state: MessageState.sent,
|
||||
);
|
||||
|
||||
state!.updateMessage(sentMessage);
|
||||
if (cooldown > 0) cooldownStartedAt = DateTime.now();
|
||||
return response;
|
||||
} catch (e) {
|
||||
if (e is StreamChatNetworkError && e.isRetriable) {
|
||||
state!._retryQueue.add([message]);
|
||||
state!._retryQueue.add([
|
||||
message.copyWith(
|
||||
// Update the message state to failed.
|
||||
state: MessageState.sendingFailed,
|
||||
),
|
||||
]);
|
||||
}
|
||||
|
||||
rethrow;
|
||||
}
|
||||
}
|
||||
@@ -653,17 +662,18 @@ class Channel {
|
||||
Message message, {
|
||||
bool skipEnrichUrl = false,
|
||||
}) async {
|
||||
_checkInitialized();
|
||||
final originalMessage = message;
|
||||
|
||||
// Cancelling previous completer in case it's called again in the process
|
||||
// Eg. Updating the message while the previous call is in progress.
|
||||
_messageAttachmentsUploadCompleter
|
||||
.remove(message.id)
|
||||
?.completeError('Message Cancelled');
|
||||
?.completeError(const StreamChatError('Message deleted'));
|
||||
|
||||
// ignore: parameter_assignments
|
||||
message = message.copyWith(
|
||||
status: MessageSendingStatus.updating,
|
||||
state: MessageState.updating,
|
||||
localUpdatedAt: DateTime.now(),
|
||||
attachments: message.attachments.map(
|
||||
(it) {
|
||||
@@ -699,19 +709,30 @@ class Channel {
|
||||
),
|
||||
);
|
||||
|
||||
final updatedMessage = response.message
|
||||
.syncWith(message)
|
||||
.copyWith(ownReactions: message.ownReactions);
|
||||
final updateMessage = response.message.syncWith(message).copyWith(
|
||||
// Update the message state to updated.
|
||||
state: MessageState.updated,
|
||||
ownReactions: message.ownReactions,
|
||||
);
|
||||
|
||||
state?.updateMessage(updatedMessage);
|
||||
state?.updateMessage(updateMessage);
|
||||
|
||||
return response;
|
||||
} catch (e) {
|
||||
if (e is StreamChatNetworkError) {
|
||||
if (e.isRetriable) {
|
||||
state!._retryQueue.add([message]);
|
||||
state!._retryQueue.add([
|
||||
message.copyWith(
|
||||
// Update the message state to failed.
|
||||
state: MessageState.updatingFailed,
|
||||
),
|
||||
]);
|
||||
} else {
|
||||
state?.updateMessage(originalMessage);
|
||||
// Reset the message to original state if the update fails and is not
|
||||
// retriable.
|
||||
state?.updateMessage(originalMessage.copyWith(
|
||||
state: MessageState.updatingFailed,
|
||||
));
|
||||
}
|
||||
}
|
||||
rethrow;
|
||||
@@ -729,6 +750,23 @@ class Channel {
|
||||
List<String>? unset,
|
||||
bool skipEnrichUrl = false,
|
||||
}) async {
|
||||
_checkInitialized();
|
||||
final originalMessage = message;
|
||||
|
||||
// Cancelling previous completer in case it's called again in the process
|
||||
// Eg. Updating the message while the previous call is in progress.
|
||||
_messageAttachmentsUploadCompleter
|
||||
.remove(message.id)
|
||||
?.completeError(const StreamChatError('Message deleted'));
|
||||
|
||||
// ignore: parameter_assignments
|
||||
message = message.copyWith(
|
||||
state: MessageState.updating,
|
||||
localUpdatedAt: DateTime.now(),
|
||||
);
|
||||
|
||||
state?.updateMessage(message);
|
||||
|
||||
try {
|
||||
// Wait for the previous update call to finish. Otherwise, the order of
|
||||
// messages will not be maintained.
|
||||
@@ -741,17 +779,33 @@ class Channel {
|
||||
),
|
||||
);
|
||||
|
||||
final updatedMessage = response.message
|
||||
.syncWith(message)
|
||||
.copyWith(ownReactions: message.ownReactions);
|
||||
final updatedMessage = response.message.syncWith(message).copyWith(
|
||||
// Update the message state to updated.
|
||||
state: MessageState.updated,
|
||||
ownReactions: message.ownReactions,
|
||||
);
|
||||
|
||||
state?.updateMessage(updatedMessage);
|
||||
|
||||
return response;
|
||||
} catch (e) {
|
||||
if (e is StreamChatNetworkError && e.isRetriable) {
|
||||
state!._retryQueue.add([message]);
|
||||
if (e is StreamChatNetworkError) {
|
||||
if (e.isRetriable) {
|
||||
state!._retryQueue.add([
|
||||
message.copyWith(
|
||||
// Update the message state to failed.
|
||||
state: MessageState.updatingFailed,
|
||||
),
|
||||
]);
|
||||
} else {
|
||||
// Reset the message to original state if the update fails and is not
|
||||
// retriable.
|
||||
state?.updateMessage(originalMessage.copyWith(
|
||||
state: MessageState.updatingFailed,
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
rethrow;
|
||||
}
|
||||
}
|
||||
@@ -759,39 +813,49 @@ class Channel {
|
||||
final _deleteMessageLock = Lock();
|
||||
|
||||
/// Deletes the [message] from the channel.
|
||||
Future<EmptyResponse> deleteMessage(Message message, {bool? hard}) async {
|
||||
final hardDelete = hard ?? false;
|
||||
Future<EmptyResponse> deleteMessage(
|
||||
Message message, {
|
||||
bool hard = false,
|
||||
}) async {
|
||||
_checkInitialized();
|
||||
|
||||
// Directly deleting the local messages which are not yet sent to server
|
||||
if (message.status == MessageSendingStatus.sending ||
|
||||
message.status == MessageSendingStatus.failed) {
|
||||
// Directly deleting the local messages which are not yet sent to server.
|
||||
final isSynced = message.state.isCompleted ||
|
||||
message.state.isUpdating ||
|
||||
message.state.isDeleting ||
|
||||
message.state.isUpdatingFailed ||
|
||||
message.state.isDeletingFailed;
|
||||
|
||||
if (!isSynced) {
|
||||
state!.deleteMessage(
|
||||
message.copyWith(
|
||||
type: 'deleted',
|
||||
localDeletedAt: DateTime.now(),
|
||||
status: MessageSendingStatus.sent,
|
||||
state: MessageState.deleted(hard: hard),
|
||||
),
|
||||
hardDelete: hardDelete,
|
||||
hardDelete: hard,
|
||||
);
|
||||
|
||||
// Removing the attachments upload completer to stop the `sendMessage`
|
||||
// waiting for attachments to complete.
|
||||
_messageAttachmentsUploadCompleter
|
||||
.remove(message.id)
|
||||
?.completeError(Exception('Message deleted'));
|
||||
?.completeError(const StreamChatError('Message deleted'));
|
||||
|
||||
// Returning empty response to mark the api call as success.
|
||||
return EmptyResponse();
|
||||
}
|
||||
|
||||
// ignore: parameter_assignments
|
||||
message = message.copyWith(
|
||||
type: 'deleted',
|
||||
deletedAt: DateTime.now(),
|
||||
state: MessageState.deleting(hard: hard),
|
||||
);
|
||||
|
||||
state?.deleteMessage(message, hardDelete: hard);
|
||||
|
||||
try {
|
||||
// ignore: parameter_assignments
|
||||
message = message.copyWith(
|
||||
type: 'deleted',
|
||||
status: MessageSendingStatus.deleting,
|
||||
deletedAt: message.deletedAt ?? DateTime.now(),
|
||||
);
|
||||
|
||||
state?.deleteMessage(message, hardDelete: hardDelete);
|
||||
|
||||
// Wait for the previous delete call to finish. Otherwise, the order of
|
||||
// messages will not be maintained.
|
||||
final response = await _deleteMessageLock.synchronized(
|
||||
@@ -799,15 +863,20 @@ class Channel {
|
||||
);
|
||||
|
||||
final deletedMessage = message.copyWith(
|
||||
status: MessageSendingStatus.sent,
|
||||
state: MessageState.deleted(hard: hard),
|
||||
);
|
||||
|
||||
state?.deleteMessage(deletedMessage, hardDelete: hardDelete);
|
||||
state?.deleteMessage(deletedMessage, hardDelete: hard);
|
||||
|
||||
return response;
|
||||
} catch (e) {
|
||||
if (e is StreamChatNetworkError && e.isRetriable) {
|
||||
state!._retryQueue.add([message]);
|
||||
state!._retryQueue.add([
|
||||
message.copyWith(
|
||||
// Update the message state to failed.
|
||||
state: MessageState.deletingFailed(hard: hard),
|
||||
),
|
||||
]);
|
||||
}
|
||||
rethrow;
|
||||
}
|
||||
@@ -1877,15 +1946,7 @@ class ChannelClientState {
|
||||
/// Retry failed message.
|
||||
Future<void> retryFailedMessages() async {
|
||||
final failedMessages = [...messages, ...threads.values.expand((v) => v)]
|
||||
.where(
|
||||
(message) =>
|
||||
message.status != MessageSendingStatus.sent &&
|
||||
message.createdAt.isBefore(
|
||||
DateTime.now().subtract(const Duration(seconds: 5)),
|
||||
),
|
||||
)
|
||||
.toList();
|
||||
|
||||
.where((it) => it.state.isFailed);
|
||||
_retryQueue.add(failedMessages);
|
||||
}
|
||||
|
||||
|
||||
@@ -109,8 +109,9 @@ class StreamChatClient {
|
||||
|
||||
_retryPolicy = retryPolicy ??
|
||||
RetryPolicy(
|
||||
shouldRetry: (_, attempt, __) => attempt < 5,
|
||||
retryTimeout: (_, attempt, __) => Duration(seconds: attempt),
|
||||
shouldRetry: (_, __, error) {
|
||||
return error is StreamChatNetworkError && error.isRetriable;
|
||||
},
|
||||
);
|
||||
|
||||
state = ClientState(this);
|
||||
@@ -1397,12 +1398,19 @@ class StreamChatClient {
|
||||
);
|
||||
|
||||
/// Deletes the given message
|
||||
Future<EmptyResponse> deleteMessage(String messageId, {bool? hard}) async {
|
||||
final response =
|
||||
await _chatApi.message.deleteMessage(messageId, hard: hard);
|
||||
if (hard == true) {
|
||||
Future<EmptyResponse> deleteMessage(
|
||||
String messageId, {
|
||||
bool hard = false,
|
||||
}) async {
|
||||
final response = await _chatApi.message.deleteMessage(
|
||||
messageId,
|
||||
hard: hard,
|
||||
);
|
||||
|
||||
if (hard) {
|
||||
await chatPersistenceClient?.deleteMessageById(messageId);
|
||||
}
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,27 +1,54 @@
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:stream_chat/src/client/client.dart';
|
||||
import 'package:stream_chat/src/core/error/error.dart';
|
||||
|
||||
/// The retry options
|
||||
/// When sending/updating/deleting a message any temporary error will trigger the retry policy
|
||||
/// The retry policy exposes 2 methods
|
||||
/// - shouldRetry: returns a boolean if the request should be retried
|
||||
/// - retryTimeout: How many milliseconds to wait till the next attempt
|
||||
/// The retry policy associated to a client.
|
||||
///
|
||||
/// maxRetryAttempts is a hard limit on maximum retry attempts before giving up
|
||||
/// This policy is used to determine if a request should be retried and when.
|
||||
///
|
||||
/// also see:
|
||||
/// - [RetryQueue]
|
||||
class RetryPolicy {
|
||||
/// Instantiate a new RetryPolicy
|
||||
RetryPolicy({
|
||||
required this.shouldRetry,
|
||||
required this.retryTimeout,
|
||||
@Deprecated("Use 'delayFactor' instead.") this.retryTimeout,
|
||||
this.maxRetryAttempts = 6,
|
||||
this.delayFactor = const Duration(milliseconds: 200),
|
||||
this.randomizationFactor = 0.25,
|
||||
this.maxDelay = const Duration(seconds: 30),
|
||||
});
|
||||
|
||||
/// Hard limit on maximum retry attempts before giving up, defaults to 6
|
||||
/// Resets once connection recovers.
|
||||
/// Delay factor to double after every attempt.
|
||||
///
|
||||
/// Defaults to 200 ms, which results in the following delays:
|
||||
///
|
||||
/// 1. 400 ms
|
||||
/// 2. 800 ms
|
||||
/// 3. 1600 ms
|
||||
/// 4. 3200 ms
|
||||
/// 5. 6400 ms
|
||||
/// 6. 12800 ms
|
||||
///
|
||||
/// Before application of [randomizationFactor].
|
||||
final Duration delayFactor;
|
||||
|
||||
/// Percentage the delay should be randomized, given as fraction between
|
||||
/// 0 and 1.
|
||||
///
|
||||
/// If [randomizationFactor] is `0.25` (default) this indicates 25 % of the
|
||||
/// delay should be increased or decreased by 25 %.
|
||||
final double randomizationFactor;
|
||||
|
||||
/// Maximum delay between retries, defaults to 30 seconds.
|
||||
final Duration maxDelay;
|
||||
|
||||
/// Maximum number of attempts before giving up, defaults to 6.
|
||||
final int maxRetryAttempts;
|
||||
|
||||
/// This function evaluates if we should retry the failure
|
||||
final bool Function(
|
||||
/// Function to determine if a retry should be attempted.
|
||||
final FutureOr<bool> Function(
|
||||
StreamChatClient client,
|
||||
int attempt,
|
||||
StreamChatError? error,
|
||||
@@ -29,9 +56,10 @@ class RetryPolicy {
|
||||
|
||||
/// In the case that we want to retry a failed request the retryTimeout
|
||||
/// method is called to determine the timeout
|
||||
@Deprecated("Use 'delayFactor' instead.")
|
||||
final Duration Function(
|
||||
StreamChatClient client,
|
||||
int attempt,
|
||||
StreamChatError? error,
|
||||
) retryTimeout;
|
||||
)? retryTimeout;
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ class RetryQueue {
|
||||
}) : client = channel.client {
|
||||
_retryPolicy = client.retryPolicy;
|
||||
_listenConnectionRecovered();
|
||||
_listenFailedEvents();
|
||||
_listenMessageEvents();
|
||||
}
|
||||
|
||||
/// The channel of this queue.
|
||||
@@ -31,153 +31,105 @@ class RetryQueue {
|
||||
final _compositeSubscription = CompositeSubscription();
|
||||
|
||||
final _messageQueue = HeapPriorityQueue(_byDate);
|
||||
bool _isRetrying = false;
|
||||
|
||||
void _listenConnectionRecovered() {
|
||||
client.on(EventType.connectionRecovered).listen((event) {
|
||||
client.on(EventType.connectionRecovered).distinct().listen((event) {
|
||||
if (event.online == true) {
|
||||
_startRetrying();
|
||||
logger?.info('Connection recovered, retrying failed messages');
|
||||
channel.state?.retryFailedMessages();
|
||||
}
|
||||
}).addTo(_compositeSubscription);
|
||||
}
|
||||
|
||||
void _listenFailedEvents() {
|
||||
void _listenMessageEvents() {
|
||||
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) {
|
||||
|
||||
if (message.state.isCompleted) {
|
||||
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]);
|
||||
}
|
||||
return _messageQueue.removeMessage(message);
|
||||
}
|
||||
}).addTo(_compositeSubscription);
|
||||
}
|
||||
|
||||
/// Add a list of messages.
|
||||
void add(List<Message> messages) {
|
||||
if (messages.isEmpty) return;
|
||||
if (!_messageQueue.containsAllMessage(messages)) {
|
||||
logger?.info('Adding ${messages.length} messages');
|
||||
final messageList = _messageQueue.toList();
|
||||
// we should not add message if already available in the queue
|
||||
_messageQueue.addAll(messages.where(
|
||||
(it) => !messageList.any((m) => m.id == it.id),
|
||||
));
|
||||
}
|
||||
void add(Iterable<Message> messages) {
|
||||
assert(
|
||||
messages.every((it) => it.state.isFailed),
|
||||
'Only failed messages can be added to the queue',
|
||||
);
|
||||
|
||||
_startRetrying();
|
||||
// Filter out messages that are already in the queue.
|
||||
final messagesToAdd = messages.where((it) {
|
||||
return !_messageQueue.containsMessage(it);
|
||||
});
|
||||
|
||||
// If there are no messages to add, return.
|
||||
if (messagesToAdd.isEmpty) return;
|
||||
|
||||
logger?.info('Adding ${messagesToAdd.length} messages to the queue');
|
||||
_messageQueue.addAll(messagesToAdd);
|
||||
|
||||
_processQueue();
|
||||
}
|
||||
|
||||
Future<void> _startRetrying() async {
|
||||
if (_isRetrying) return;
|
||||
_isRetrying = true;
|
||||
bool _isProcessing = false;
|
||||
|
||||
Future<void> _processQueue() async {
|
||||
if (_isProcessing) return;
|
||||
_isProcessing = true;
|
||||
|
||||
logger?.info('Started retrying failed messages');
|
||||
while (_messageQueue.isNotEmpty) {
|
||||
logger?.info('${_messageQueue.length} messages remaining in the queue');
|
||||
|
||||
final message = _messageQueue.first;
|
||||
final succeeded = await _runAndRetry(message);
|
||||
if (!succeeded) {
|
||||
_messageQueue.toList().forEach(_sendFailedEvent);
|
||||
break;
|
||||
}
|
||||
}
|
||||
_isRetrying = false;
|
||||
}
|
||||
|
||||
Future<bool> _runAndRetry(Message message) async {
|
||||
var attempt = 1;
|
||||
|
||||
final maxAttempt = _retryPolicy.maxRetryAttempts;
|
||||
// early return in case maxAttempt is less than 0
|
||||
if (attempt > maxAttempt) return false;
|
||||
|
||||
// ignore: literal_only_boolean_expressions
|
||||
while (true) {
|
||||
final retryPolicy = _retryPolicy;
|
||||
try {
|
||||
logger?.info('Message (${message.id}) retry attempt $attempt');
|
||||
await _retryMessage(message);
|
||||
logger?.info('Message (${message.id}) sent successfully');
|
||||
_messageQueue.removeMessage(message);
|
||||
return true;
|
||||
} catch (e) {
|
||||
if (e is! StreamChatNetworkError || !e.isRetriable) {
|
||||
_messageQueue.removeMessage(message);
|
||||
_sendFailedEvent(message);
|
||||
return true;
|
||||
}
|
||||
// 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;
|
||||
}
|
||||
await backOff(
|
||||
() => _retryMessage(message),
|
||||
delayFactor: retryPolicy.delayFactor,
|
||||
randomizationFactor: retryPolicy.randomizationFactor,
|
||||
maxDelay: retryPolicy.maxDelay,
|
||||
maxAttempts: retryPolicy.maxRetryAttempts,
|
||||
retryIf: (error, attempt) {
|
||||
if (error is! StreamChatError) return false;
|
||||
return retryPolicy.shouldRetry(client, attempt, error);
|
||||
},
|
||||
);
|
||||
} catch (error) {
|
||||
logger?.severe('Error while retrying message ${message.id}', error);
|
||||
// If we are unable to successfully retry the message, update the state
|
||||
// with the failed state.
|
||||
channel.state?.updateMessage(message);
|
||||
} finally {
|
||||
// remove the message from the queue after it's handled.
|
||||
_messageQueue.removeFirst();
|
||||
}
|
||||
}
|
||||
return false;
|
||||
|
||||
_isProcessing = 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?.updateMessage(message.copyWith(status: newStatus));
|
||||
}
|
||||
|
||||
Future<void> _retryMessage(Message message) async {
|
||||
if (message.status == MessageSendingStatus.failed_update ||
|
||||
message.status == MessageSendingStatus.updating) {
|
||||
await channel.updateMessage(message);
|
||||
} else if (message.status == MessageSendingStatus.failed ||
|
||||
message.status == MessageSendingStatus.sending) {
|
||||
await channel.sendMessage(message);
|
||||
} else if (message.status == MessageSendingStatus.failed_delete ||
|
||||
message.status == MessageSendingStatus.deleting) {
|
||||
await channel.deleteMessage(message);
|
||||
}
|
||||
Future<Object> _retryMessage(Message message) async {
|
||||
return message.state.maybeWhen(
|
||||
failed: (state, _) => state.when(
|
||||
sendingFailed: () => channel.sendMessage(message),
|
||||
updatingFailed: () => channel.updateMessage(message),
|
||||
deletingFailed: (hard) => channel.deleteMessage(message, hard: hard),
|
||||
),
|
||||
orElse: () => throw StateError('Message state is not failed'),
|
||||
);
|
||||
}
|
||||
|
||||
/// Whether our [_messageQueue] has messages or not.
|
||||
bool get hasMessages => _messageQueue.isNotEmpty;
|
||||
|
||||
/// Returns true if the queue contains the given [message].
|
||||
bool contains(Message message) => _messageQueue.containsMessage(message);
|
||||
|
||||
/// Call this method to dispose this object.
|
||||
void dispose() {
|
||||
_messageQueue.clear();
|
||||
@@ -188,33 +140,25 @@ class RetryQueue {
|
||||
final date1 = _getMessageDate(m1);
|
||||
final date2 = _getMessageDate(m2);
|
||||
|
||||
if (date1 == null || date2 == null) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (date1 == null && date2 == null) return 0;
|
||||
if (date1 == null) return -1;
|
||||
if (date2 == null) return 1;
|
||||
return date1.compareTo(date2);
|
||||
}
|
||||
|
||||
static DateTime? _getMessageDate(Message m1) {
|
||||
switch (m1.status) {
|
||||
case MessageSendingStatus.failed_delete:
|
||||
case MessageSendingStatus.deleting:
|
||||
return m1.deletedAt;
|
||||
|
||||
case MessageSendingStatus.failed:
|
||||
case MessageSendingStatus.sending:
|
||||
return m1.createdAt;
|
||||
|
||||
case MessageSendingStatus.failed_update:
|
||||
case MessageSendingStatus.updating:
|
||||
return m1.updatedAt;
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
static DateTime? _getMessageDate(Message message) {
|
||||
return message.state.maybeWhen(
|
||||
failed: (state, _) => state.when(
|
||||
sendingFailed: () => message.createdAt,
|
||||
updatingFailed: () => message.updatedAt,
|
||||
deletingFailed: (_) => message.deletedAt,
|
||||
),
|
||||
orElse: () => null,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
extension _MessageHeapPriorityQueue on HeapPriorityQueue<Message> {
|
||||
extension on HeapPriorityQueue<Message> {
|
||||
void removeMessage(Message message) {
|
||||
final list = toUnorderedList();
|
||||
final index = list.indexWhere((it) => it.id == message.id);
|
||||
@@ -229,11 +173,4 @@ extension _MessageHeapPriorityQueue on HeapPriorityQueue<Message> {
|
||||
if (index == -1) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
bool containsAllMessage(List<Message> messages) {
|
||||
if (isEmpty) return false;
|
||||
final list = toUnorderedList();
|
||||
final messageIds = messages.map((it) => it.id);
|
||||
return list.every((it) => messageIds.contains(it.id));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user