xmpp: Communicate stream closure from the buffer

lib/xmpp/buffer.dart

@@ -20,9 +20,11 @@ class XmlStreamBuffer extends StreamTransformerBase<String, XMLNode> {
       toParse = toParse + "</stream:stream>";
     } else {
       if (toParse.endsWith("</stream:stream>")) {
-        // TODO: Maybe prepend <stream:stream> so that we can detect it within
-        //       [XmppConnection]
         toParse = toParse.substring(0, toParse.length - 16);
+        // In order for this class to have as little logic as possible, replace the
+        // </stream:stream> with a <stream:stream /> such that we can parse it without
+        // issue and catch it in [XmppConnection].
+        toParse += "<stream:stream />";
       }
     } 
 

lib/xmpp/connection.dart

@@ -93,7 +93,6 @@ class XmppConnection {
 
     _socketStream = _socket.getDataStream();
     // TODO: Handle on done
-    // TODO: Handle the stream buffer in the socket
     _socketStream.transform(_streamBuffer).forEach(handleXmlStream);
     _socket.getErrorStream().listen(_handleError);
 
@@ -351,6 +350,11 @@ class XmppConnection {
   void handleXmlStream(XMLNode node) async {
     _log("(xml) <== " + node.toXml());
 
+    if (node.tag == "stream:stream" && node.children.isEmpty) {
+      _handleError(null);
+      return;
+    }
+    
     switch (_routingState) {
       case RoutingState.unauthenticated: {
         // We expect the stream header here

test/xmlstreambuffer_test.dart

@@ -54,4 +54,37 @@ void main() {
       expect(childb, true);
     });
   });
+
+  test("Test closing the stream", () async {
+    bool childa = false;
+    bool childb = false;
+    bool closed = false;
+
+    final buffer = XmlStreamBuffer();
+    final controller = StreamController<String>();
+
+    controller
+    .stream
+    .transform(buffer)
+    .forEach((node) {
+      if (node.tag == "childa") {
+        childa = true;
+      } else if (node.tag == "childb") {
+        childb = true;
+      } else if (node.tag == "stream:stream" && node.children.isEmpty) {
+        assert (childa);
+        assert (childb);
+        closed = true;
+      }
+    });
+    controller.add("<childa");
+    controller.add(" /><childb />");
+    controller.add("</stream:stream>");
+
+    await Future.delayed(const Duration(seconds: 2), () {
+      expect(childa, true);
+      expect(childb, true);
+      expect(closed, true);
+    });
+  });
 }