Merge pull request #24 from atsign-foundation/fix-2.3.2

fix: stability under load

Author: gkc

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 2.3.2
+
+- fix: stability under load
+  - ensure that socket.done is handled in all cases
+  - check a side's state before attempting to write to that side's socket
+
 ## 2.3.1
 
 - fix: correctly handle situation where a socket has been closed but the other

lib/src/socket_connector.dart

@@ -136,23 +136,31 @@ class SocketConnector {
           'Added connection. There are now ${connections.length} connections.'));
 
       for (final side in [thisSide, thisSide.farSide!]) {
+        unawaited(side.socket.done
+            .then((v) => _destroySide(side))
+            .catchError((err) => _destroySide(side)));
         if (side.transformer != null) {
           // transformer is there to transform data originating FROM its side
           StreamController<Uint8List> sc = StreamController<Uint8List>();
           side.farSide!.sink = sc;
           Stream<List<int>> transformed = side.transformer!(sc.stream);
-          transformed.listen((event) async {
+          transformed.listen((data) {
             try {
-              side.farSide!.socket.add(event);
-              await side.farSide!.socket.flush();
-            } catch (e) {
+              if (side.farSide!.state == SideState.open) {
+                side.farSide!.socket.add(data);
+              } else {
+                throw StateError(
+                    'Will not write to side ${side.farSide!.name} as its state is ${side.farSide!.state}');
+              }
+            } catch (e, st) {
               _log('Failed to write to side ${side.farSide!.name} - closing',
                   force: true);
+              _log('(Error was $e; Stack trace follows\n$st');
               _destroySide(side.farSide!);
             }
           });
         }
-        side.stream.listen((Uint8List data) async {
+        side.stream.listen((Uint8List data) {
           if (logTraffic) {
             final message = String.fromCharCodes(data);
             if (side.isSideA) {
@@ -164,11 +172,16 @@ class SocketConnector {
             }
           }
           try {
-            side.farSide!.sink.add(data);
-            await side.farSide!.socket.flush();
-          } catch (e) {
+            if (side.farSide!.state == SideState.open) {
+              side.farSide!.sink.add(data);
+            } else {
+              throw StateError(
+                  'Will not write to side ${side.farSide!.name} as its state is ${side.farSide!.state}');
+            }
+          } catch (e, st) {
             _log('Failed to write to side ${side.farSide!.name} - closing',
                 force: true);
+            _log('(Error was $e; Stack trace follows\n$st');
             _destroySide(side.farSide!);
           }
         }, onDone: () {

pubspec.yaml

@@ -1,7 +1,7 @@
 name: socket_connector
 description: Package for joining sockets together to create socket relays.
 
-version: 2.3.1
+version: 2.3.2
 repository: https://github.com/cconstab/socket_connector
 
 environment: