feat(core): Add a callback for raw data events

packages/moxxmpp/lib/src/connection.dart

@@ -71,7 +71,11 @@ class XmppConnection {
     this._socket, {
     this.connectingTimeout = const Duration(minutes: 2),
   })  : _reconnectionPolicy = reconnectionPolicy,
-        _connectivityManager = connectivityManager {
+        _connectivityManager = connectivityManager,
+        assert(
+          _socket.getDataStream().isBroadcast,
+          "The socket's data stream must be a broadcast stream",
+        ) {
     // Allow the reconnection policy to perform reconnections by itself
     _reconnectionPolicy.register(
       _attemptReconnection,
@@ -95,10 +99,10 @@ class XmppConnection {
     );
     _incomingStanzaQueue = IncomingStanzaQueue(handleXmlStream, _stanzaAwaiter);
     _socketStream = _socket.getDataStream();
-    // TODO(Unknown): Handle on done
     _socketStream
         .transform(_streamParser)
         .forEach(_incomingStanzaQueue.addStanza);
+    _socketStream.listen(_handleOnDataCallbacks);
     _socket.getEventStream().listen(handleSocketEvent);
 
     _stanzaQueue = AsyncStanzaQueue(
@@ -314,6 +318,13 @@ class XmppConnection {
     return getManagerById(csiManager);
   }
 
+  /// Called whenever we receive data on the socket.
+  Future<void> _handleOnDataCallbacks(String _) async {
+    for (final manager in _xmppManagers.values) {
+      unawaited(manager.onData());
+    }
+  }
+
   /// Attempts to reconnect to the server by following an exponential backoff.
   Future<void> _attemptReconnection() async {
     _log.finest('_attemptReconnection: Setting state to notConnected');

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

@@ -80,6 +80,9 @@ abstract class XmppManagerBase {
   /// handler's priority, the earlier it is run.
   List<NonzaHandler> getNonzaHandlers() => [];
 
+  /// Whenever the socket receives data, this method is called, if it is non-null.
+  Future<void> onData() async {}
+
   /// Return a list of features that should be included in a disco response.
   List<String> getDiscoFeatures() => [];
 

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

@@ -75,6 +75,11 @@ class StreamManagementManager extends XmppManagerBase {
     return acks;
   }
 
+  @override
+  Future<void> onData() async {
+    logger.finest('Got data!');
+  }
+
   /// Called when a stanza has been acked to decide whether we should trigger a
   /// StanzaAckedEvent.
   ///