.gitlint

@@ -7,7 +7,7 @@ line-length=72
 [title-trailing-punctuation]
 [title-hard-tab]
 [title-match-regex]
-regex=^((feat|fix|chore|refactor|docs|release|test)\((meta|tests|style|docs|xep|core|example|all)+(,(meta|tests|style|docs|xep|core|example|all))*\)|release): [A-Z0-9].*$
+regex=^((feat|fix|chore|refactor|docs|release|test)\((meta|tests|style|docs|xep|core|example)+(,(meta|tests|style|docs|xep|core|example))*\)|release): [A-Z0-9].*$
 
 
 [body-trailing-whitespace]

packages/moxxmpp/CHANGELOG.md

@@ -9,8 +9,6 @@
 - **BREAKING**: Remove `DiscoManager.discoInfoCapHashQuery`.
 - **BREAKING**: The entity argument of `DiscoManager.discoInfoQuery` and `DiscoManager.discoItemsQuery` are now `JID` instead of `String`.
 - **BREAKING**: `PubSubManager` and `UserAvatarManager` now use `JID` instead of `String`.
-- **BREAKING**: `XmppConnection.sendStanza` not only takes a `StanzaDetails` argument.
-- Sent stanzas are not kept in a queue until sent.
 
 ## 0.3.1
 

packages/moxxmpp/lib/src/connection.dart

@@ -26,7 +26,6 @@ import 'package:moxxmpp/src/socket.dart';
 import 'package:moxxmpp/src/stanza.dart';
 import 'package:moxxmpp/src/stringxml.dart';
 import 'package:moxxmpp/src/types/result.dart';
-import 'package:moxxmpp/src/util/queue.dart';
 import 'package:moxxmpp/src/xeps/xep_0030/xep_0030.dart';
 import 'package:moxxmpp/src/xeps/xep_0198/xep_0198.dart';
 import 'package:moxxmpp/src/xeps/xep_0352.dart';
@@ -49,6 +48,18 @@ enum XmppConnectionState {
   error
 }
 
+/// Metadata for [XmppConnection.sendStanza].
+enum StanzaFromType {
+  /// Add the full JID to the stanza as the from attribute
+  full,
+
+  /// Add the bare JID to the stanza as the from attribute
+  bare,
+
+  /// Add no JID as the from attribute
+  none,
+}
+
 /// This class is a connection to the server.
 class XmppConnection {
   XmppConnection(
@@ -80,12 +91,7 @@ class XmppConnection {
     _socketStream = _socket.getDataStream();
     // TODO(Unknown): Handle on done
     _socketStream.transform(_streamParser).forEach(handleXmlStream);
-    _socket.getEventStream().listen(handleSocketEvent);
-
-    _stanzaQueue = AsyncStanzaQueue(
-      _sendStanzaImpl,
-      _canSendData,
-    );
+    _socket.getEventStream().listen(_handleSocketEvent);
   }
 
   /// The state that the connection is currently in
@@ -169,8 +175,6 @@ class XmppConnection {
 
   bool get isAuthenticated => _isAuthenticated;
 
-  late final AsyncStanzaQueue _stanzaQueue;
-
   /// Returns the JID we authenticate with and add the resource that we have bound.
   JID _getJidWithResource() {
     assert(_resource.isNotEmpty, 'The resource must not be empty');
@@ -362,8 +366,7 @@ class XmppConnection {
   }
 
   /// Called whenever the socket creates an event
-  @visibleForTesting
-  Future<void> handleSocketEvent(XmppSocketEvent event) async {
+  Future<void> _handleSocketEvent(XmppSocketEvent event) async {
     if (event is XmppSocketErrorEvent) {
       await handleError(SocketError(event));
     } else if (event is XmppSocketClosureEvent) {
@@ -409,135 +412,133 @@ class XmppConnection {
         .contains(await getConnectionState());
   }
 
-  /// Sends a stanza described by [details] to the server. Until sent, the stanza is
-  /// kept in a queue, that is flushed after going online again. If Stream Management
-  /// is active, stanza's acknowledgement is tracked.
+  /// Sends a [stanza] to the server. If stream management is enabled, then keeping track
+  /// of the stanza is taken care of. Returns a Future that resolves when we receive a
+  /// response to the stanza.
+  ///
+  /// If addFrom is true, then a 'from' attribute will be added to the stanza if
+  /// [stanza] has none.
+  /// If addId is true, then an 'id' attribute will be added to the stanza if [stanza] has
+  /// none.
   // TODO(Unknown): if addId = false, the function crashes.
-  Future<XMLNode?> sendStanza(StanzaDetails details) async {
+  Future<XMLNode> sendStanza(
+    Stanza stanza, {
+    StanzaFromType addFrom = StanzaFromType.full,
+    bool addId = true,
+    bool awaitable = true,
+    bool encrypted = false,
+    bool forceEncryption = false,
+  }) async {
     assert(
-      implies(
-        details.awaitable,
-        details.stanza.id != null && details.stanza.id!.isNotEmpty ||
-            details.addId,
-      ),
-      'An awaitable stanza must have an id',
+      implies(addId == false && stanza.id == null, !awaitable),
+      'Cannot await a stanza with no id',
     );
 
-    final completer = details.awaitable ? Completer<XMLNode>() : null;
-    await _stanzaQueue.enqueueStanza(
-      StanzaQueueEntry(
-        details,
-        completer,
-      ),
-    );
-
-    return completer?.future;
-  }
-
-  Future<void> _sendStanzaImpl(StanzaQueueEntry entry) async {
-    final details = entry.details;
-    var newStanza = details.stanza;
-
-    // Generate an id, if requested
-    if (details.addId && (newStanza.id == null || newStanza.id == '')) {
-      newStanza = newStanza.copyWith(id: generateId());
+    // Add extra data in case it was not set
+    var stanza_ = stanza;
+    if (addId && (stanza_.id == null || stanza_.id == '')) {
+      stanza_ = stanza.copyWith(id: generateId());
     }
-
-    // NOTE: Originally, we handled adding a "from" attribute to the stanza here.
-    //       However, this is not neccessary as RFC 6120 states:
-    //
-    //       > When a server receives an XML stanza from a connected client, the
-    //       > server MUST add a 'from' attribute to the stanza or override the
-    //       > 'from' attribute specified by the client, where the value of the
-    //       > 'from' attribute MUST be the full JID
-    //       > (<localpart@domainpart/resource>) determined by the server for
-    //       > the connected resource that generated the stanza (see
-    //       > Section 4.3.6), or the bare JID (<localpart@domainpart>) in the
-    //       > case of subscription-related presence stanzas (see [XMPP-IM]).
-    //
-    //       This means that even if we add a "from" attribute, the server will discard
-    //       it. If we don't specify it, then the server will add the correct value
-    //       itself.
-
-    // Add the correct stanza namespace
-    newStanza = newStanza.copyWith(
+    if (addFrom != StanzaFromType.none &&
+        (stanza_.from == null || stanza_.from == '')) {
+      switch (addFrom) {
+        case StanzaFromType.full:
+          {
+            stanza_ = stanza_.copyWith(
+              from: _getJidWithResource().toString(),
+            );
+          }
+          break;
+        case StanzaFromType.bare:
+          {
+            stanza_ = stanza_.copyWith(
+              from: connectionSettings.jid.toBare().toString(),
+            );
+          }
+          break;
+        case StanzaFromType.none:
+          break;
+      }
+    }
+    stanza_ = stanza_.copyWith(
       xmlns: _negotiationsHandler.getStanzaNamespace(),
     );
 
-    // Run pre-send handlers
     _log.fine('Running pre stanza handlers..');
     final data = await _runOutgoingPreStanzaHandlers(
-      newStanza,
+      stanza_,
       initial: StanzaHandlerData(
         false,
         false,
         null,
-        newStanza,
-        encrypted: details.encrypted,
-        forceEncryption: details.forceEncryption,
+        stanza_,
+        encrypted: encrypted,
+        forceEncryption: forceEncryption,
       ),
     );
     _log.fine('Done');
 
-    // Cancel sending, if the pre-send handlers indicated it.
     if (data.cancel) {
       _log.fine('A stanza handler indicated that it wants to cancel sending.');
       await _sendEvent(StanzaSendingCancelledEvent(data));
-
-      // Resolve the future, if one was given.
-      if (details.awaitable) {
-        entry.completer!.complete(
-          Stanza(
-            tag: data.stanza.tag,
-            to: data.stanza.from,
-            from: data.stanza.to,
-            attributes: <String, String>{
-              'type': 'error',
-              if (data.stanza.id != null) 'id': data.stanza.id!,
-            },
-          ),
-        );
-      }
-      return;
+      return Stanza(
+        tag: data.stanza.tag,
+        to: data.stanza.from,
+        from: data.stanza.to,
+        attributes: <String, String>{
+          'type': 'error',
+          ...data.stanza.id != null
+              ? {
+                  'id': data.stanza.id!,
+                }
+              : {},
+        },
+      );
     }
 
-    // Log the (raw) stanza
     final prefix = data.encrypted ? '(Encrypted) ' : '';
-    _log.finest('==> $prefix${newStanza.toXml()}');
+    _log.finest('==> $prefix${stanza_.toXml()}');
 
-    if (details.awaitable) {
-      await _stanzaAwaiter
-          .addPending(
+    final stanzaString = data.stanza.toXml();
+
+    // ignore: cascade_invocations
+    _log.fine('Attempting to acquire lock for ${data.stanza.id}...');
+    // TODO(PapaTutuWawa): Handle this much more graceful
+    var future = Future.value(XMLNode(tag: 'not-used'));
+    if (awaitable) {
+      future = await _stanzaAwaiter.addPending(
         // A stanza with no to attribute is for direct processing by the server. As such,
         // we can correlate it by just *assuming* we have that attribute
         // (RFC 6120 Section 8.1.1.1)
         data.stanza.to ?? connectionSettings.jid.toBare().toString(),
         data.stanza.id!,
         data.stanza.tag,
-      )
-          .then((result) {
-        entry.completer!.complete(result);
-      });
+      );
     }
 
+    // This uses the StreamManager to behave like a send queue
     if (await _canSendData()) {
-      _socket.write(data.stanza.toXml());
+      _socket.write(stanzaString);
+
+      // Try to ack every stanza
+      // NOTE: Here we have send an Ack request nonza. This is now done by StreamManagementManager when receiving the StanzaSentEvent
     } else {
-      _log.fine('Not sending dat as _canSendData() returned false.');
+      _log.fine('_canSendData() returned false.');
     }
 
-    // Run post-send handlers
     _log.fine('Running post stanza handlers..');
     await _runOutgoingPostStanzaHandlers(
-      newStanza,
+      stanza_,
       initial: StanzaHandlerData(
         false,
         false,
         null,
-        newStanza,
+        stanza_,
       ),
     );
     _log.fine('Done');
+
+    return future;
   }
 
   /// Called when we timeout during connecting
@@ -561,11 +562,18 @@ class XmppConnection {
     // Set the new routing state
     _updateRoutingState(RoutingState.handleStanzas);
 
+    // Set the connection state
+    await _setConnectionState(XmppConnectionState.connected);
+
     // Enable reconnections
     if (_enableReconnectOnSuccess) {
       await _reconnectionPolicy.setShouldReconnect(true);
     }
 
+    // Resolve the connection completion future
+    _connectionCompleter?.complete(const Result(true));
+    _connectionCompleter = null;
+
     // Tell consumers of the event stream that we're done with stream feature
     // negotiations
     await _sendEvent(
@@ -574,16 +582,6 @@ class XmppConnection {
             false,
       ),
     );
-
-    // Set the connection state
-    await _setConnectionState(XmppConnectionState.connected);
-
-    // Resolve the connection completion future
-    _connectionCompleter?.complete(const Result(true));
-    _connectionCompleter = null;
-
-    // Flush the stanza send queue
-    await _stanzaQueue.restart();
   }
 
   /// Sets the connection state to [state] and triggers an event of type

packages/moxxmpp/lib/src/iq.dart

@@ -23,11 +23,9 @@ Future<void> handleUnhandledStanza(
     );
 
     await conn.sendStanza(
-      StanzaDetails(
-        stanza,
-        awaitable: false,
-        forceEncryption: data.encrypted,
-      ),
+      stanza,
+      awaitable: false,
+      forceEncryption: data.encrypted,
     );
   }
 }

packages/moxxmpp/lib/src/managers/attributes.dart

@@ -23,7 +23,14 @@ class XmppManagerAttributes {
   });
 
   /// Send a stanza whose response can be awaited.
-  final Future<XMLNode?> Function(StanzaDetails) sendStanza;
+  final Future<XMLNode> Function(
+    Stanza stanza, {
+    StanzaFromType addFrom,
+    bool addId,
+    bool awaitable,
+    bool encrypted,
+    bool forceEncryption,
+  }) sendStanza;
 
   /// Send a nonza.
   final void Function(XMLNode) sendNonza;

packages/moxxmpp/lib/src/managers/base.dart

@@ -5,7 +5,6 @@ import 'package:moxxmpp/src/managers/attributes.dart';
 import 'package:moxxmpp/src/managers/data.dart';
 import 'package:moxxmpp/src/managers/handlers.dart';
 import 'package:moxxmpp/src/managers/namespaces.dart';
-import 'package:moxxmpp/src/stanza.dart';
 import 'package:moxxmpp/src/stringxml.dart';
 import 'package:moxxmpp/src/xeps/xep_0030/errors.dart';
 import 'package:moxxmpp/src/xeps/xep_0030/types.dart';
@@ -166,11 +165,9 @@ abstract class XmppManagerBase {
     );
 
     await getAttributes().sendStanza(
-      StanzaDetails(
-        stanza,
-        awaitable: false,
-        forceEncryption: data.encrypted,
-      ),
+      stanza,
+      awaitable: false,
+      forceEncryption: data.encrypted,
     );
   }
 }

packages/moxxmpp/lib/src/message.dart

@@ -320,11 +320,6 @@ class MessageManager extends XmppManagerBase {
       );
     }
 
-    getAttributes().sendStanza(
-      StanzaDetails(
-        stanza,
-        awaitable: false,
-      ),
-    );
+    getAttributes().sendStanza(stanza, awaitable: false);
   }
 }

packages/moxxmpp/lib/src/presence.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'package:moxxmpp/src/connection.dart';
 import 'package:moxxmpp/src/events.dart';
 import 'package:moxxmpp/src/jid.dart';
 import 'package:moxxmpp/src/managers/base.dart';
@@ -6,8 +7,10 @@ import 'package:moxxmpp/src/managers/data.dart';
 import 'package:moxxmpp/src/managers/handlers.dart';
 import 'package:moxxmpp/src/managers/namespaces.dart';
 import 'package:moxxmpp/src/namespaces.dart';
+import 'package:moxxmpp/src/negotiators/namespaces.dart';
 import 'package:moxxmpp/src/stanza.dart';
 import 'package:moxxmpp/src/stringxml.dart';
+import 'package:moxxmpp/src/xeps/xep_0198/negotiator.dart';
 
 /// A function that will be called when presence, outside of subscription request
 /// management, will be sent. Useful for managers that want to add [XMLNode]s to said
@@ -46,8 +49,12 @@ class PresenceManager extends XmppManagerBase {
   Future<void> onXmppEvent(XmppEvent event) async {
     if (event is StreamNegotiationsDoneEvent) {
       // Send initial presence only when we have not resumed the stream
-      if (!event.resumed) {
-        await sendInitialPresence();
+      final sm = getAttributes().getNegotiatorById<StreamManagementNegotiator>(
+        streamManagementNegotiator,
+      );
+      final isResumed = sm?.isResumed ?? false;
+      if (!isResumed) {
+        unawaited(sendInitialPresence());
       }
     }
   }
@@ -101,77 +108,66 @@ class PresenceManager extends XmppManagerBase {
 
     final attrs = getAttributes();
     await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          children: children,
-        ),
-        awaitable: false,
-        addId: false,
+      Stanza.presence(
+        from: attrs.getFullJID().toString(),
+        children: children,
       ),
+      awaitable: false,
+      addId: false,
     );
   }
 
   /// Send an unavailable presence with no 'to' attribute.
   void sendUnavailablePresence() {
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          type: 'unavailable',
-        ),
-        awaitable: false,
+      Stanza.presence(
+        type: 'unavailable',
       ),
+      addFrom: StanzaFromType.full,
     );
   }
 
   /// Sends a subscription request to [to].
   void sendSubscriptionRequest(String to) {
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          type: 'subscribe',
-          to: to,
-        ),
-        awaitable: false,
+      Stanza.presence(
+        type: 'subscribe',
+        to: to,
       ),
+      addFrom: StanzaFromType.none,
     );
   }
 
   /// Sends an unsubscription request to [to].
   void sendUnsubscriptionRequest(String to) {
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          type: 'unsubscribe',
-          to: to,
-        ),
-        awaitable: false,
+      Stanza.presence(
+        type: 'unsubscribe',
+        to: to,
       ),
+      addFrom: StanzaFromType.none,
     );
   }
 
   /// Accept a presence subscription request for [to].
   void sendSubscriptionRequestApproval(String to) {
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          type: 'subscribed',
-          to: to,
-        ),
-        awaitable: false,
+      Stanza.presence(
+        type: 'subscribed',
+        to: to,
       ),
+      addFrom: StanzaFromType.none,
     );
   }
 
   /// Reject a presence subscription request for [to].
   void sendSubscriptionRequestRejection(String to) {
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.presence(
-          type: 'unsubscribed',
-          to: to,
-        ),
-        awaitable: false,
+      Stanza.presence(
+        type: 'unsubscribed',
+        to: to,
       ),
+      addFrom: StanzaFromType.none,
     );
   }
 }

packages/moxxmpp/lib/src/roster/roster.dart

@@ -235,16 +235,14 @@ class RosterManager extends XmppManagerBase {
       query.attributes['ver'] = rosterVersion;
     }
 
-    final response = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          children: [
-            query,
-          ],
-        ),
+    final response = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'get',
+        children: [
+          query,
+        ],
       ),
-    ))!;
+    );
 
     if (response.attributes['type'] != 'result') {
       logger.warning('Error requesting roster: ${response.toXml()}');
@@ -260,22 +258,20 @@ class RosterManager extends XmppManagerBase {
   Future<Result<RosterRequestResult?, RosterError>>
       requestRosterPushes() async {
     final attrs = getAttributes();
-    final result = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          children: [
-            XMLNode.xmlns(
-              tag: 'query',
-              xmlns: rosterXmlns,
-              attributes: {
-                'ver': await _stateManager.getRosterVersion() ?? '',
-              },
-            )
-          ],
-        ),
+    final result = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'get',
+        children: [
+          XMLNode.xmlns(
+            tag: 'query',
+            xmlns: rosterXmlns,
+            attributes: {
+              'ver': await _stateManager.getRosterVersion() ?? '',
+            },
+          )
+        ],
       ),
-    ))!;
+    );
 
     if (result.attributes['type'] != 'result') {
       logger.warning('Requesting roster pushes failed: ${result.toXml()}');
@@ -300,33 +296,31 @@ class RosterManager extends XmppManagerBase {
     List<String>? groups,
   }) async {
     final attrs = getAttributes();
-    final response = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'query',
-              xmlns: rosterXmlns,
-              children: [
-                XMLNode(
-                  tag: 'item',
-                  attributes: <String, String>{
-                    'jid': jid,
-                    ...title == jid.split('@')[0]
-                        ? <String, String>{}
-                        : <String, String>{'name': title}
-                  },
-                  children: (groups ?? [])
-                      .map((group) => XMLNode(tag: 'group', text: group))
-                      .toList(),
-                )
-              ],
-            ),
-          ],
-        ),
+    final response = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'query',
+            xmlns: rosterXmlns,
+            children: [
+              XMLNode(
+                tag: 'item',
+                attributes: <String, String>{
+                  'jid': jid,
+                  ...title == jid.split('@')[0]
+                      ? <String, String>{}
+                      : <String, String>{'name': title}
+                },
+                children: (groups ?? [])
+                    .map((group) => XMLNode(tag: 'group', text: group))
+                    .toList(),
+              )
+            ],
+          )
+        ],
       ),
-    ))!;
+    );
 
     if (response.attributes['type'] != 'result') {
       logger.severe('Error adding $jid to roster: $response');
@@ -340,28 +334,26 @@ class RosterManager extends XmppManagerBase {
   /// false otherwise.
   Future<RosterRemovalResult> removeFromRoster(String jid) async {
     final attrs = getAttributes();
-    final response = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'query',
-              xmlns: rosterXmlns,
-              children: [
-                XMLNode(
-                  tag: 'item',
-                  attributes: <String, String>{
-                    'jid': jid,
-                    'subscription': 'remove'
-                  },
-                )
-              ],
-            )
-          ],
-        ),
+    final response = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'query',
+            xmlns: rosterXmlns,
+            children: [
+              XMLNode(
+                tag: 'item',
+                attributes: <String, String>{
+                  'jid': jid,
+                  'subscription': 'remove'
+                },
+              )
+            ],
+          )
+        ],
       ),
-    ))!;
+    );
 
     if (response.attributes['type'] != 'result') {
       logger.severe('Failed to remove roster item: ${response.toXml()}');

packages/moxxmpp/lib/src/stanza.dart

@@ -1,30 +1,6 @@
 import 'package:moxxmpp/src/namespaces.dart';
 import 'package:moxxmpp/src/stringxml.dart';
 
-/// A description of a stanza to send.
-class StanzaDetails {
-  const StanzaDetails(
-    this.stanza, {
-    this.addId = true,
-    this.awaitable = true,
-    this.encrypted = false,
-    this.forceEncryption = false,
-  });
-
-  /// The stanza to send.
-  final Stanza stanza;
-
-  /// Flag indicating whether a stanza id should be added before sending.
-  final bool addId;
-
-  /// Track the stanza to allow awaiting its response.
-  final bool awaitable;
-
-  final bool encrypted;
-
-  final bool forceEncryption;
-}
-
 /// A simple description of the <error /> element that may be inside a stanza
 class StanzaError {
   StanzaError(this.type, this.error);

packages/moxxmpp/lib/src/util/queue.dart

@@ -1,68 +1,37 @@
 import 'dart:async';
 import 'dart:collection';
 import 'package:meta/meta.dart';
-import 'package:moxxmpp/src/stanza.dart';
-import 'package:moxxmpp/src/stringxml.dart';
 import 'package:synchronized/synchronized.dart';
 
-class StanzaQueueEntry {
-  const StanzaQueueEntry(
-    this.details,
-    this.completer,
-  );
-
-  /// The actual data to send.
-  final StanzaDetails details;
-
-  /// The [Completer] to resolve when the response is received.
-  final Completer<XMLNode>? completer;
-}
-
-/// A function that is executed when a job is popped from the queue.
-typedef SendStanzaFunction = Future<void> Function(StanzaQueueEntry);
-
-/// A function that is called before popping a queue item. Should return true when
-/// the [SendStanzaFunction] can be executed.
-typedef CanSendCallback = Future<bool> Function();
+/// 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 AsyncStanzaQueue {
-  AsyncStanzaQueue(
-    this._sendStanzaFunction,
-    this._canSendCallback,
-  );
-
-  /// The lock for accessing [AsyncStanzaQueue._lock] and [AsyncStanzaQueue._running].
+class AsyncQueue {
+  /// The lock for accessing [AsyncQueue._lock] and [AsyncQueue._running].
   final Lock _lock = Lock();
 
   /// The actual job queue.
-  final Queue<StanzaQueueEntry> _queue = Queue<StanzaQueueEntry>();
-
-  /// Sends the stanza when we can pop from the queue.
-  final SendStanzaFunction _sendStanzaFunction;
-
-  final CanSendCallback _canSendCallback;
+  final Queue<AsyncQueueJob> _queue = Queue<AsyncQueueJob>();
 
   /// Indicates whether we are currently executing a job.
   bool _running = false;
 
   @visibleForTesting
-  Queue<StanzaQueueEntry> get queue => _queue;
+  Queue<AsyncQueueJob> get queue => _queue;
 
   @visibleForTesting
   bool get isRunning => _running;
 
-  /// Adds a job [entry] to the queue.
-  Future<void> enqueueStanza(StanzaQueueEntry entry) async {
-    await _lock.synchronized(() async {
-      _queue.add(entry);
+  /// Adds a job [job] to the queue.
+  Future<void> addJob(AsyncQueueJob job) async {
+    await _lock.synchronized(() {
+      _queue.add(job);
 
-      if (!_running && _queue.isNotEmpty && await _canSendCallback()) {
+      if (!_running && _queue.isNotEmpty) {
         _running = true;
-        unawaited(
-          _runJob(_queue.removeFirst()),
-        );
+        unawaited(_popJob());
       }
     });
   }
@@ -71,30 +40,17 @@ class AsyncStanzaQueue {
     await _lock.synchronized(_queue.clear);
   }
 
-  Future<void> _runJob(StanzaQueueEntry details) async {
-    await _sendStanzaFunction(details);
+  Future<void> _popJob() async {
+    final job = _queue.removeFirst();
+    final future = job();
+    await future;
 
-    await _lock.synchronized(() async {
-      if (_queue.isNotEmpty && await _canSendCallback()) {
-        unawaited(
-          _runJob(_queue.removeFirst()),
-        );
+    await _lock.synchronized(() {
+      if (_queue.isNotEmpty) {
+        unawaited(_popJob());
       } else {
         _running = false;
       }
     });
   }
-
-  Future<void> restart() async {
-    if (!(await _canSendCallback())) return;
-
-    await _lock.synchronized(() {
-      if (_queue.isNotEmpty) {
-        _running = true;
-        unawaited(
-          _runJob(_queue.removeFirst()),
-        );
-      }
-    });
-  }
 }

packages/moxxmpp/lib/src/xeps/xep_0030/xep_0030.dart

@@ -291,12 +291,10 @@ class DiscoManager extends XmppManagerBase {
       }
     }
 
-    final stanza = (await getAttributes().sendStanza(
-      StanzaDetails(
-        buildDiscoInfoQueryStanza(entity, node),
-        encrypted: !shouldEncrypt,
-      ),
-    ))!;
+    final stanza = await getAttributes().sendStanza(
+      buildDiscoInfoQueryStanza(entity, node),
+      encrypted: !shouldEncrypt,
+    );
     final query = stanza.firstTag('query');
     if (query == null) {
       final result = Result<DiscoError, DiscoInfo>(InvalidResponseDiscoError());
@@ -333,12 +331,10 @@ class DiscoManager extends XmppManagerBase {
       return future;
     }
 
-    final stanza = (await getAttributes().sendStanza(
-      StanzaDetails(
-        buildDiscoItemsQueryStanza(entity, node: node),
-        encrypted: !shouldEncrypt,
-      ),
-    ))!;
+    final stanza = await getAttributes().sendStanza(
+      buildDiscoItemsQueryStanza(entity, node: node),
+      encrypted: !shouldEncrypt,
+    ) as Stanza;
 
     final query = stanza.firstTag('query');
     if (query == null) {
@@ -348,7 +344,7 @@ class DiscoManager extends XmppManagerBase {
       return result;
     }
 
-    if (stanza.attributes['type'] == 'error') {
+    if (stanza.type == 'error') {
       //final error = stanza.firstTag('error');
       //print("Disco Items error: " + error.toXml());
       final result =

packages/moxxmpp/lib/src/xeps/xep_0054.dart

@@ -103,21 +103,19 @@ class VCardManager extends XmppManagerBase {
   }
 
   Future<Result<VCardError, VCard>> requestVCard(String jid) async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          to: jid,
-          type: 'get',
-          children: [
-            XMLNode.xmlns(
-              tag: 'vCard',
-              xmlns: vCardTempXmlns,
-            )
-          ],
-        ),
-        encrypted: true,
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        to: jid,
+        type: 'get',
+        children: [
+          XMLNode.xmlns(
+            tag: 'vCard',
+            xmlns: vCardTempXmlns,
+          )
+        ],
       ),
-    ))!;
+      encrypted: true,
+    );
 
     if (result.attributes['type'] != 'result') {
       return Result(UnknownVCardError());

packages/moxxmpp/lib/src/xeps/xep_0060/xep_0060.dart

@@ -181,29 +181,27 @@ class PubSubManager extends XmppManagerBase {
 
   Future<Result<PubSubError, bool>> subscribe(String jid, String node) async {
     final attrs = getAttributes();
-    final result = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: jid,
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'subscribe',
-                  attributes: <String, String>{
-                    'node': node,
-                    'jid': attrs.getFullJID().toBare().toString(),
-                  },
-                ),
-              ],
-            ),
-          ],
-        ),
+    final result = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: jid,
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(
+                tag: 'subscribe',
+                attributes: <String, String>{
+                  'node': node,
+                  'jid': attrs.getFullJID().toBare().toString(),
+                },
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    );
 
     if (result.attributes['type'] != 'result') {
       return Result(UnknownPubSubError());
@@ -224,29 +222,27 @@ class PubSubManager extends XmppManagerBase {
 
   Future<Result<PubSubError, bool>> unsubscribe(String jid, String node) async {
     final attrs = getAttributes();
-    final result = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: jid,
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'unsubscribe',
-                  attributes: <String, String>{
-                    'node': node,
-                    'jid': attrs.getFullJID().toBare().toString(),
-                  },
-                ),
-              ],
-            ),
-          ],
-        ),
+    final result = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: jid,
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(
+                tag: 'unsubscribe',
+                attributes: <String, String>{
+                  'node': node,
+                  'jid': attrs.getFullJID().toBare().toString(),
+                },
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    );
 
     if (result.attributes['type'] != 'result') {
       return Result(UnknownPubSubError());
@@ -297,40 +293,38 @@ class PubSubManager extends XmppManagerBase {
       pubOptions = await preprocessPublishOptions(jid, node, options);
     }
 
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: jid.toString(),
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'publish',
-                  attributes: <String, String>{'node': node},
-                  children: [
-                    XMLNode(
-                      tag: 'item',
-                      attributes: id != null
-                          ? <String, String>{'id': id}
-                          : <String, String>{},
-                      children: [payload],
-                    )
-                  ],
-                ),
-                if (pubOptions != null)
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: jid.toString(),
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(
+                tag: 'publish',
+                attributes: <String, String>{'node': node},
+                children: [
                   XMLNode(
-                    tag: 'publish-options',
-                    children: [pubOptions.toXml()],
-                  ),
-              ],
-            )
-          ],
-        ),
+                    tag: 'item',
+                    attributes: id != null
+                        ? <String, String>{'id': id}
+                        : <String, String>{},
+                    children: [payload],
+                  )
+                ],
+              ),
+              if (pubOptions != null)
+                XMLNode(
+                  tag: 'publish-options',
+                  children: [pubOptions.toXml()],
+                ),
+            ],
+          )
+        ],
       ),
-    ))!;
+    );
     if (result.attributes['type'] != 'result') {
       final error = getPubSubError(result);
 
@@ -401,26 +395,21 @@ class PubSubManager extends XmppManagerBase {
     String jid,
     String node,
   ) async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          to: jid,
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'items',
-                  attributes: <String, String>{'node': node},
-                ),
-              ],
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'get',
+        to: jid,
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(tag: 'items', attributes: <String, String>{'node': node}),
+            ],
+          )
+        ],
       ),
-    ))!;
+    );
 
     if (result.attributes['type'] != 'result') {
       return Result(getPubSubError(result));
@@ -447,32 +436,30 @@ class PubSubManager extends XmppManagerBase {
     String node,
     String id,
   ) async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          to: jid,
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'items',
-                  attributes: <String, String>{'node': node},
-                  children: [
-                    XMLNode(
-                      tag: 'item',
-                      attributes: <String, String>{'id': id},
-                    ),
-                  ],
-                ),
-              ],
-            ),
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'get',
+        to: jid,
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(
+                tag: 'items',
+                attributes: <String, String>{'node': node},
+                children: [
+                  XMLNode(
+                    tag: 'item',
+                    attributes: <String, String>{'id': id},
+                  ),
+                ],
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    );
 
     if (result.attributes['type'] != 'result') {
       return Result(getPubSubError(result));
@@ -501,57 +488,53 @@ class PubSubManager extends XmppManagerBase {
     final attrs = getAttributes();
 
     // Request the form
-    final form = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          to: jid.toString(),
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubOwnerXmlns,
-              children: [
-                XMLNode(
-                  tag: 'configure',
-                  attributes: <String, String>{
-                    'node': node,
-                  },
-                ),
-              ],
-            ),
-          ],
-        ),
+    final form = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'get',
+        to: jid.toString(),
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubOwnerXmlns,
+            children: [
+              XMLNode(
+                tag: 'configure',
+                attributes: <String, String>{
+                  'node': node,
+                },
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    );
     if (form.attributes['type'] != 'result') {
       return Result(getPubSubError(form));
     }
 
-    final submit = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: jid.toString(),
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubOwnerXmlns,
-              children: [
-                XMLNode(
-                  tag: 'configure',
-                  attributes: <String, String>{
-                    'node': node,
-                  },
-                  children: [
-                    options.toXml(),
-                  ],
-                ),
-              ],
-            ),
-          ],
-        ),
+    final submit = await attrs.sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: jid.toString(),
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubOwnerXmlns,
+            children: [
+              XMLNode(
+                tag: 'configure',
+                attributes: <String, String>{
+                  'node': node,
+                },
+                children: [
+                  options.toXml(),
+                ],
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    );
     if (submit.attributes['type'] != 'result') {
       return Result(getPubSubError(form));
     }
@@ -560,30 +543,28 @@ class PubSubManager extends XmppManagerBase {
   }
 
   Future<Result<PubSubError, bool>> delete(JID host, String node) async {
-    final request = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: host.toString(),
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubOwnerXmlns,
-              children: [
-                XMLNode(
-                  tag: 'delete',
-                  attributes: <String, String>{
-                    'node': node,
-                  },
-                ),
-              ],
-            ),
-          ],
-        ),
+    final request = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: host.toString(),
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubOwnerXmlns,
+            children: [
+              XMLNode(
+                tag: 'delete',
+                attributes: <String, String>{
+                  'node': node,
+                },
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    ) as Stanza;
 
-    if (request.attributes['type'] != 'result') {
+    if (request.type != 'result') {
       // TODO(Unknown): Be more specific
       return Result(UnknownPubSubError());
     }
@@ -596,38 +577,36 @@ class PubSubManager extends XmppManagerBase {
     String node,
     String itemId,
   ) async {
-    final request = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          to: host.toString(),
-          children: [
-            XMLNode.xmlns(
-              tag: 'pubsub',
-              xmlns: pubsubXmlns,
-              children: [
-                XMLNode(
-                  tag: 'retract',
-                  attributes: <String, String>{
-                    'node': node,
-                  },
-                  children: [
-                    XMLNode(
-                      tag: 'item',
-                      attributes: <String, String>{
-                        'id': itemId,
-                      },
-                    ),
-                  ],
-                ),
-              ],
-            ),
-          ],
-        ),
+    final request = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        to: host.toString(),
+        children: [
+          XMLNode.xmlns(
+            tag: 'pubsub',
+            xmlns: pubsubXmlns,
+            children: [
+              XMLNode(
+                tag: 'retract',
+                attributes: <String, String>{
+                  'node': node,
+                },
+                children: [
+                  XMLNode(
+                    tag: 'item',
+                    attributes: <String, String>{
+                      'id': itemId,
+                    },
+                  ),
+                ],
+              ),
+            ],
+          ),
+        ],
       ),
-    ))!;
+    ) as Stanza;
 
-    if (request.attributes['type'] != 'result') {
+    if (request.type != 'result') {
       // TODO(Unknown): Be more specific
       return Result(UnknownPubSubError());
     }

packages/moxxmpp/lib/src/xeps/xep_0085.dart

@@ -111,14 +111,10 @@ class ChatStateManager extends XmppManagerBase {
     final tagName = state.toString().split('.').last;
 
     getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.message(
-          to: to,
-          type: messageType,
-          children: [
-            XMLNode.xmlns(tag: tagName, xmlns: chatStateXmlns),
-          ],
-        ),
+      Stanza.message(
+        to: to,
+        type: messageType,
+        children: [XMLNode.xmlns(tag: tagName, xmlns: chatStateXmlns)],
       ),
     );
   }

packages/moxxmpp/lib/src/xeps/xep_0191.dart

@@ -96,43 +96,39 @@ class BlockingManager extends XmppManagerBase {
   }
 
   Future<bool> block(List<String> items) async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'block',
-              xmlns: blockingXmlns,
-              children: items.map((item) {
-                return XMLNode(
-                  tag: 'item',
-                  attributes: <String, String>{'jid': item},
-                );
-              }).toList(),
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'block',
+            xmlns: blockingXmlns,
+            children: items.map((item) {
+              return XMLNode(
+                tag: 'item',
+                attributes: <String, String>{'jid': item},
+              );
+            }).toList(),
+          )
+        ],
       ),
-    ))!;
+    );
 
     return result.attributes['type'] == 'result';
   }
 
   Future<bool> unblockAll() async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'unblock',
-              xmlns: blockingXmlns,
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'unblock',
+            xmlns: blockingXmlns,
+          )
+        ],
       ),
-    ))!;
+    );
 
     return result.attributes['type'] == 'result';
   }
@@ -140,45 +136,41 @@ class BlockingManager extends XmppManagerBase {
   Future<bool> unblock(List<String> items) async {
     assert(items.isNotEmpty, 'The list of items to unblock must be non-empty');
 
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'unblock',
-              xmlns: blockingXmlns,
-              children: items
-                  .map(
-                    (item) => XMLNode(
-                      tag: 'item',
-                      attributes: <String, String>{'jid': item},
-                    ),
-                  )
-                  .toList(),
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'unblock',
+            xmlns: blockingXmlns,
+            children: items
+                .map(
+                  (item) => XMLNode(
+                    tag: 'item',
+                    attributes: <String, String>{'jid': item},
+                  ),
+                )
+                .toList(),
+          )
+        ],
       ),
-    ))!;
+    );
 
     return result.attributes['type'] == 'result';
   }
 
   Future<List<String>> getBlocklist() async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'get',
-          children: [
-            XMLNode.xmlns(
-              tag: 'blocklist',
-              xmlns: blockingXmlns,
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'get',
+        children: [
+          XMLNode.xmlns(
+            tag: 'blocklist',
+            xmlns: blockingXmlns,
+          )
+        ],
       ),
-    ))!;
+    );
 
     final blocklist = result.firstTag('blocklist', xmlns: blockingXmlns)!;
     return blocklist

packages/moxxmpp/lib/src/xeps/xep_0198/xep_0198.dart

@@ -414,12 +414,7 @@ class StreamManagementManager extends XmppManagerBase {
     _unackedStanzas.clear();
 
     for (final stanza in stanzas) {
-      await getAttributes().sendStanza(
-        StanzaDetails(
-          stanza,
-          awaitable: false,
-        ),
-      );
+      await getAttributes().sendStanza(stanza, awaitable: false);
     }
   }
 

packages/moxxmpp/lib/src/xeps/xep_0280.dart

@@ -1,5 +1,6 @@
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
+import 'package:moxxmpp/src/connection.dart';
 import 'package:moxxmpp/src/events.dart';
 import 'package:moxxmpp/src/jid.dart';
 import 'package:moxxmpp/src/managers/base.dart';
@@ -110,20 +111,20 @@ class CarbonsManager extends XmppManagerBase {
   /// Returns true if carbons were enabled. False, if not.
   Future<bool> enableCarbons() async {
     final attrs = getAttributes();
-    final result = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          to: attrs.getFullJID().toBare().toString(),
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'enable',
-              xmlns: carbonsXmlns,
-            )
-          ],
-        ),
+    final result = await attrs.sendStanza(
+      Stanza.iq(
+        to: attrs.getFullJID().toBare().toString(),
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'enable',
+            xmlns: carbonsXmlns,
+          )
+        ],
       ),
-    ))!;
+      addFrom: StanzaFromType.full,
+      addId: true,
+    );
 
     if (result.attributes['type'] != 'result') {
       logger.warning('Failed to enable message carbons');
@@ -141,19 +142,19 @@ class CarbonsManager extends XmppManagerBase {
   ///
   /// Returns true if carbons were disabled. False, if not.
   Future<bool> disableCarbons() async {
-    final result = (await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          type: 'set',
-          children: [
-            XMLNode.xmlns(
-              tag: 'disable',
-              xmlns: carbonsXmlns,
-            )
-          ],
-        ),
+    final result = await getAttributes().sendStanza(
+      Stanza.iq(
+        type: 'set',
+        children: [
+          XMLNode.xmlns(
+            tag: 'disable',
+            xmlns: carbonsXmlns,
+          )
+        ],
       ),
-    ))!;
+      addFrom: StanzaFromType.full,
+      addId: true,
+    );
 
     if (result.attributes['type'] != 'result') {
       logger.warning('Failed to disable message carbons');

packages/moxxmpp/lib/src/xeps/xep_0363/xep_0363.dart

@@ -149,25 +149,23 @@ class HttpFileUploadManager extends XmppManagerBase {
     }
 
     final attrs = getAttributes();
-    final response = (await attrs.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          to: _entityJid.toString(),
-          type: 'get',
-          children: [
-            XMLNode.xmlns(
-              tag: 'request',
-              xmlns: httpFileUploadXmlns,
-              attributes: {
-                'filename': filename,
-                'size': filesize.toString(),
-                ...contentType != null ? {'content-type': contentType} : {}
-              },
-            )
-          ],
-        ),
+    final response = await attrs.sendStanza(
+      Stanza.iq(
+        to: _entityJid.toString(),
+        type: 'get',
+        children: [
+          XMLNode.xmlns(
+            tag: 'request',
+            xmlns: httpFileUploadXmlns,
+            attributes: {
+              'filename': filename,
+              'size': filesize.toString(),
+              ...contentType != null ? {'content-type': contentType} : {}
+            },
+          )
+        ],
       ),
-    ))!;
+    );
 
     if (response.attributes['type']! != 'result') {
       logger.severe('Failed to request HTTP File Upload slot.');

packages/moxxmpp/lib/src/xeps/xep_0384/xep_0384.dart

@@ -262,26 +262,24 @@ abstract class BaseOmemoManager extends XmppManagerBase {
     String toJid,
   ) async {
     await getAttributes().sendStanza(
-      StanzaDetails(
-        Stanza.message(
-          to: toJid,
-          type: 'chat',
-          children: [
-            _buildEncryptedElement(
-              result,
-              toJid,
-              await _getDeviceId(),
-            ),
+      Stanza.message(
+        to: toJid,
+        type: 'chat',
+        children: [
+          _buildEncryptedElement(
+            result,
+            toJid,
+            await _getDeviceId(),
+          ),
 
-            // Add a storage hint in case this is a message
-            // Taken from the example at
-            // https://xmpp.org/extensions/xep-0384.html#message-structure-description.
-            MessageProcessingHint.store.toXml(),
-          ],
-        ),
-        awaitable: false,
-        encrypted: true,
+          // Add a storage hint in case this is a message
+          // Taken from the example at
+          // https://xmpp.org/extensions/xep-0384.html#message-structure-description.
+          MessageProcessingHint.store.toXml(),
+        ],
       ),
+      awaitable: false,
+      encrypted: true,
     );
   }
 

packages/moxxmpp/test/async_queue_test.dart

@@ -1,100 +1,58 @@
-import 'package:moxxmpp/moxxmpp.dart';
 import 'package:moxxmpp/src/util/queue.dart';
 import 'package:test/test.dart';
 
 void main() {
-  test('Test not sending', () async {
-    final queue = AsyncStanzaQueue(
-      (entry) async {
-        assert(false, 'No stanza should be sent');
-      },
-      () async => false,
-    );
+  test('Test the async queue', () async {
+    final queue = AsyncQueue();
+    var future1Finish = 0;
+    var future2Finish = 0;
+    var future3Finish = 0;
 
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
+    await queue.addJob(
+      () => Future<void>.delayed(
+        const Duration(seconds: 3),
+        () => future1Finish = DateTime.now().millisecondsSinceEpoch,
       ),
     );
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
+    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: 1));
-    expect(queue.queue.length, 2);
-    expect(queue.isRunning, false);
-  });
+    await Future<void>.delayed(const Duration(seconds: 12));
 
-  test('Test sending', () async {
-    final queue = AsyncStanzaQueue(
-      (entry) async {},
-      () async => true,
+    // 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,
     );
 
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
-      ),
-    );
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
-      ),
-    );
+    // The queue must be empty at the end
+    expect(queue.queue.isEmpty, true);
 
-    await Future<void>.delayed(const Duration(seconds: 1));
-    expect(queue.queue.length, 0);
-    expect(queue.isRunning, false);
-  });
-
-  test('Test partial sending and resuming', () async {
-    var canRun = true;
-    final queue = AsyncStanzaQueue(
-      (entry) async {
-        canRun = false;
-      },
-      () async => canRun,
-    );
-
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
-      ),
-    );
-    await queue.enqueueStanza(
-      StanzaQueueEntry(
-        StanzaDetails(
-          Stanza.message(),
-        ),
-        null,
-      ),
-    );
-
-    await Future<void>.delayed(const Duration(seconds: 1));
-    expect(queue.queue.length, 1);
-    expect(queue.isRunning, false);
-
-    canRun = true;
-    await queue.restart();
-    await Future<void>.delayed(const Duration(seconds: 1));
-    expect(queue.queue.length, 0);
+    // The queue must not be executing anything at the end
     expect(queue.isRunning, false);
   });
 }

packages/moxxmpp/test/helpers/manager.dart

@@ -33,6 +33,7 @@ class TestingManagerHolder {
 
   Future<XMLNode> _sendStanza(
     stanza, {
+    StanzaFromType addFrom = StanzaFromType.full,
     bool addId = true,
     bool awaitable = true,
     bool encrypted = false,

packages/moxxmpp/test/helpers/xmpp.dart

@@ -98,7 +98,7 @@ List<ExpectationBase> buildAuthenticatedPlay(ConnectionSettings settings) {
       ignoreId: true,
     ),
     StanzaExpectation(
-      "<presence xmlns='jabber:client'><show>chat</show></presence>",
+      "<presence xmlns='jabber:client' from='${settings.jid.toBare()}/MU29eEZn'><show>chat</show></presence>",
       '',
     ),
   ];

packages/moxxmpp/test/xeps/xep_0030_test.dart

@@ -58,11 +58,11 @@ void main() {
           ignoreId: true,
         ),
         StanzaExpectation(
-          "<presence xmlns='jabber:client'><show>chat</show><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://moxxmpp.example' ver='3QvQ2RAy45XBDhArjxy/vEWMl+E=' /></presence>",
+          "<presence xmlns='jabber:client' from='polynomdivision@test.server/MU29eEZn'><show>chat</show><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://moxxmpp.example' ver='3QvQ2RAy45XBDhArjxy/vEWMl+E=' /></presence>",
           '',
         ),
         StanzaExpectation(
-          "<iq type='get' id='ec325efc-9924-4c48-93f8-ed34a2b0e5fc' to='romeo@montague.lit/orchard' xmlns='jabber:client'><query xmlns='http://jabber.org/protocol/disco#info' /></iq>",
+          "<iq type='get' id='ec325efc-9924-4c48-93f8-ed34a2b0e5fc' to='romeo@montague.lit/orchard' from='polynomdivision@test.server/MU29eEZn' xmlns='jabber:client'><query xmlns='http://jabber.org/protocol/disco#info' /></iq>",
           '',
           ignoreId: true,
         ),

packages/moxxmpp/test/xeps/xep_0060_test.dart

@@ -92,7 +92,7 @@ void main() {
       [
         StanzaExpectation(
           '''
-<iq type="get" to="pubsub.server.example.org" id="a" xmlns="jabber:client">
+<iq type="get" to="pubsub.server.example.org" id="a" from="testuser@example.org/MU29eEZn" xmlns="jabber:client">
   <query xmlns="http://jabber.org/protocol/disco#info" />
 </iq>
 ''',
@@ -110,7 +110,7 @@ void main() {
         ),
         StanzaExpectation(
           '''
-<iq type="get" to="pubsub.server.example.org" id="a" xmlns="jabber:client">
+<iq type="get" to="pubsub.server.example.org" id="a" from="testuser@example.org/MU29eEZn" xmlns="jabber:client">
   <query xmlns="http://jabber.org/protocol/disco#items" node="princely_musings" />
 </iq>
 ''',
@@ -124,7 +124,7 @@ void main() {
         ),
         StanzaExpectation(
           '''
-<iq type="set" to="pubsub.server.example.org" id="a" xmlns="jabber:client">
+<iq type="set" to="pubsub.server.example.org" id="a" from="testuser@example.org/MU29eEZn" xmlns="jabber:client">
   <pubsub xmlns='http://jabber.org/protocol/pubsub'>
     <publish node='princely_musings'>
       <item id="current">

packages/moxxmpp/test/xeps/xep_0198_test.dart

@@ -44,8 +44,15 @@ Future<void> runOutgoingStanzaHandlers(
 
 XmppManagerAttributes mkAttributes(void Function(Stanza) callback) {
   return XmppManagerAttributes(
-    sendStanza: (StanzaDetails details) async {
-      callback(details.stanza);
+    sendStanza: (
+      stanza, {
+      StanzaFromType addFrom = StanzaFromType.full,
+      bool addId = true,
+      bool awaitable = true,
+      bool encrypted = false,
+      bool forceEncryption = false,
+    }) async {
+      callback(stanza);
 
       return Stanza.message();
     },
@@ -283,8 +290,12 @@ void main() {
         );
       final sm = StreamManagementManager();
       await conn.registerManagers([
+        PresenceManager(),
+        RosterManager(TestingRosterStateManager('', [])),
+        DiscoManager([]),
         sm,
         CarbonsManager()..forceEnable(),
+        EntityCapabilitiesManager('http://moxxmpp.example'),
       ]);
       await conn.registerFeatureNegotiators([
         SaslPlainNegotiator(),
@@ -380,7 +391,7 @@ void main() {
           '<enabled xmlns="urn:xmpp:sm:3" id="some-long-sm-id" resume="true" />',
         ),
         StanzaExpectation(
-          "<presence xmlns='jabber:client'><show>chat</show></presence>",
+          "<presence xmlns='jabber:client' from='polynomdivision@test.server/MU29eEZn'><show>chat</show></presence>",
           '<iq type="result" />',
         ),
         StringExpectation(
@@ -670,7 +681,7 @@ void main() {
         "<resumed xmlns='urn:xmpp:sm:3' h='id-1' h='12' />",
       ),
       StanzaExpectation(
-        "<iq to='localhost' type='get' xmlns='jabber:client' />",
+        "<iq to='localhost' type='get' from='polynomdivision@test.server/abc123' xmlns='jabber:client' />",
         '',
         ignoreId: true,
       ),
@@ -723,7 +734,7 @@ void main() {
         "<resumed xmlns='urn:xmpp:sm:3' h='id-1' h='12' />",
       ),
       StanzaExpectation(
-        "<iq to='localhost' type='get' xmlns='jabber:client' />",
+        "<iq to='localhost' type='get' from='polynomdivision@test.server/abc123' xmlns='jabber:client' />",
         '',
         ignoreId: true,
       ),
@@ -765,11 +776,7 @@ void main() {
 
     // Send a bogus stanza
     unawaited(
-      conn.sendStanza(
-        StanzaDetails(
-          Stanza.iq(to: 'localhost', type: 'get'),
-        ),
-      ),
+      conn.sendStanza(Stanza.iq(to: 'localhost', type: 'get')),
     );
 
     await Future<void>.delayed(const Duration(seconds: 5));

packages/moxxmpp/test/xeps/xep_0280_test.dart

@@ -9,9 +9,17 @@ void main() {
   test("Test if we're vulnerable against CVE-2020-26547 style vulnerabilities",
       () async {
     final attributes = XmppManagerAttributes(
-      sendStanza: (StanzaDetails details) async {
+      sendStanza: (
+        stanza, {
+        StanzaFromType addFrom = StanzaFromType.full,
+        bool addId = true,
+        bool retransmitted = false,
+        bool awaitable = true,
+        bool encrypted = false,
+        bool forceEncryption = false,
+      }) async {
         // ignore: avoid_print
-        print('==> ${details.stanza.toXml()}');
+        print('==> ${stanza.toXml()}');
         return XMLNode(tag: 'iq', attributes: {'type': 'result'});
       },
       sendNonza: (nonza) {},

packages/moxxmpp/test/xeps/xep_0352_test.dart

@@ -39,6 +39,7 @@ void main() {
           XmppManagerAttributes(
             sendStanza: (
               _, {
+              StanzaFromType addFrom = StanzaFromType.full,
               bool addId = true,
               bool retransmitted = false,
               bool awaitable = true,
@@ -77,6 +78,7 @@ void main() {
           XmppManagerAttributes(
             sendStanza: (
               _, {
+              StanzaFromType addFrom = StanzaFromType.full,
               bool addId = true,
               bool retransmitted = false,
               bool awaitable = true,

packages/moxxmpp/test/xmpp_test.dart

@@ -4,32 +4,6 @@ import 'package:test/test.dart';
 import 'helpers/logging.dart';
 import 'helpers/xmpp.dart';
 
-class StubConnectivityManager extends ConnectivityManager {
-  bool _hasConnection = true;
-
-  Completer<void> _goingOnlineCompleter = Completer<void>();
-
-  @override
-  Future<bool> hasConnection() async => _hasConnection;
-
-  @override
-  Future<void> waitForConnection() async {
-    if (!_hasConnection) {
-      await _goingOnlineCompleter.future;
-    }
-  }
-
-  void goOffline() {
-    _hasConnection = false;
-  }
-
-  void goOnline() {
-    _hasConnection = true;
-    _goingOnlineCompleter.complete();
-    _goingOnlineCompleter = Completer<void>();
-  }
-}
-
 /// Returns true if the roster manager triggeres an event for a given stanza
 Future<bool> testRosterManager(
   String bareJid,
@@ -42,6 +16,7 @@ Future<bool> testRosterManager(
       XmppManagerAttributes(
         sendStanza: (
           _, {
+          StanzaFromType addFrom = StanzaFromType.full,
           bool addId = true,
           bool retransmitted = false,
           bool awaitable = true,
@@ -156,7 +131,11 @@ void main() {
         password: 'aaaa',
       );
     await conn.registerManagers([
+      PresenceManager(),
+      RosterManager(TestingRosterStateManager('', [])),
+      DiscoManager([]),
       StreamManagementManager(),
+      EntityCapabilitiesManager('http://moxxmpp.example'),
     ]);
     await conn.registerFeatureNegotiators([
       SaslPlainNegotiator(),
@@ -292,6 +271,7 @@ void main() {
           XmppManagerAttributes(
             sendStanza: (
               _, {
+              StanzaFromType addFrom = StanzaFromType.full,
               bool addId = true,
               bool retransmitted = false,
               bool awaitable = true,
@@ -645,152 +625,4 @@ void main() {
       true,
     );
   });
-
-  test('Test sending stanzas while offline', () async {
-    final fakeSocket = StubTCPSocket(
-      [
-        StringExpectation(
-          "<stream:stream xmlns='jabber:client' version='1.0' xmlns:stream='http://etherx.jabber.org/streams' to='test.server' from='polynomdivision@test.server' xml:lang='en'>",
-          '''
-<stream:stream
-    xmlns="jabber:client"
-    version="1.0"
-    xmlns:stream="http://etherx.jabber.org/streams"
-    from="test.server"
-    xml:lang="en">
-  <stream:features xmlns="http://etherx.jabber.org/streams">
-    <mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl">
-      <mechanism>PLAIN</mechanism>
-    </mechanisms>
-  </stream:features>''',
-        ),
-        StringExpectation(
-          "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>AHBvbHlub21kaXZpc2lvbgBhYWFh</auth>",
-          '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl" />',
-        ),
-        StringExpectation(
-          "<stream:stream xmlns='jabber:client' version='1.0' xmlns:stream='http://etherx.jabber.org/streams' to='test.server' from='polynomdivision@test.server' xml:lang='en'>",
-          '''
-<stream:stream
-    xmlns="jabber:client"
-    version="1.0"
-    xmlns:stream="http://etherx.jabber.org/streams"
-    from="test.server"
-    xml:lang="en">
-  <stream:features xmlns="http://etherx.jabber.org/streams">
-    <bind xmlns="urn:ietf:params:xml:ns:xmpp-bind">
-      <required/>
-    </bind>
-    <session xmlns="urn:ietf:params:xml:ns:xmpp-session">
-      <optional/>
-    </session>
-    <csi xmlns="urn:xmpp:csi:0"/>
-    <sm xmlns="urn:xmpp:sm:3"/>
-  </stream:features>
-''',
-        ),
-        StanzaExpectation(
-          '<iq xmlns="jabber:client" type="set" id="a"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"/></iq>',
-          '<iq xmlns="jabber:client" type="result" id="a"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><jid>polynomdivision@test.server/MU29eEZn</jid></bind></iq>',
-          ignoreId: true,
-        ),
-        StringExpectation(
-          "<stream:stream xmlns='jabber:client' version='1.0' xmlns:stream='http://etherx.jabber.org/streams' to='test.server' from='polynomdivision@test.server' xml:lang='en'>",
-          '''
-<stream:stream
-    xmlns="jabber:client"
-    version="1.0"
-    xmlns:stream="http://etherx.jabber.org/streams"
-    from="test.server"
-    xml:lang="en">
-  <stream:features xmlns="http://etherx.jabber.org/streams">
-    <mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl">
-      <mechanism>PLAIN</mechanism>
-    </mechanisms>
-  </stream:features>''',
-        ),
-        StringExpectation(
-          "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>AHBvbHlub21kaXZpc2lvbgBhYWFh</auth>",
-          '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl" />',
-        ),
-        StringExpectation(
-          "<stream:stream xmlns='jabber:client' version='1.0' xmlns:stream='http://etherx.jabber.org/streams' to='test.server' from='polynomdivision@test.server' xml:lang='en'>",
-          '''
-<stream:stream
-    xmlns="jabber:client"
-    version="1.0"
-    xmlns:stream="http://etherx.jabber.org/streams"
-    from="test.server"
-    xml:lang="en">
-  <stream:features xmlns="http://etherx.jabber.org/streams">
-    <bind xmlns="urn:ietf:params:xml:ns:xmpp-bind">
-      <required/>
-    </bind>
-    <session xmlns="urn:ietf:params:xml:ns:xmpp-session">
-      <optional/>
-    </session>
-    <csi xmlns="urn:xmpp:csi:0"/>
-    <sm xmlns="urn:xmpp:sm:3"/>
-  </stream:features>
-''',
-        ),
-        StanzaExpectation(
-          '<iq xmlns="jabber:client" type="set" id="a"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"/></iq>',
-          '<iq xmlns="jabber:client" type="result" id="a"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><jid>polynomdivision@test.server/MU29eEZn</jid></bind></iq>',
-          ignoreId: true,
-        ),
-        StanzaExpectation(
-          '<iq xmlns="jabber:client" type="get" id="abc123"></iq>',
-          '<iq xmlns="jabber:client" type="result" id="abc123"></iq>',
-          ignoreId: true,
-        ),
-      ],
-    );
-    final connectivity = StubConnectivityManager();
-    final conn = XmppConnection(
-      TestingReconnectionPolicy(),
-      connectivity,
-      ClientToServerNegotiator(),
-      fakeSocket,
-    )..connectionSettings = ConnectionSettings(
-        jid: JID.fromString('polynomdivision@test.server'),
-        password: 'aaaa',
-      );
-    await conn.registerFeatureNegotiators([
-      SaslPlainNegotiator(),
-      SaslScramNegotiator(10, '', '', ScramHashType.sha512),
-      ResourceBindingNegotiator(),
-    ]);
-
-    await conn.connect(
-      waitUntilLogin: true,
-    );
-    expect(fakeSocket.getState(), 4);
-
-    // Fake going offline
-    connectivity.goOffline();
-    await conn.handleSocketEvent(
-      XmppSocketClosureEvent(false),
-    );
-
-    // Send a stanza while offline
-    final stanzaFuture = conn.sendStanza(
-      StanzaDetails(
-        Stanza.iq(
-          id: 'abc123',
-          type: 'get',
-        ),
-      ),
-    );
-
-    // Come online again
-    connectivity.goOnline();
-    await conn.connect(
-      waitUntilLogin: true,
-    );
-    await Future<void>.delayed(const Duration(seconds: 6));
-
-    expect(fakeSocket.getState(), 9);
-    expect(await stanzaFuture != null, true);
-  });
 }