feat: Rework how the ReconnectionPolicy system works
This commit is contained in:
parent
1cc266c675
commit
bff4a6f707
@ -1,6 +1,7 @@
|
|||||||
library moxxmpp;
|
library moxxmpp;
|
||||||
|
|
||||||
export 'package:moxxmpp/src/connection.dart';
|
export 'package:moxxmpp/src/connection.dart';
|
||||||
|
export 'package:moxxmpp/src/connectivity.dart';
|
||||||
export 'package:moxxmpp/src/errors.dart';
|
export 'package:moxxmpp/src/errors.dart';
|
||||||
export 'package:moxxmpp/src/events.dart';
|
export 'package:moxxmpp/src/events.dart';
|
||||||
export 'package:moxxmpp/src/iq.dart';
|
export 'package:moxxmpp/src/iq.dart';
|
||||||
|
@ -4,6 +4,7 @@ import 'package:meta/meta.dart';
|
|||||||
import 'package:moxlib/moxlib.dart';
|
import 'package:moxlib/moxlib.dart';
|
||||||
import 'package:moxxmpp/src/awaiter.dart';
|
import 'package:moxxmpp/src/awaiter.dart';
|
||||||
import 'package:moxxmpp/src/buffer.dart';
|
import 'package:moxxmpp/src/buffer.dart';
|
||||||
|
import 'package:moxxmpp/src/connectivity.dart';
|
||||||
import 'package:moxxmpp/src/errors.dart';
|
import 'package:moxxmpp/src/errors.dart';
|
||||||
import 'package:moxxmpp/src/events.dart';
|
import 'package:moxxmpp/src/events.dart';
|
||||||
import 'package:moxxmpp/src/iq.dart';
|
import 'package:moxxmpp/src/iq.dart';
|
||||||
@ -94,12 +95,14 @@ class XmppConnectionResult {
|
|||||||
class XmppConnection {
|
class XmppConnection {
|
||||||
XmppConnection(
|
XmppConnection(
|
||||||
ReconnectionPolicy reconnectionPolicy,
|
ReconnectionPolicy reconnectionPolicy,
|
||||||
|
ConnectivityManager connectivityManager,
|
||||||
this._socket,
|
this._socket,
|
||||||
{
|
{
|
||||||
this.connectionPingDuration = const Duration(minutes: 3),
|
this.connectionPingDuration = const Duration(minutes: 3),
|
||||||
this.connectingTimeout = const Duration(minutes: 2),
|
this.connectingTimeout = const Duration(minutes: 2),
|
||||||
}
|
}
|
||||||
) : _reconnectionPolicy = reconnectionPolicy {
|
) : _reconnectionPolicy = reconnectionPolicy,
|
||||||
|
_connectivityManager = connectivityManager {
|
||||||
// Allow the reconnection policy to perform reconnections by itself
|
// Allow the reconnection policy to perform reconnections by itself
|
||||||
_reconnectionPolicy.register(
|
_reconnectionPolicy.register(
|
||||||
_attemptReconnection,
|
_attemptReconnection,
|
||||||
@ -122,13 +125,16 @@ class XmppConnection {
|
|||||||
/// The data stream of the socket
|
/// The data stream of the socket
|
||||||
late final Stream<String> _socketStream;
|
late final Stream<String> _socketStream;
|
||||||
|
|
||||||
|
|
||||||
/// Connection settings
|
/// Connection settings
|
||||||
late ConnectionSettings _connectionSettings;
|
late ConnectionSettings _connectionSettings;
|
||||||
|
|
||||||
/// A policy on how to reconnect
|
/// A policy on how to reconnect
|
||||||
final ReconnectionPolicy _reconnectionPolicy;
|
final ReconnectionPolicy _reconnectionPolicy;
|
||||||
|
|
||||||
|
/// The class responsible for preventing errors on initial connection due
|
||||||
|
/// to no network.
|
||||||
|
final ConnectivityManager _connectivityManager;
|
||||||
|
|
||||||
/// A helper for handling await semantics with stanzas
|
/// A helper for handling await semantics with stanzas
|
||||||
final StanzaAwaiter _stanzaAwaiter = StanzaAwaiter();
|
final StanzaAwaiter _stanzaAwaiter = StanzaAwaiter();
|
||||||
|
|
||||||
@ -378,7 +384,7 @@ class XmppConnection {
|
|||||||
// Connect again
|
// Connect again
|
||||||
// ignore: cascade_invocations
|
// ignore: cascade_invocations
|
||||||
_log.finest('Calling connect() from _attemptReconnection');
|
_log.finest('Calling connect() from _attemptReconnection');
|
||||||
await connect();
|
await connect(waitForConnection: true);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Called when a stream ending error has occurred
|
/// Called when a stream ending error has occurred
|
||||||
@ -401,7 +407,11 @@ class XmppConnection {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (await _connectivityManager.hasConnection()) {
|
||||||
await _setConnectionState(XmppConnectionState.error);
|
await _setConnectionState(XmppConnectionState.error);
|
||||||
|
} else {
|
||||||
|
await _setConnectionState(XmppConnectionState.notConnected);
|
||||||
|
}
|
||||||
await _reconnectionPolicy.onFailure();
|
await _reconnectionPolicy.onFailure();
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -832,7 +842,6 @@ class XmppConnection {
|
|||||||
if (_isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
|
if (_isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
|
||||||
_log.finest('Negotiations done!');
|
_log.finest('Negotiations done!');
|
||||||
_updateRoutingState(RoutingState.handleStanzas);
|
_updateRoutingState(RoutingState.handleStanzas);
|
||||||
await _reconnectionPolicy.onSuccess();
|
|
||||||
await _resetIsConnectionRunning();
|
await _resetIsConnectionRunning();
|
||||||
await _onNegotiationsDone();
|
await _onNegotiationsDone();
|
||||||
} else {
|
} else {
|
||||||
@ -857,7 +866,6 @@ class XmppConnection {
|
|||||||
_log.finest('Negotiations done!');
|
_log.finest('Negotiations done!');
|
||||||
|
|
||||||
_updateRoutingState(RoutingState.handleStanzas);
|
_updateRoutingState(RoutingState.handleStanzas);
|
||||||
await _reconnectionPolicy.onSuccess();
|
|
||||||
await _resetIsConnectionRunning();
|
await _resetIsConnectionRunning();
|
||||||
await _onNegotiationsDone();
|
await _onNegotiationsDone();
|
||||||
} else {
|
} else {
|
||||||
@ -875,7 +883,6 @@ class XmppConnection {
|
|||||||
_log.finest('Negotiator wants to skip the remaining negotiation... Negotiations (assumed) done!');
|
_log.finest('Negotiator wants to skip the remaining negotiation... Negotiations (assumed) done!');
|
||||||
|
|
||||||
_updateRoutingState(RoutingState.handleStanzas);
|
_updateRoutingState(RoutingState.handleStanzas);
|
||||||
await _reconnectionPolicy.onSuccess();
|
|
||||||
await _resetIsConnectionRunning();
|
await _resetIsConnectionRunning();
|
||||||
await _onNegotiationsDone();
|
await _onNegotiationsDone();
|
||||||
break;
|
break;
|
||||||
@ -987,7 +994,7 @@ class XmppConnection {
|
|||||||
}
|
}
|
||||||
|
|
||||||
Future<void> _disconnect({required XmppConnectionState state, bool triggeredByUser = true}) async {
|
Future<void> _disconnect({required XmppConnectionState state, bool triggeredByUser = true}) async {
|
||||||
_reconnectionPolicy.setShouldReconnect(false);
|
await _reconnectionPolicy.setShouldReconnect(false);
|
||||||
|
|
||||||
if (triggeredByUser) {
|
if (triggeredByUser) {
|
||||||
getPresenceManager().sendUnavailablePresence();
|
getPresenceManager().sendUnavailablePresence();
|
||||||
@ -1018,17 +1025,21 @@ class XmppConnection {
|
|||||||
|
|
||||||
/// Like [connect] but the Future resolves when the resource binding is either done or
|
/// Like [connect] but the Future resolves when the resource binding is either done or
|
||||||
/// SASL has failed.
|
/// SASL has failed.
|
||||||
Future<XmppConnectionResult> connectAwaitable({ String? lastResource }) async {
|
Future<XmppConnectionResult> connectAwaitable({ String? lastResource, bool waitForConnection = false }) async {
|
||||||
_runPreConnectionAssertions();
|
_runPreConnectionAssertions();
|
||||||
await _resetIsConnectionRunning();
|
await _resetIsConnectionRunning();
|
||||||
_connectionCompleter = Completer();
|
_connectionCompleter = Completer();
|
||||||
_log.finest('Calling connect() from connectAwaitable');
|
_log.finest('Calling connect() from connectAwaitable');
|
||||||
await connect(lastResource: lastResource);
|
await connect(
|
||||||
|
lastResource: lastResource,
|
||||||
|
waitForConnection: waitForConnection,
|
||||||
|
shouldReconnect: false,
|
||||||
|
);
|
||||||
return _connectionCompleter!.future;
|
return _connectionCompleter!.future;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Start the connection process using the provided connection settings.
|
/// Start the connection process using the provided connection settings.
|
||||||
Future<void> connect({ String? lastResource }) async {
|
Future<void> connect({ String? lastResource, bool waitForConnection = false, bool shouldReconnect = true }) async {
|
||||||
if (_connectionState != XmppConnectionState.notConnected && _connectionState != XmppConnectionState.error) {
|
if (_connectionState != XmppConnectionState.notConnected && _connectionState != XmppConnectionState.error) {
|
||||||
_log.fine('Cancelling this connection attempt as one appears to be already running.');
|
_log.fine('Cancelling this connection attempt as one appears to be already running.');
|
||||||
return;
|
return;
|
||||||
@ -1036,15 +1047,25 @@ class XmppConnection {
|
|||||||
|
|
||||||
_runPreConnectionAssertions();
|
_runPreConnectionAssertions();
|
||||||
await _resetIsConnectionRunning();
|
await _resetIsConnectionRunning();
|
||||||
_reconnectionPolicy.setShouldReconnect(true);
|
|
||||||
|
|
||||||
if (lastResource != null) {
|
if (lastResource != null) {
|
||||||
setResource(lastResource);
|
setResource(lastResource);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (shouldReconnect) {
|
||||||
|
await _reconnectionPolicy.setShouldReconnect(true);
|
||||||
|
}
|
||||||
|
|
||||||
await _reconnectionPolicy.reset();
|
await _reconnectionPolicy.reset();
|
||||||
await _sendEvent(ConnectingEvent());
|
await _sendEvent(ConnectingEvent());
|
||||||
|
|
||||||
|
// If requested, wait until we have a network connection
|
||||||
|
if (waitForConnection) {
|
||||||
|
_log.info('Waiting for okay from connectivityManager');
|
||||||
|
await _connectivityManager.waitForConnection();
|
||||||
|
_log.info('Got okay from connectivityManager');
|
||||||
|
}
|
||||||
|
|
||||||
final smManager = getStreamManagementManager();
|
final smManager = getStreamManagementManager();
|
||||||
String? host;
|
String? host;
|
||||||
int? port;
|
int? port;
|
||||||
|
18
packages/moxxmpp/lib/src/connectivity.dart
Normal file
18
packages/moxxmpp/lib/src/connectivity.dart
Normal file
@ -0,0 +1,18 @@
|
|||||||
|
/// This manager class is responsible to tell the moxxmpp XmppConnection
|
||||||
|
/// when a connection can be established or not, regarding the network availability.
|
||||||
|
abstract class ConnectivityManager {
|
||||||
|
/// Returns true if a network connection is available. If not, returns false.
|
||||||
|
Future<bool> hasConnection();
|
||||||
|
|
||||||
|
/// Returns a future that resolves once we have a network connection.
|
||||||
|
Future<void> waitForConnection();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An implementation of [ConnectivityManager] that is always connected.
|
||||||
|
class AlwaysConnectedConnectivityManager extends ConnectivityManager {
|
||||||
|
@override
|
||||||
|
Future<bool> hasConnection() async => true;
|
||||||
|
|
||||||
|
@override
|
||||||
|
Future<void> waitForConnection() async {}
|
||||||
|
}
|
@ -2,6 +2,7 @@ import 'dart:async';
|
|||||||
import 'dart:math';
|
import 'dart:math';
|
||||||
import 'package:logging/logging.dart';
|
import 'package:logging/logging.dart';
|
||||||
import 'package:meta/meta.dart';
|
import 'package:meta/meta.dart';
|
||||||
|
import 'package:moxxmpp/src/util/queue.dart';
|
||||||
import 'package:synchronized/synchronized.dart';
|
import 'package:synchronized/synchronized.dart';
|
||||||
|
|
||||||
/// A callback function to be called when the connection to the server has been lost.
|
/// A callback function to be called when the connection to the server has been lost.
|
||||||
@ -24,10 +25,16 @@ abstract class ReconnectionPolicy {
|
|||||||
bool _shouldAttemptReconnection = false;
|
bool _shouldAttemptReconnection = false;
|
||||||
|
|
||||||
/// Indicate if a reconnection attempt is currently running.
|
/// Indicate if a reconnection attempt is currently running.
|
||||||
bool _isReconnecting = false;
|
@protected
|
||||||
|
bool isReconnecting = false;
|
||||||
|
|
||||||
/// And the corresponding lock
|
/// And the corresponding lock
|
||||||
final Lock _isReconnectingLock = Lock();
|
@protected
|
||||||
|
final Lock lock = Lock();
|
||||||
|
|
||||||
|
/// The lock for accessing [_shouldAttemptReconnection]
|
||||||
|
@protected
|
||||||
|
final Lock shouldReconnectLock = Lock();
|
||||||
|
|
||||||
/// Called by XmppConnection to register the policy.
|
/// Called by XmppConnection to register the policy.
|
||||||
void register(PerformReconnectFunction performReconnect, ConnectionLostCallback triggerConnectionLost) {
|
void register(PerformReconnectFunction performReconnect, ConnectionLostCallback triggerConnectionLost) {
|
||||||
@ -48,96 +55,121 @@ abstract class ReconnectionPolicy {
|
|||||||
/// Caled by the XmppConnection when the reconnection was successful.
|
/// Caled by the XmppConnection when the reconnection was successful.
|
||||||
Future<void> onSuccess();
|
Future<void> onSuccess();
|
||||||
|
|
||||||
bool get shouldReconnect => _shouldAttemptReconnection;
|
Future<bool> getShouldReconnect() async {
|
||||||
|
return shouldReconnectLock.synchronized(() => _shouldAttemptReconnection);
|
||||||
|
}
|
||||||
|
|
||||||
/// Set whether a reconnection attempt should be made.
|
/// Set whether a reconnection attempt should be made.
|
||||||
void setShouldReconnect(bool value) {
|
Future<void> setShouldReconnect(bool value) async {
|
||||||
_shouldAttemptReconnection = value;
|
return shouldReconnectLock.synchronized(() => _shouldAttemptReconnection = value);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns true if the manager is currently triggering a reconnection. If not, returns
|
/// Returns true if the manager is currently triggering a reconnection. If not, returns
|
||||||
/// false.
|
/// false.
|
||||||
Future<bool> isReconnectionRunning() async {
|
Future<bool> isReconnectionRunning() async {
|
||||||
return _isReconnectingLock.synchronized(() => _isReconnecting);
|
return lock.synchronized(() => isReconnecting);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Set the _isReconnecting state to [value].
|
/// Set the isReconnecting state to [value].
|
||||||
@protected
|
@protected
|
||||||
Future<void> setIsReconnecting(bool value) async {
|
Future<void> setIsReconnecting(bool value) async {
|
||||||
await _isReconnectingLock.synchronized(() async {
|
await lock.synchronized(() async {
|
||||||
_isReconnecting = value;
|
isReconnecting = value;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@protected
|
|
||||||
Future<bool> testAndSetIsReconnecting() async {
|
|
||||||
return _isReconnectingLock.synchronized(() {
|
|
||||||
if (_isReconnecting) {
|
|
||||||
return false;
|
|
||||||
} else {
|
|
||||||
_isReconnecting = true;
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A simple reconnection strategy: Make the reconnection delays exponentially longer
|
/// A simple reconnection strategy: Make the reconnection delays exponentially longer
|
||||||
/// for every failed attempt.
|
/// for every failed attempt.
|
||||||
/// NOTE: This ReconnectionPolicy may be broken
|
/// NOTE: This ReconnectionPolicy may be broken
|
||||||
class ExponentialBackoffReconnectionPolicy extends ReconnectionPolicy {
|
class RandomBackoffReconnectionPolicy extends ReconnectionPolicy {
|
||||||
ExponentialBackoffReconnectionPolicy(this._maxBackoffTime) : super();
|
RandomBackoffReconnectionPolicy(
|
||||||
|
this._minBackoffTime,
|
||||||
|
this._maxBackoffTime,
|
||||||
|
) : assert(_minBackoffTime < _maxBackoffTime, '_minBackoffTime must be smaller than _maxBackoffTime'),
|
||||||
|
super();
|
||||||
|
|
||||||
/// The maximum time in seconds that a backoff step should be.
|
/// The maximum time in seconds that a backoff should be.
|
||||||
final int _maxBackoffTime;
|
final int _maxBackoffTime;
|
||||||
|
|
||||||
/// Amount of consecutive failed reconnections.
|
/// The minimum time in seconds that a backoff should be.
|
||||||
int _counter = 0;
|
final int _minBackoffTime;
|
||||||
|
|
||||||
/// Backoff timer.
|
/// Backoff timer.
|
||||||
Timer? _timer;
|
Timer? _timer;
|
||||||
|
|
||||||
|
final Lock _timerLock = Lock();
|
||||||
|
|
||||||
/// Logger.
|
/// Logger.
|
||||||
final Logger _log = Logger('ExponentialBackoffReconnectionPolicy');
|
final Logger _log = Logger('RandomBackoffReconnectionPolicy');
|
||||||
|
|
||||||
|
/// Event queue
|
||||||
|
final AsyncQueue _eventQueue = AsyncQueue();
|
||||||
|
|
||||||
/// Called when the backoff expired
|
/// Called when the backoff expired
|
||||||
Future<void> _onTimerElapsed() async {
|
Future<void> _onTimerElapsed() async {
|
||||||
final isReconnecting = await isReconnectionRunning();
|
_log.fine('Timer elapsed. Waiting for lock');
|
||||||
if (shouldReconnect) {
|
await lock.synchronized(() async {
|
||||||
if (!isReconnecting) {
|
_log.fine('Lock aquired');
|
||||||
await setIsReconnecting(true);
|
if (!(await getShouldReconnect())) {
|
||||||
|
_log.fine('Backoff timer expired but getShouldReconnect() returned false');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (isReconnecting) {
|
||||||
|
_log.fine('Backoff timer expired but a reconnection is running, so doing nothing.');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_log.fine('Triggering reconnect');
|
||||||
|
isReconnecting = true;
|
||||||
await performReconnect!();
|
await performReconnect!();
|
||||||
} else {
|
});
|
||||||
// Should never happen.
|
|
||||||
_log.fine('Backoff timer expired but reconnection is running, so doing nothing.');
|
await _timerLock.synchronized(() {
|
||||||
}
|
_timer?.cancel();
|
||||||
|
_timer = null;
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Future<void> _reset() async {
|
||||||
|
_log.finest('Resetting internal state');
|
||||||
|
|
||||||
|
await _timerLock.synchronized(() {
|
||||||
|
_timer?.cancel();
|
||||||
|
_timer = null;
|
||||||
|
});
|
||||||
|
|
||||||
|
await setIsReconnecting(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
@override
|
@override
|
||||||
Future<void> reset() async {
|
Future<void> reset() async {
|
||||||
_log.finest('Resetting internal state');
|
// ignore: unnecessary_lambdas
|
||||||
_counter = 0;
|
await _eventQueue.addJob(() => _reset());
|
||||||
await setIsReconnecting(false);
|
|
||||||
|
|
||||||
if (_timer != null) {
|
|
||||||
_timer!.cancel();
|
|
||||||
_timer = null;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Future<void> _onFailure() async {
|
||||||
|
final shouldContinue = await _timerLock.synchronized(() {
|
||||||
|
return _timer == null;
|
||||||
|
});
|
||||||
|
if (!shouldContinue) {
|
||||||
|
_log.finest('_onFailure: Not backing off since _timer is already running');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
final seconds = Random().nextInt(_maxBackoffTime - _minBackoffTime) + _minBackoffTime;
|
||||||
|
_log.finest('Failure occured. Starting random backoff with ${seconds}s');
|
||||||
|
_timer?.cancel();
|
||||||
|
|
||||||
|
_timer = Timer(Duration(seconds: seconds), _onTimerElapsed);
|
||||||
}
|
}
|
||||||
|
|
||||||
@override
|
@override
|
||||||
Future<void> onFailure() async {
|
Future<void> onFailure() async {
|
||||||
_log.finest('Failure occured. Starting exponential backoff');
|
// ignore: unnecessary_lambdas
|
||||||
_counter++;
|
await _eventQueue.addJob(() => _onFailure());
|
||||||
|
|
||||||
if (_timer != null) {
|
|
||||||
_timer!.cancel();
|
|
||||||
}
|
|
||||||
|
|
||||||
// Wait at max 80 seconds.
|
|
||||||
final seconds = min(min(pow(2, _counter).toInt(), 80), _maxBackoffTime);
|
|
||||||
_timer = Timer(Duration(seconds: seconds), _onTimerElapsed);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@override
|
@override
|
||||||
|
56
packages/moxxmpp/lib/src/util/queue.dart
Normal file
56
packages/moxxmpp/lib/src/util/queue.dart
Normal file
@ -0,0 +1,56 @@
|
|||||||
|
import 'dart:async';
|
||||||
|
import 'dart:collection';
|
||||||
|
import 'package:meta/meta.dart';
|
||||||
|
import 'package:synchronized/synchronized.dart';
|
||||||
|
|
||||||
|
/// A job to be submitted to an [AsyncQueue].
|
||||||
|
typedef AsyncQueueJob = Future<void> Function();
|
||||||
|
|
||||||
|
/// A (hopefully) async-safe queue that attempts to force
|
||||||
|
/// in-order execution of its jobs.
|
||||||
|
class AsyncQueue {
|
||||||
|
/// The lock for accessing [AsyncQueue._lock] and [AsyncQueue._running].
|
||||||
|
final Lock _lock = Lock();
|
||||||
|
|
||||||
|
/// The actual job queue.
|
||||||
|
final Queue<AsyncQueueJob> _queue = Queue<AsyncQueueJob>();
|
||||||
|
|
||||||
|
/// Indicates whether we are currently executing a job.
|
||||||
|
bool _running = false;
|
||||||
|
|
||||||
|
@visibleForTesting
|
||||||
|
Queue<AsyncQueueJob> get queue => _queue;
|
||||||
|
|
||||||
|
@visibleForTesting
|
||||||
|
bool get isRunning => _running;
|
||||||
|
|
||||||
|
/// Adds a job [job] to the queue.
|
||||||
|
Future<void> addJob(AsyncQueueJob job) async {
|
||||||
|
await _lock.synchronized(() {
|
||||||
|
_queue.add(job);
|
||||||
|
|
||||||
|
if (!_running && _queue.isNotEmpty) {
|
||||||
|
_running = true;
|
||||||
|
unawaited(_popJob());
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Future<void> clear() async {
|
||||||
|
await _lock.synchronized(_queue.clear);
|
||||||
|
}
|
||||||
|
|
||||||
|
Future<void> _popJob() async {
|
||||||
|
final job = _queue.removeFirst();
|
||||||
|
final future = job();
|
||||||
|
await future;
|
||||||
|
|
||||||
|
await _lock.synchronized(() {
|
||||||
|
if (_queue.isNotEmpty) {
|
||||||
|
unawaited(_popJob());
|
||||||
|
} else {
|
||||||
|
_running = false;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
43
packages/moxxmpp/test/async_queue_test.dart
Normal file
43
packages/moxxmpp/test/async_queue_test.dart
Normal file
@ -0,0 +1,43 @@
|
|||||||
|
import 'package:moxxmpp/src/util/queue.dart';
|
||||||
|
import 'package:test/test.dart';
|
||||||
|
|
||||||
|
void main() {
|
||||||
|
test('Test the async queue', () async {
|
||||||
|
final queue = AsyncQueue();
|
||||||
|
int future1Finish = 0;
|
||||||
|
int future2Finish = 0;
|
||||||
|
int future3Finish = 0;
|
||||||
|
|
||||||
|
await queue.addJob(() => Future<void>.delayed(const Duration(seconds: 3), () => future1Finish = DateTime.now().millisecondsSinceEpoch));
|
||||||
|
await queue.addJob(() => Future<void>.delayed(const Duration(seconds: 3), () => future2Finish = DateTime.now().millisecondsSinceEpoch));
|
||||||
|
await queue.addJob(() => Future<void>.delayed(const Duration(seconds: 3), () => future3Finish = DateTime.now().millisecondsSinceEpoch));
|
||||||
|
|
||||||
|
await Future<void>.delayed(const Duration(seconds: 12));
|
||||||
|
|
||||||
|
// The three futures must be done
|
||||||
|
expect(future1Finish != 0, true);
|
||||||
|
expect(future2Finish != 0, true);
|
||||||
|
expect(future3Finish != 0, true);
|
||||||
|
|
||||||
|
// The end times of the futures must be ordered (on a timeline)
|
||||||
|
// |-- future1Finish -- future2Finish -- future3Finish --|
|
||||||
|
expect(
|
||||||
|
future1Finish < future2Finish && future1Finish < future3Finish,
|
||||||
|
true,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
future2Finish < future3Finish && future2Finish > future1Finish,
|
||||||
|
true,
|
||||||
|
);
|
||||||
|
expect(
|
||||||
|
future3Finish > future1Finish && future3Finish > future2Finish,
|
||||||
|
true,
|
||||||
|
);
|
||||||
|
|
||||||
|
// The queue must be empty at the end
|
||||||
|
expect(queue.queue.isEmpty, true);
|
||||||
|
|
||||||
|
// The queue must not be executing anything at the end
|
||||||
|
expect(queue.isRunning, false);
|
||||||
|
});
|
||||||
|
}
|
@ -54,8 +54,11 @@ void main() {
|
|||||||
],
|
],
|
||||||
);
|
);
|
||||||
|
|
||||||
final connection = XmppConnection(TestingReconnectionPolicy(), stubSocket)
|
final connection = XmppConnection(
|
||||||
..registerFeatureNegotiators([
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
stubSocket,
|
||||||
|
)..registerFeatureNegotiators([
|
||||||
StubNegotiator1(),
|
StubNegotiator1(),
|
||||||
StubNegotiator2(),
|
StubNegotiator2(),
|
||||||
])
|
])
|
||||||
|
@ -65,7 +65,11 @@ void main() {
|
|||||||
|
|
||||||
],
|
],
|
||||||
);
|
);
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), socket: fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -74,7 +78,7 @@ void main() {
|
|||||||
),);
|
),);
|
||||||
conn.registerManagers([
|
conn.registerManagers([
|
||||||
PresenceManager('http://moxxmpp.example'),
|
PresenceManager('http://moxxmpp.example'),
|
||||||
RosterManager(),
|
RosterManager(TestingRosterStateManager(null, [])),
|
||||||
DiscoManager(),
|
DiscoManager(),
|
||||||
PingManager(),
|
PingManager(),
|
||||||
]);
|
]);
|
@ -34,7 +34,7 @@ XmppManagerAttributes mkAttributes(void Function(Stanza) callback) {
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('hallo@example.server/uwu'),
|
getFullJID: () => JID.fromString('hallo@example.server/uwu'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
getNegotiatorById: getNegotiatorNullStub,
|
getNegotiatorById: getNegotiatorNullStub,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@ -233,7 +233,11 @@ void main() {
|
|||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -355,7 +359,11 @@ void main() {
|
|||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -510,7 +518,11 @@ void main() {
|
|||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -602,7 +614,11 @@ void main() {
|
|||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -694,7 +710,11 @@ void main() {
|
|||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
|
@ -22,7 +22,7 @@ void main() {
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('bob@xmpp.example/uwu'),
|
getFullJID: () => JID.fromString('bob@xmpp.example/uwu'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
getNegotiatorById: getNegotiatorNullStub,
|
getNegotiatorById: getNegotiatorNullStub,
|
||||||
);
|
);
|
||||||
final manager = CarbonsManager();
|
final manager = CarbonsManager();
|
||||||
|
@ -50,7 +50,7 @@ void main() {
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -79,7 +79,7 @@ void main() {
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
|
|
||||||
|
@ -25,7 +25,7 @@ Future<bool> testRosterManager(String bareJid, String resource, String stanzaStr
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('$bareJid/$resource'),
|
getFullJID: () => JID.fromString('$bareJid/$resource'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
),);
|
),);
|
||||||
|
|
||||||
final stanza = Stanza.fromXMLNode(XMLNode.fromString(stanzaString));
|
final stanza = Stanza.fromXMLNode(XMLNode.fromString(stanzaString));
|
||||||
@ -118,7 +118,10 @@ void main() {
|
|||||||
],
|
],
|
||||||
);
|
);
|
||||||
// TODO: This test is broken since we query the server and enable carbons
|
// TODO: This test is broken since we query the server and enable carbons
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -172,7 +175,11 @@ void main() {
|
|||||||
],
|
],
|
||||||
);
|
);
|
||||||
var receivedEvent = false;
|
var receivedEvent = false;
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -226,7 +233,11 @@ void main() {
|
|||||||
],
|
],
|
||||||
);
|
);
|
||||||
var receivedEvent = false;
|
var receivedEvent = false;
|
||||||
final XmppConnection conn = XmppConnection(TestingReconnectionPolicy(), fakeSocket);
|
final XmppConnection conn = XmppConnection(
|
||||||
|
TestingReconnectionPolicy(),
|
||||||
|
AlwaysConnectedConnectivityManager(),
|
||||||
|
fakeSocket,
|
||||||
|
);
|
||||||
conn.setConnectionSettings(ConnectionSettings(
|
conn.setConnectionSettings(ConnectionSettings(
|
||||||
jid: JID.fromString('polynomdivision@test.server'),
|
jid: JID.fromString('polynomdivision@test.server'),
|
||||||
password: 'aaaa',
|
password: 'aaaa',
|
||||||
@ -326,7 +337,7 @@ void main() {
|
|||||||
isFeatureSupported: (_) => false,
|
isFeatureSupported: (_) => false,
|
||||||
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
getFullJID: () => JID.fromString('some.user@example.server/aaaaa'),
|
||||||
getSocket: () => StubTCPSocket(play: []),
|
getSocket: () => StubTCPSocket(play: []),
|
||||||
getConnection: () => XmppConnection(TestingReconnectionPolicy(), StubTCPSocket(play: [])),
|
getConnection: () => XmppConnection(TestingReconnectionPolicy(), AlwaysConnectedConnectivityManager(), StubTCPSocket(play: [])),
|
||||||
),);
|
),);
|
||||||
|
|
||||||
// NOTE: Based on https://gultsch.de/gajim_roster_push_and_message_interception.html
|
// NOTE: Based on https://gultsch.de/gajim_roster_push_and_message_interception.html
|
||||||
|
Loading…
Reference in New Issue
Block a user