change file structure, replace rate_limit with the pub dependency
Signed-off-by: Sahil Kumar <[email protected]>
This commit is contained in:
Sahil Kumar
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,31 @@
|
||||
import 'package:stream_chat/src/client/client.dart';
|
||||
import 'package:stream_chat/src/core/error/error.dart';
|
||||
|
||||
/// The retry options
|
||||
class RetryPolicy {
|
||||
/// Instantiate a new RetryPolicy
|
||||
RetryPolicy({
|
||||
required this.shouldRetry,
|
||||
required this.retryTimeout,
|
||||
this.maxRetryAttempts = 6,
|
||||
});
|
||||
|
||||
/// 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,
|
||||
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,
|
||||
StreamChatError? error,
|
||||
) retryTimeout;
|
||||
}
|
||||
@@ -0,0 +1,238 @@
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:collection/collection.dart';
|
||||
import 'package:logging/logging.dart';
|
||||
import 'package:rxdart/rxdart.dart';
|
||||
import 'package:stream_chat/src/client/channel.dart';
|
||||
import 'package:stream_chat/src/client/retry_policy.dart';
|
||||
import 'package:stream_chat/src/core/error/error.dart';
|
||||
import 'package:stream_chat/src/event_type.dart';
|
||||
import 'package:stream_chat/src/core/models/message.dart';
|
||||
import 'package:stream_chat/stream_chat.dart';
|
||||
|
||||
/// The retry queue associated to a channel
|
||||
class RetryQueue {
|
||||
/// Instantiate a new RetryQueue object
|
||||
RetryQueue({
|
||||
required this.channel,
|
||||
this.logger,
|
||||
}) : 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;
|
||||
|
||||
late final RetryPolicy _retryPolicy;
|
||||
|
||||
final _compositeSubscription = CompositeSubscription();
|
||||
|
||||
final _messageQueue = HeapPriorityQueue(_byDate);
|
||||
bool _isRetrying = false;
|
||||
|
||||
void _listenConnectionRecovered() {
|
||||
client.on(EventType.connectionRecovered).listen((event) {
|
||||
if (event.online == true) {
|
||||
_startRetrying();
|
||||
}
|
||||
}).addTo(_compositeSubscription);
|
||||
}
|
||||
|
||||
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<Message> messages) {
|
||||
if (messages.isEmpty) return;
|
||||
if (_messageQueue.containsAllMessage(messages)) return;
|
||||
|
||||
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),
|
||||
));
|
||||
_startRetrying();
|
||||
}
|
||||
|
||||
Future<void> _startRetrying() async {
|
||||
if (_isRetrying) return;
|
||||
_isRetrying = true;
|
||||
|
||||
logger?.info('Started retrying failed messages');
|
||||
while (_messageQueue.isNotEmpty) {
|
||||
logger?.info('${_messageQueue.length} messages remaining in the queue');
|
||||
final message = _messageQueue.first;
|
||||
await _runAndRetry(message);
|
||||
}
|
||||
_isRetrying = false;
|
||||
}
|
||||
|
||||
Future<void> _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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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<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);
|
||||
}
|
||||
}
|
||||
|
||||
/// Call this method to dispose this object
|
||||
void dispose() {
|
||||
_messageQueue.clear();
|
||||
_compositeSubscription.dispose();
|
||||
}
|
||||
|
||||
static int _byDate(Message m1, Message m2) {
|
||||
final date1 = _getMessageDate(m1);
|
||||
final date2 = _getMessageDate(m2);
|
||||
|
||||
if (date1 == null || date2 == null) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
extension _MessageHeapPriorityQueue on HeapPriorityQueue<Message> {
|
||||
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<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