Fix a race condition in IsolateChannel.connectReceive() (dart-lang/st…

…ream_channel#92)

pkgs/stream_channel/CHANGELOG.md

@@ -1,7 +1,9 @@
-## 2.1.2-dev
+## 2.1.2
 
 * Require Dart 2.19
 * Add an example.
+* Fix a race condition in `IsolateChannel.connectReceive()` where the channel
+  could hang forever if its sink was closed before the connection was established.
 
 ## 2.1.1
 

pkgs/stream_channel/lib/src/isolate_channel.dart

@@ -43,16 +43,26 @@ class IsolateChannel<T> extends StreamChannelMixin<T> {
   factory IsolateChannel.connectReceive(ReceivePort receivePort) {
     // We can't use a [StreamChannelCompleter] here because we need the return
     // value to be an [IsolateChannel].
+    var isCompleted = false;
     var streamCompleter = StreamCompleter<T>();
     var sinkCompleter = StreamSinkCompleter<T>();
-    var channel =
-        IsolateChannel<T>._(streamCompleter.stream, sinkCompleter.sink);
+
+    var channel = IsolateChannel<T>._(streamCompleter.stream, sinkCompleter.sink
+        .transform(StreamSinkTransformer.fromHandlers(handleDone: (sink) {
+      if (!isCompleted) {
+        receivePort.close();
+        streamCompleter.setSourceStream(Stream.empty());
+        sinkCompleter.setDestinationSink(NullStreamSink<T>());
+      }
+      sink.close();
+    })));
 
     // The first message across the ReceivePort should be a SendPort pointing to
     // the remote end. If it's not, we'll make the stream emit an error
     // complaining.
     late StreamSubscription<dynamic> subscription;
     subscription = receivePort.listen((message) {
+      isCompleted = true;
       if (message is SendPort) {
         var controller =
             StreamChannelController<T>(allowForeignErrors: false, sync: true);

pkgs/stream_channel/pubspec.yaml

@@ -1,5 +1,5 @@
 name: stream_channel
-version: 2.1.2-dev
+version: 2.1.2
 description: >-
   An abstraction for two-way communication channels based on the Dart Stream
   class.

pkgs/stream_channel/test/isolate_channel_test.dart

@@ -162,5 +162,13 @@ void main() {
       expect(connectedChannel.stream.toList(), throwsStateError);
       expect(connectedChannel.sink.done, completes);
     });
+
+    test('the receiving channel closes gracefully without a connection',
+        () async {
+      var connectedChannel = IsolateChannel.connectReceive(connectPort);
+      await connectedChannel.sink.close();
+      await expectLater(connectedChannel.stream.toList(), completion(isEmpty));
+      await expectLater(connectedChannel.sink.done, completes);
+    });
   });
 }