Merge pull request #17 from xconnio/remove-stream-controller

Remove stream controller & close stream subscription
xconnio · May 10, 2024 · 77043dd · 77043dd
2 parents 01f8141 + b66734f
commit 77043dd
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 21 deletions.
diff --git a/lib/src/acceptor.dart b/lib/src/acceptor.dart
@@ -22,28 +22,22 @@ class WAMPSessionAcceptor {
 
     Completer<BaseSession> completer = Completer<BaseSession>();
 
+    // ignore: cancel_subscriptions
     late StreamSubscription<dynamic> wsStreamSubscription;
-    final sessionStreamController = StreamController.broadcast();
 
     wsStreamSubscription = ws.listen((message) {
       MapEntry<Object, bool> received = acceptor.receive(message);
       ws.add(received.key);
       if (received.value) {
-        wsStreamSubscription.onData(null);
+        wsStreamSubscription
+          ..onData(null)
+          ..onDone(null);
+
         var base = BaseSession(ws, wsStreamSubscription, acceptor.getSessionDetails(), _serializer);
         completer.complete(base);
       }
     });
 
-    wsStreamSubscription.onDone(() {
-      sessionStreamController.stream.isEmpty.then(
-        (isEmpty) => {
-          if (!isEmpty) {sessionStreamController.close()},
-        },
-      );
-      wsStreamSubscription.cancel();
-    });
-
     return completer.future;
   }
 }
diff --git a/lib/src/joiner.dart b/lib/src/joiner.dart
@@ -26,28 +26,22 @@ class WAMPSessionJoiner {
 
     var welcomeCompleter = Completer<BaseSession>();
 
+    // ignore: cancel_subscriptions
     late StreamSubscription<dynamic> wsStreamSubscription;
-    final sessionStreamController = StreamController.broadcast();
 
     wsStreamSubscription = ws.listen((event) {
       dynamic toSend = joiner.receive(event);
       if (toSend == null) {
-        wsStreamSubscription.onData(null);
+        wsStreamSubscription
+          ..onData(null)
+          ..onDone(null);
 
         BaseSession baseSession = BaseSession(ws, wsStreamSubscription, joiner.getSessionDetails(), _serializer);
         welcomeCompleter.complete(baseSession);
       } else {
         ws.add(toSend);
       }
     });
-    wsStreamSubscription.onDone(() {
-      sessionStreamController.stream.isEmpty.then(
-        (isEmpty) => {
-          if (!isEmpty) {sessionStreamController.close()},
-        },
-      );
-      wsStreamSubscription.cancel();
-    });
 
     return welcomeCompleter.future;
   }

diff --git a/lib/src/types.dart b/lib/src/types.dart
@@ -113,6 +113,7 @@ class BaseSession extends IBaseSession {
 
   @override
   Future<void> close() async {
+    await _wsStreamSubscription.cancel();
     await _ws.close();
   }
 }