Compare commits

..

3 Commits

Author SHA1 Message Date
300a52f9fe refactor: Replace MayFail by Result 2022-11-19 22:26:02 +01:00
2e3472d88f feat: Rework how the negotiator system works
We can now return what exactly made a connection attempt fail.
2022-11-19 21:48:28 +01:00
6b106fe365 test: Add integration test for ExponentialBackoffReconnectionPolicy 2022-11-16 20:48:18 +01:00
21 changed files with 239 additions and 224 deletions

View File

@ -10,7 +10,7 @@ void main() {
});
final log = Logger('FailureReconnectionTest');
test('Failing an awaited connection', () async {
test('Failing an awaited connection with TestingSleepReconnectionPolicy', () async {
var errors = 0;
final connection = XmppConnection(
TestingSleepReconnectionPolicy(10),
@ -52,4 +52,47 @@ void main() {
await Future.delayed(const Duration(seconds: 20));
expect(errors, 1);
}, timeout: Timeout.factor(2));
test('Failing an awaited connection with ExponentialBackoffReconnectionPolicy', () async {
var errors = 0;
final connection = XmppConnection(
ExponentialBackoffReconnectionPolicy(1),
TCPSocketWrapper(false),
);
connection.registerFeatureNegotiators([
StartTlsNegotiator(),
]);
connection.registerManagers([
DiscoManager(),
RosterManager(),
PingManager(),
MessageManager(),
PresenceManager('http://moxxmpp.example'),
]);
connection.asBroadcastStream().listen((event) {
if (event is ConnectionStateChangedEvent) {
if (event.state == XmppConnectionState.error) {
errors++;
}
}
});
connection.setConnectionSettings(
ConnectionSettings(
jid: JID.fromString('testuser@no-sasl.badxmpp.eu'),
password: 'abc123',
useDirectTLS: true,
allowPlainAuth: true,
),
);
final result = await connection.connectAwaitable();
log.info('Connection failed as expected');
expect(result.success, false);
expect(errors, 1);
log.info('Waiting 20 seconds for unexpected reconnections');
await Future.delayed(const Duration(seconds: 20));
expect(errors, 1);
}, timeout: Timeout.factor(2));
}

View File

@ -25,12 +25,12 @@ export 'package:moxxmpp/src/presence.dart';
export 'package:moxxmpp/src/reconnect.dart';
export 'package:moxxmpp/src/rfcs/rfc_2782.dart';
export 'package:moxxmpp/src/rfcs/rfc_4790.dart';
export 'package:moxxmpp/src/roster.dart';
export 'package:moxxmpp/src/roster/errors.dart';
export 'package:moxxmpp/src/roster/roster.dart';
export 'package:moxxmpp/src/settings.dart';
export 'package:moxxmpp/src/socket.dart';
export 'package:moxxmpp/src/stanza.dart';
export 'package:moxxmpp/src/stringxml.dart';
export 'package:moxxmpp/src/types/error.dart';
export 'package:moxxmpp/src/types/result.dart';
export 'package:moxxmpp/src/xeps/staging/extensible_file_thumbnails.dart';
export 'package:moxxmpp/src/xeps/staging/file_upload_notification.dart';
@ -61,7 +61,8 @@ export 'package:moxxmpp/src/xeps/xep_0333.dart';
export 'package:moxxmpp/src/xeps/xep_0334.dart';
export 'package:moxxmpp/src/xeps/xep_0352.dart';
export 'package:moxxmpp/src/xeps/xep_0359.dart';
export 'package:moxxmpp/src/xeps/xep_0363.dart';
export 'package:moxxmpp/src/xeps/xep_0363/errors.dart';
export 'package:moxxmpp/src/xeps/xep_0363/xep_0363.dart';
export 'package:moxxmpp/src/xeps/xep_0380.dart';
export 'package:moxxmpp/src/xeps/xep_0384/crypto.dart';
export 'package:moxxmpp/src/xeps/xep_0384/errors.dart';

View File

@ -2,6 +2,7 @@ import 'dart:async';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:moxxmpp/src/buffer.dart';
import 'package:moxxmpp/src/errors.dart';
import 'package:moxxmpp/src/events.dart';
import 'package:moxxmpp/src/iq.dart';
import 'package:moxxmpp/src/managers/attributes.dart';
@ -14,7 +15,7 @@ import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/presence.dart';
import 'package:moxxmpp/src/reconnect.dart';
import 'package:moxxmpp/src/roster.dart';
import 'package:moxxmpp/src/roster/roster.dart';
import 'package:moxxmpp/src/routing.dart';
import 'package:moxxmpp/src/settings.dart';
import 'package:moxxmpp/src/socket.dart';
@ -61,14 +62,14 @@ class XmppConnectionResult {
const XmppConnectionResult(
this.success,
{
this.reason,
this.error,
}
);
final bool success;
// NOTE: [reason] is not human-readable, but the type of SASL error.
// See sasl/errors.dart
final String? reason;
// If a connection attempt fails, i.e. success is false, then this indicates the
// reason the connection failed.
final XmppError? error;
}
class XmppConnection {
@ -345,12 +346,8 @@ class XmppConnection {
}
/// Called when a stream ending error has occurred
Future<void> handleError(Object? error) async {
if (error != null) {
_log.severe('handleError: $error');
} else {
_log.severe('handleError: Called with null');
}
Future<void> handleError(XmppError error) async {
_log.severe('handleError called with ${error.toString()}');
// Whenever we encounter an error that would trigger a reconnection attempt while
// the connection result is being awaited, don't attempt a reconnection but instead
@ -358,7 +355,12 @@ class XmppConnection {
if (_connectionCompleter != null) {
_log.info('Not triggering reconnection since connection result is being awaited');
await _disconnect(triggeredByUser: false, state: XmppConnectionState.error);
_connectionCompleter?.complete(const XmppConnectionResult(false));
_connectionCompleter?.complete(
XmppConnectionResult(
false,
error: error,
),
);
_connectionCompleter = null;
return;
}
@ -370,7 +372,7 @@ class XmppConnection {
/// Called whenever the socket creates an event
Future<void> _handleSocketEvent(XmppSocketEvent event) async {
if (event is XmppSocketErrorEvent) {
await handleError(event.error);
await handleError(SocketError(event));
} else if (event is XmppSocketClosureEvent) {
if (_socketClosureTriggersReconnect) {
_log.fine('Received XmppSocketClosureEvent. Reconnecting...');
@ -525,7 +527,7 @@ class XmppConnection {
/// Called when we timeout during connecting
Future<void> _onConnectingTimeout() async {
_log.severe('Connection stuck in "connecting". Causing a reconnection...');
await handleError('Connecting timeout');
await handleError(TimeoutError());
}
void _destroyConnectingTimer() {
@ -750,19 +752,33 @@ class XmppConnection {
await getPresenceManager().sendInitialPresence();
}
/// To be called after _currentNegotiator!.negotiate(..) has been called. Checks the
/// state of the negotiator and picks the next negotiatior, ends negotiation or
/// waits, depending on what the negotiator did.
Future<void> _checkCurrentNegotiator() async {
if (_currentNegotiator!.state == NegotiatorState.done) {
_log.finest('Negotiator ${_currentNegotiator!.id} done');
Future<void> _executeCurrentNegotiator(XMLNode nonza) async {
// If we don't have a negotiator get one
_currentNegotiator ??= getNextNegotiator(_streamFeatures);
if (_currentNegotiator == null && _isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
_log.finest('Negotiations done!');
_updateRoutingState(RoutingState.handleStanzas);
await _onNegotiationsDone();
return;
}
final result = await _currentNegotiator!.negotiate(nonza);
if (result.isType<NegotiatorError>()) {
_log.severe('Negotiator returned an error');
await handleError(result.get<NegotiatorError>());
return;
}
final state = result.get<NegotiatorState>();
_currentNegotiator!.state = state;
switch (state) {
case NegotiatorState.ready: return;
case NegotiatorState.done:
if (_currentNegotiator!.sendStreamHeaderWhenDone) {
_currentNegotiator = null;
_streamFeatures.clear();
_sendStreamHeader();
} else {
// Track what features we still have
_streamFeatures
.removeWhere((node) {
return node.attributes['xmlns'] == _currentNegotiator!.negotiatingXmlns;
@ -772,7 +788,6 @@ class XmppConnection {
if (_isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
_log.finest('Negotiations done!');
_updateRoutingState(RoutingState.handleStanzas);
await _onNegotiationsDone();
} else {
_currentNegotiator = getNextNegotiator(_streamFeatures);
@ -782,15 +797,16 @@ class XmppConnection {
tag: 'stream:features',
children: _streamFeatures,
);
await _currentNegotiator!.negotiate(fakeStanza);
await _checkCurrentNegotiator();
await _executeCurrentNegotiator(fakeStanza);
}
}
} else if (_currentNegotiator!.state == NegotiatorState.retryLater) {
break;
case NegotiatorState.retryLater:
_log.finest('Negotiator wants to continue later. Picking new one...');
_currentNegotiator!.state = NegotiatorState.ready;
if (_isMandatoryNegotiationDone(_streamFeatures) && !_isNegotiationPossible(_streamFeatures)) {
_log.finest('Negotiations done!');
@ -804,24 +820,18 @@ class XmppConnection {
tag: 'stream:features',
children: _streamFeatures,
);
await _currentNegotiator!.negotiate(fakeStanza);
await _checkCurrentNegotiator();
await _executeCurrentNegotiator(fakeStanza);
}
} else if (_currentNegotiator!.state == NegotiatorState.skipRest) {
break;
case NegotiatorState.skipRest:
_log.finest('Negotiator wants to skip the remaining negotiation... Negotiations (assumed) done!');
_updateRoutingState(RoutingState.handleStanzas);
await _onNegotiationsDone();
} else if (_currentNegotiator!.state == NegotiatorState.error) {
_log.severe('Negotiator returned an error');
await handleError(null);
break;
}
}
void _closeSocket() {
_socket.close();
}
/// Called whenever we receive data that has been parsed as XML.
Future<void> handleXmlStream(XMLNode node) async {
// Check if we received a stream error
@ -829,7 +839,7 @@ class XmppConnection {
_log
..finest('<== ${node.toXml()}')
..severe('Received a stream error! Attempting reconnection');
await handleError('Stream error');
await handleError(StreamError());
return;
}
@ -849,53 +859,14 @@ class XmppConnection {
return;
}
if (_currentNegotiator != null) {
// If we already have a negotiator, just let it do its thing
_log.finest('Negotiator currently active...');
await _currentNegotiator!.negotiate(node);
await _checkCurrentNegotiator();
} else {
if (node.tag == 'stream:features') {
// Store the received stream features
_streamFeatures
..clear()
..addAll(node.children);
// We need to pick a new one
if (_isMandatoryNegotiationDone(node.children)) {
// Mandatory features are done but can we still negotiate more?
if (_isNegotiationPossible(node.children)) {// We can still negotiate features, so do that.
_log.finest('All required stream features done! Continuing negotiation');
_currentNegotiator = getNextNegotiator(node.children);
_log.finest('Chose $_currentNegotiator as next negotiator');
await _currentNegotiator!.negotiate(node);
await _checkCurrentNegotiator();
} else {
_updateRoutingState(RoutingState.handleStanzas);
}
} else {
// There still are mandatory features
if (!_isNegotiationPossible(node.children)) {
_log.severe('Mandatory negotiations not done but continuation not possible');
_updateRoutingState(RoutingState.error);
await _setConnectionState(XmppConnectionState.error);
// Resolve the connection completion future
_connectionCompleter?.complete(
const XmppConnectionResult(
false,
reason: 'Could not complete connection negotiations',
),
);
_connectionCompleter = null;
return;
}
_currentNegotiator = getNextNegotiator(node.children);
_log.finest('Chose $_currentNegotiator as next negotiator');
await _currentNegotiator!.negotiate(node);
await _checkCurrentNegotiator();
}
..clear()
..addAll(node.children);
}
await _executeCurrentNegotiator(node);
});
break;
case RoutingState.handleStanzas:
@ -927,20 +898,6 @@ class XmppConnection {
} else if (event is AuthenticationSuccessEvent) {
_log.finest('Received AuthenticationSuccessEvent. Setting _isAuthenticated to true');
_isAuthenticated = true;
} else if (event is AuthenticationFailedEvent) {
_log.finest('Failed authentication');
_updateRoutingState(RoutingState.error);
await _setConnectionState(XmppConnectionState.error);
// Resolve the connection completion future
_connectionCompleter?.complete(
XmppConnectionResult(
false,
reason: 'Authentication failed: ${event.saslError}',
),
);
_connectionCompleter = null;
_closeSocket();
}
for (final manager in _xmppManagers.values) {
@ -1053,7 +1010,7 @@ class XmppConnection {
port: port,
);
if (!result) {
await handleError(null);
await handleError(NoConnectionError());
} else {
await _reconnectionPolicy.onSuccess();
_log.fine('Preparing the internal state for a connection attempt');

View File

@ -0,0 +1,20 @@
import 'package:moxxmpp/src/socket.dart';
/// An internal error class
abstract class XmppError {}
/// Returned if we could not establish a TCP connection
/// to the server.
class NoConnectionError extends XmppError {}
/// Returned if a socket error occured
class SocketError extends XmppError {
SocketError(this.event);
final XmppSocketErrorEvent event;
}
/// Returned if we time out
class TimeoutError extends XmppError {}
/// Returned if we received a stream error
class StreamError extends XmppError {}

View File

@ -1,10 +1,12 @@
import 'package:moxlib/moxlib.dart';
import 'package:moxxmpp/src/errors.dart';
import 'package:moxxmpp/src/events.dart';
import 'package:moxxmpp/src/jid.dart';
import 'package:moxxmpp/src/managers/base.dart';
import 'package:moxxmpp/src/settings.dart';
import 'package:moxxmpp/src/socket.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
/// The state a negotiator is currently in
enum NegotiatorState {
@ -14,15 +16,15 @@ enum NegotiatorState {
done,
// Cancel the current attempt but we are not done
retryLater,
// The negotiator is in an error state
error,
// Skip the rest of the negotiation and assume the stream ready. Only use this when
// using stream restoration XEPs, like Stream Management.
skipRest,
}
class NegotiatorAttributes {
/// A base class for all errors that may occur during feature negotiation
abstract class NegotiatorError extends XmppError {}
class NegotiatorAttributes {
const NegotiatorAttributes(
this.sendNonza,
this.getConnectionSettings,
@ -97,7 +99,7 @@ abstract class XmppFeatureNegotiatorBase {
/// must switch some internal state to prevent getting matched immediately again.
/// If ready is returned, then the negotiator indicates that it is not done with
/// negotiation.
Future<void> negotiate(XMLNode nonza);
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza);
/// Reset the negotiator to a state that negotation can happen again.
void reset() {

View File

@ -4,9 +4,12 @@ import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
import 'package:moxxmpp/src/xeps/xep_0198/xep_0198.dart';
import 'package:uuid/uuid.dart';
class ResourceBindingFailedError extends NegotiatorError {}
class ResourceBindingNegotiator extends XmppFeatureNegotiatorBase {
ResourceBindingNegotiator() : _requestSent = false, super(0, false, bindXmlns, resourceBindingNegotiator);
@ -23,7 +26,7 @@ class ResourceBindingNegotiator extends XmppFeatureNegotiatorBase {
}
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
if (!_requestSent) {
final stanza = XMLNode.xmlns(
tag: 'iq',
@ -42,10 +45,10 @@ class ResourceBindingNegotiator extends XmppFeatureNegotiatorBase {
_requestSent = true;
attributes.sendNonza(stanza);
return const Result(NegotiatorState.ready);
} else {
if (nonza.tag != 'iq' || nonza.attributes['type'] != 'result') {
state = NegotiatorState.error;
return;
return Result(ResourceBindingFailedError());
}
final bind = nonza.firstTag('bind')!;
@ -53,7 +56,7 @@ class ResourceBindingNegotiator extends XmppFeatureNegotiatorBase {
final resource = jid.innerText().split('/')[1];
await attributes.sendEvent(ResourceBindingSuccessEvent(resource: resource));
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
}

View File

@ -0,0 +1,3 @@
import 'package:moxxmpp/src/negotiators/negotiator.dart';
class SaslFailedError extends NegotiatorError {}

View File

@ -1,12 +1,13 @@
import 'dart:convert';
import 'package:logging/logging.dart';
import 'package:moxxmpp/src/events.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/negotiators/sasl/errors.dart';
import 'package:moxxmpp/src/negotiators/sasl/negotiator.dart';
import 'package:moxxmpp/src/negotiators/sasl/nonza.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
class SaslPlainAuthNonza extends SaslAuthNonza {
SaslPlainAuthNonza(String username, String password) : super(
@ -15,7 +16,6 @@ class SaslPlainAuthNonza extends SaslAuthNonza {
}
class SaslPlainNegotiator extends SaslNegotiator {
SaslPlainNegotiator()
: _authSent = false,
_log = Logger('SaslPlainNegotiator'),
@ -41,7 +41,7 @@ class SaslPlainNegotiator extends SaslNegotiator {
}
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
if (!_authSent) {
final settings = attributes.getConnectionSettings();
attributes.sendNonza(
@ -49,17 +49,17 @@ class SaslPlainNegotiator extends SaslNegotiator {
redact: SaslPlainAuthNonza('******', '******').toXml(),
);
_authSent = true;
return const Result(NegotiatorState.ready);
} else {
final tag = nonza.tag;
if (tag == 'success') {
await attributes.sendEvent(AuthenticationSuccessEvent());
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
} else {
// We assume it's a <failure/>
final error = nonza.children.first.tag;
await attributes.sendEvent(AuthenticationFailedEvent(error));
state = NegotiatorState.error;
return Result(SaslFailedError());
}
}
}

View File

@ -1,16 +1,17 @@
import 'dart:convert';
import 'dart:math' show Random;
import 'package:cryptography/cryptography.dart';
import 'package:logging/logging.dart';
import 'package:moxxmpp/src/events.dart';
import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/negotiators/sasl/errors.dart';
import 'package:moxxmpp/src/negotiators/sasl/kv.dart';
import 'package:moxxmpp/src/negotiators/sasl/negotiator.dart';
import 'package:moxxmpp/src/negotiators/sasl/nonza.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
import 'package:random_string/random_string.dart';
import 'package:saslprep/saslprep.dart';
@ -89,7 +90,6 @@ enum ScramState {
const gs2Header = 'n,,';
class SaslScramNegotiator extends SaslNegotiator {
// NOTE: NEVER, and I mean, NEVER set clientNonce or initalMessageNoGS2. They are just there for testing
SaslScramNegotiator(
int priority,
@ -197,7 +197,7 @@ class SaslScramNegotiator extends SaslNegotiator {
}
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
switch (_scramState) {
case ScramState.preSent:
if (clientNonce == null || clientNonce == '') {
@ -211,15 +211,14 @@ class SaslScramNegotiator extends SaslNegotiator {
SaslScramAuthNonza(body: base64.encode(utf8.encode(gs2Header + initialMessageNoGS2)), type: hashType),
redact: SaslScramAuthNonza(body: '******', type: hashType).toXml(),
);
break;
return const Result(NegotiatorState.ready);
case ScramState.initialMessageSent:
if (nonza.tag != 'challenge') {
final error = nonza.children.first.tag;
await attributes.sendEvent(AuthenticationFailedEvent(error));
state = NegotiatorState.error;
_scramState = ScramState.error;
return;
return Result(SaslFailedError());
}
final challengeBase64 = nonza.innerText();
@ -230,15 +229,14 @@ class SaslScramNegotiator extends SaslNegotiator {
SaslScramResponseNonza(body: responseBase64),
redact: SaslScramResponseNonza(body: '******').toXml(),
);
return;
return const Result(NegotiatorState.ready);
case ScramState.challengeResponseSent:
if (nonza.tag != 'success') {
// We assume it's a <failure />
final error = nonza.children.first.tag;
await attributes.sendEvent(AuthenticationFailedEvent(error));
_scramState = ScramState.error;
state = NegotiatorState.error;
return;
return Result(SaslFailedError());
}
// NOTE: This assumes that the string is always "v=..." and contains no other parameters
@ -248,16 +246,13 @@ class SaslScramNegotiator extends SaslNegotiator {
//final error = nonza.children.first.tag;
//attributes.sendEvent(AuthenticationFailedEvent(error));
_scramState = ScramState.error;
state = NegotiatorState.error;
return;
return Result(SaslFailedError());
}
await attributes.sendEvent(AuthenticationSuccessEvent());
state = NegotiatorState.done;
return;
return const Result(NegotiatorState.done);
case ScramState.error:
state = NegotiatorState.error;
return;
return Result(SaslFailedError());
}
}

View File

@ -3,12 +3,15 @@ import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
enum _StartTlsState {
ready,
requested
}
class StartTLSFailedError extends NegotiatorError {}
class StartTLSNonza extends XMLNode {
StartTLSNonza() : super.xmlns(
tag: 'starttls',
@ -27,18 +30,17 @@ class StartTlsNegotiator extends XmppFeatureNegotiatorBase {
final Logger _log;
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
switch (_state) {
case _StartTlsState.ready:
_log.fine('StartTLS is available. Performing StartTLS upgrade...');
_state = _StartTlsState.requested;
attributes.sendNonza(StartTLSNonza());
break;
return const Result(NegotiatorState.ready);
case _StartTlsState.requested:
if (nonza.tag != 'proceed' || nonza.attributes['xmlns'] != startTlsXmlns) {
_log.severe('Failed to perform StartTLS negotiation');
state = NegotiatorState.error;
return;
return Result(StartTLSFailedError());
}
_log.fine('Securing socket');
@ -46,13 +48,11 @@ class StartTlsNegotiator extends XmppFeatureNegotiatorBase {
.secure(attributes.getConnectionSettings().jid.domain);
if (!result) {
_log.severe('Failed to secure stream');
state = NegotiatorState.error;
return;
return Result(StartTLSFailedError());
}
_log.fine('Stream is now TLS secured');
state = NegotiatorState.done;
break;
return const Result(NegotiatorState.done);
}
}

View File

@ -80,10 +80,11 @@ abstract class ReconnectionPolicy {
/// for every failed attempt.
class ExponentialBackoffReconnectionPolicy extends ReconnectionPolicy {
ExponentialBackoffReconnectionPolicy()
ExponentialBackoffReconnectionPolicy(this._maxBackoffTime)
: _counter = 0,
_log = Logger('ExponentialBackoffReconnectionPolicy'),
super();
final int _maxBackoffTime;
int _counter;
Timer? _timer;
final Logger _log;
@ -124,7 +125,7 @@ class ExponentialBackoffReconnectionPolicy extends ReconnectionPolicy {
}
// Wait at max 80 seconds.
final seconds = min(pow(2, _counter).toInt(), 80);
final seconds = min(min(pow(2, _counter).toInt(), 80), _maxBackoffTime);
_timer = Timer(Duration(seconds: seconds), _onTimerElapsed);
}

View File

@ -0,0 +1,7 @@
abstract class RosterError {}
/// Returned when the server's response did not contain a <query /> element
class NoQueryError extends RosterError {}
/// Unspecified error
class UnknownError extends RosterError {}

View File

@ -7,15 +7,12 @@ import 'package:moxxmpp/src/managers/namespaces.dart';
import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/roster/errors.dart';
import 'package:moxxmpp/src/stanza.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/error.dart';
const rosterErrorNoQuery = 1;
const rosterErrorNonResult = 2;
import 'package:moxxmpp/src/types/result.dart';
class XmppRosterItem {
XmppRosterItem({ required this.jid, required this.subscription, this.ask, this.name, this.groups = const [] });
final String jid;
final String? name;
@ -31,14 +28,12 @@ enum RosterRemovalResult {
}
class RosterRequestResult {
RosterRequestResult({ required this.items, this.ver });
List<XmppRosterItem> items;
String? ver;
}
class RosterPushEvent extends XmppEvent {
RosterPushEvent({ required this.item, this.ver });
final XmppRosterItem item;
final String? ver;
@ -53,11 +48,11 @@ class RosterFeatureNegotiator extends XmppFeatureNegotiatorBase {
bool get isSupported => _supported;
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
// negotiate is only called when the negotiator matched, meaning the server
// advertises roster versioning.
_supported = true;
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
@override
@ -148,7 +143,7 @@ class RosterManager extends XmppManagerBase {
/// Shared code between requesting rosters without and with roster versioning, if
/// the server deems a regular roster response more efficient than n roster pushes.
Future<MayFail<RosterRequestResult>> _handleRosterResponse(XMLNode? query) async {
Future<Result<RosterRequestResult, RosterError>> _handleRosterResponse(XMLNode? query) async {
final List<XmppRosterItem> items;
if (query != null) {
items = query.children.map((item) => XmppRosterItem(
@ -166,7 +161,7 @@ class RosterManager extends XmppManagerBase {
}
} else {
logger.warning('Server response to roster request without roster versioning does not contain a <query /> element, while the type is not error. This violates RFC6121');
return MayFail.failure(rosterErrorNoQuery);
return Result(NoQueryError());
}
final ver = query.attributes['ver'] as String?;
@ -175,7 +170,7 @@ class RosterManager extends XmppManagerBase {
await commitLastRosterVersion(ver);
}
return MayFail.success(
return Result(
RosterRequestResult(
items: items,
ver: ver,
@ -185,7 +180,7 @@ class RosterManager extends XmppManagerBase {
}
/// Requests the roster following RFC 6121 without using roster versioning.
Future<MayFail<RosterRequestResult>> requestRoster() async {
Future<Result<RosterRequestResult, RosterError>> requestRoster() async {
final attrs = getAttributes();
final response = await attrs.sendStanza(
Stanza.iq(
@ -201,7 +196,7 @@ class RosterManager extends XmppManagerBase {
if (response.attributes['type'] != 'result') {
logger.warning('Error requesting roster without roster versioning: ${response.toXml()}');
return MayFail.failure(rosterErrorNonResult);
return Result(UnknownError());
}
final query = response.firstTag('query', xmlns: rosterXmlns);
@ -210,7 +205,7 @@ class RosterManager extends XmppManagerBase {
/// Requests a series of roster pushes according to RFC6121. Requires that the server
/// advertises urn:xmpp:features:rosterver in the stream features.
Future<MayFail<RosterRequestResult?>> requestRosterPushes() async {
Future<Result<RosterRequestResult?, RosterError>> requestRosterPushes() async {
if (_rosterVersion == null) {
await loadLastRosterVersion();
}
@ -233,7 +228,7 @@ class RosterManager extends XmppManagerBase {
if (result.attributes['type'] != 'result') {
logger.warning('Requesting roster pushes failed: ${result.toXml()}');
return MayFail.failure(rosterErrorNonResult);
return Result(UnknownError());
}
final query = result.firstTag('query', xmlns: rosterXmlns);

View File

@ -1,19 +0,0 @@
/// A wrapper class that can be used to indicate that a function may return a valid
/// instance of [T] but may also fail.
/// The way [MayFail] is intended to be used to to have function specific - or application
/// specific - error codes that can be either handled by code or be translated into a
/// localised error message for the user.
class MayFail<T> {
MayFail({ this.result, this.errorCode });
MayFail.success(this.result);
MayFail.failure(this.errorCode);
T? result;
int? errorCode;
bool isError() => result == null && errorCode != null;
T getValue() => result!;
int getErrorCode() => errorCode!;
}

View File

@ -5,6 +5,7 @@ import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
import 'package:moxxmpp/src/xeps/xep_0198/nonzas.dart';
import 'package:moxxmpp/src/xeps/xep_0198/state.dart';
import 'package:moxxmpp/src/xeps/xep_0198/xep_0198.dart';
@ -23,7 +24,6 @@ enum _StreamManagementNegotiatorState {
/// StreamManagementManager at least once before connecting, if stream resumption
/// is wanted.
class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
StreamManagementNegotiator()
: _state = _StreamManagementNegotiatorState.ready,
_supported = false,
@ -59,7 +59,7 @@ class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
}
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
// negotiate is only called when we matched the stream feature, so we know
// that the server advertises it.
_supported = true;
@ -80,7 +80,8 @@ class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
_state = _StreamManagementNegotiatorState.enableRequested;
attributes.sendNonza(StreamManagementEnableNonza());
}
break;
return const Result(NegotiatorState.ready);
case _StreamManagementNegotiatorState.resumeRequested:
if (nonza.tag == 'resumed') {
_log.finest('Stream Management resumption successful');
@ -97,7 +98,7 @@ class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
_resumeFailed = false;
_isResumed = true;
state = NegotiatorState.skipRest;
return const Result(NegotiatorState.skipRest);
} else {
// We assume it is <failed />
_log.info('Stream resumption failed. Expected <resumed />, got ${nonza.tag}, Proceeding with new stream...');
@ -113,9 +114,8 @@ class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
_resumeFailed = true;
_isResumed = false;
_state = _StreamManagementNegotiatorState.ready;
state = NegotiatorState.retryLater;
return const Result(NegotiatorState.retryLater);
}
break;
case _StreamManagementNegotiatorState.enableRequested:
if (nonza.tag == 'enabled') {
_log.finest('Stream Management enabled');
@ -133,14 +133,12 @@ class StreamManagementNegotiator extends XmppFeatureNegotiatorBase {
),
);
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
} else {
// We assume a <failed />
_log.warning('Stream Management enablement failed');
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
break;
}
}

View File

@ -4,6 +4,7 @@ import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/negotiators/namespaces.dart';
import 'package:moxxmpp/src/negotiators/negotiator.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/result.dart';
class CSIActiveNonza extends XMLNode {
CSIActiveNonza() : super(
@ -32,11 +33,11 @@ class CSINegotiator extends XmppFeatureNegotiatorBase {
bool get isSupported => _supported;
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
// negotiate is only called when the negotiator matched, meaning the server
// advertises CSI.
_supported = true;
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
@override

View File

@ -0,0 +1,10 @@
abstract class HttpFileUploadError {}
/// Returned when we don't know what JID to ask for an upload slot
class NoEntityKnownError extends HttpFileUploadError {}
/// Returned when the file we want to upload is too big
class FileTooBigError extends HttpFileUploadError {}
/// Unspecified errors
class UnknownHttpFileUploadError extends HttpFileUploadError {}

View File

@ -7,19 +7,15 @@ import 'package:moxxmpp/src/managers/namespaces.dart';
import 'package:moxxmpp/src/namespaces.dart';
import 'package:moxxmpp/src/stanza.dart';
import 'package:moxxmpp/src/stringxml.dart';
import 'package:moxxmpp/src/types/error.dart';
import 'package:moxxmpp/src/types/result.dart';
import 'package:moxxmpp/src/xeps/xep_0030/errors.dart';
import 'package:moxxmpp/src/xeps/xep_0030/types.dart';
import 'package:moxxmpp/src/xeps/xep_0030/xep_0030.dart';
const errorNoUploadServer = 1;
const errorFileTooBig = 2;
const errorGeneric = 3;
import 'package:moxxmpp/src/xeps/xep_0363/errors.dart';
const allowedHTTPHeaders = [ 'authorization', 'cookie', 'expires' ];
class HttpFileUploadSlot {
const HttpFileUploadSlot(this.putUrl, this.getUrl, this.headers);
final String putUrl;
final String getUrl;
@ -45,7 +41,6 @@ Map<String, String> prepareHeaders(Map<String, String> headers) {
}
class HttpFileUploadManager extends XmppManagerBase {
HttpFileUploadManager() : _gotSupported = false, _supported = false, super();
JID? _entityJid;
int? _maxUploadSize;
@ -119,17 +114,17 @@ class HttpFileUploadManager extends XmppManagerBase {
/// the file's size in octets. [contentType] is optional and refers to the file's
/// Mime type.
/// Returns an [HttpFileUploadSlot] if the request was successful; null otherwise.
Future<MayFail<HttpFileUploadSlot>> requestUploadSlot(String filename, int filesize, { String? contentType }) async {
if (!(await isSupported())) return MayFail.failure(errorNoUploadServer);
Future<Result<HttpFileUploadSlot, HttpFileUploadError>> requestUploadSlot(String filename, int filesize, { String? contentType }) async {
if (!(await isSupported())) return Result(NoEntityKnownError());
if (_entityJid == null) {
logger.warning('Attempted to request HTTP File Upload slot but no entity is known to send this request to.');
return MayFail.failure(errorNoUploadServer);
return Result(NoEntityKnownError());
}
if (_maxUploadSize != null && filesize > _maxUploadSize!) {
logger.warning('Attempted to request HTTP File Upload slot for a file that exceeds the filesize limit');
return MayFail.failure(errorFileTooBig);
return Result(FileTooBigError());
}
final attrs = getAttributes();
@ -154,7 +149,7 @@ class HttpFileUploadManager extends XmppManagerBase {
if (response.attributes['type']! != 'result') {
logger.severe('Failed to request HTTP File Upload slot.');
// TODO(Unknown): Be more precise
return MayFail.failure(errorGeneric);
return Result(UnknownHttpFileUploadError());
}
final slot = response.firstTag('slot', xmlns: httpFileUploadXmlns)!;
@ -169,7 +164,7 @@ class HttpFileUploadManager extends XmppManagerBase {
}),
);
return MayFail.success(
return Result(
HttpFileUploadSlot(
putUrl,
getUrl,

View File

@ -44,7 +44,6 @@ const _doNotEncryptList = [
];
abstract class OmemoManager extends XmppManagerBase {
OmemoManager() : _handlerLock = Lock(), _handlerFutures = {}, super();
final Lock _handlerLock;

View File

@ -3,10 +3,10 @@ import 'package:test/test.dart';
import 'helpers/logging.dart';
import 'helpers/xmpp.dart';
const exampleXmlns1 = 'im:moxxy:example1';
const exampleNamespace1 = 'im.moxxy.test.example1';
const exampleXmlns2 = 'im:moxxy:example2';
const exampleNamespace2 = 'im.moxxy.test.example2';
const exampleXmlns1 = 'im:moxxmpp:example1';
const exampleNamespace1 = 'im.moxxmpp.test.example1';
const exampleXmlns2 = 'im:moxxmpp:example2';
const exampleNamespace2 = 'im.moxxmpp.test.example2';
class StubNegotiator1 extends XmppFeatureNegotiatorBase {
StubNegotiator1() : called = false, super(1, false, exampleXmlns1, exampleNamespace1);
@ -14,9 +14,9 @@ class StubNegotiator1 extends XmppFeatureNegotiatorBase {
bool called;
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
called = true;
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
}
@ -26,9 +26,9 @@ class StubNegotiator2 extends XmppFeatureNegotiatorBase {
bool called;
@override
Future<void> negotiate(XMLNode nonza) async {
Future<Result<NegotiatorState, NegotiatorError>> negotiate(XMLNode nonza) async {
called = true;
state = NegotiatorState.done;
return const Result(NegotiatorState.done);
}
}
@ -47,8 +47,8 @@ void main() {
from="test.server"
xml:lang="en">
<stream:features xmlns="http://etherx.jabber.org/streams">
<example1 xmlns="im:moxxy:example1" />
<example2 xmlns="im:moxxy:example2" />
<example1 xmlns="im:moxxmpp:example1" />
<example2 xmlns="im:moxxmpp:example2" />
</stream:features>''',
),
],

View File

@ -128,9 +128,9 @@ void main() {
'c=biws,r=rOprNGfwEbeRWgbNEkqO%hvYDpWUa2RaTCAfuxFIlj)hNlF\$k0,p=dHzbZapWIk4jUhN+Ute9ytag9zjfMHgsqmmiz7AndVQ=',
);
await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj02cnJpVFJCaTIzV3BSUi93dHVwK21NaFVaVW4vZEI1bkxUSlJzamw5NUc0PQ==</success>"));
final result = await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj02cnJpVFJCaTIzV3BSUi93dHVwK21NaFVaVW4vZEI1bkxUSlJzamw5NUc0PQ==</success>"));
expect(negotiator.state, NegotiatorState.done);
expect(result.get<NegotiatorState>(), NegotiatorState.done);
});
test('Test a positive server signature check', () async {
@ -150,9 +150,9 @@ void main() {
await negotiator.negotiate(scramSha1StreamFeatures);
await negotiator.negotiate(XMLNode.fromString("<challenge xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>cj1meWtvK2QybGJiRmdPTlJ2OXFreGRhd0wzcmZjTkhZSlkxWlZ2V1ZzN2oscz1RU1hDUitRNnNlazhiZjkyLGk9NDA5Ng==</challenge>"));
await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
final result = await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(negotiator.state, NegotiatorState.done);
expect(result.get<NegotiatorState>(), NegotiatorState.done);
});
test('Test a negative server signature check', () async {
@ -170,11 +170,15 @@ void main() {
),
);
await negotiator.negotiate(scramSha1StreamFeatures);
await negotiator.negotiate(XMLNode.fromString("<challenge xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>cj1meWtvK2QybGJiRmdPTlJ2OXFreGRhd0wzcmZjTkhZSlkxWlZ2V1ZzN2oscz1RU1hDUitRNnNlazhiZjkyLGk9NDA5Ng==</challenge>"));
await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1zbUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
var result;
result = await negotiator.negotiate(scramSha1StreamFeatures);
expect(result.isType<NegotiatorState>(), true);
expect(negotiator.state, NegotiatorState.error);
result = await negotiator.negotiate(XMLNode.fromString("<challenge xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>cj1meWtvK2QybGJiRmdPTlJ2OXFreGRhd0wzcmZjTkhZSlkxWlZ2V1ZzN2oscz1RU1hDUitRNnNlazhiZjkyLGk9NDA5Ng==</challenge>"));
expect(result.isType<NegotiatorState>(), true);
result = await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1zbUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(result.isType<NegotiatorError>(), true);
});
test('Test a resetting the SCRAM negotiator', () async {
@ -194,14 +198,14 @@ void main() {
await negotiator.negotiate(scramSha1StreamFeatures);
await negotiator.negotiate(XMLNode.fromString("<challenge xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>cj1meWtvK2QybGJiRmdPTlJ2OXFreGRhd0wzcmZjTkhZSlkxWlZ2V1ZzN2oscz1RU1hDUitRNnNlazhiZjkyLGk9NDA5Ng==</challenge>"));
await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(negotiator.state, NegotiatorState.done);
final result1 = await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(result1.get<NegotiatorState>(), NegotiatorState.done);
// Reset and try again
negotiator.reset();
await negotiator.negotiate(scramSha1StreamFeatures);
await negotiator.negotiate(XMLNode.fromString("<challenge xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>cj1meWtvK2QybGJiRmdPTlJ2OXFreGRhd0wzcmZjTkhZSlkxWlZ2V1ZzN2oscz1RU1hDUitRNnNlazhiZjkyLGk9NDA5Ng==</challenge>"));
await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(negotiator.state, NegotiatorState.done);
final result2 = await negotiator.negotiate(XMLNode.fromString("<success xmlns='urn:ietf:params:xml:ns:xmpp-sasl'>dj1ybUY5cHFWOFM3c3VBb1pXamE0ZEpSa0ZzS1E9</success>"));
expect(result2.get<NegotiatorState>(), NegotiatorState.done);
});
}