Fix a race condition (flutter#16)

MultiChannel wasn't buffering incoming remote events if their virtual
channels hadn't been created locally yet.

Author: nex3

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 1.6.4
+
+* Fix a race condition in `MultiChannel` where messages from a remote virtual
+  channel could get dropped if the corresponding local channel wasn't registered
+  quickly enough.
+
 ## 1.6.3
 
 * Use `pumpEventQueue()` from test.

lib/src/multi_channel.dart

@@ -93,10 +93,18 @@ class _MultiChannel extends StreamChannelMixin implements MultiChannel {
   /// The controller for this channel.
   final _mainController = new StreamChannelController(sync: true);
 
-  /// A map from virtual channel ids to [StreamChannelController]s that should
-  /// be used to communicate over those channels.
+  /// A map from input IDs to [StreamChannelController]s that should be used to
+  /// communicate over those channels.
   final _controllers = <int, StreamChannelController>{};
 
+  /// Input IDs of controllers in [_controllers] that we've received messages
+  /// for but that have not yet had a local [virtualChannel] created.
+  final _pendingIds = new Set<int>();
+
+  /// Input IDs of virtual channels that used to exist but have since been
+  /// closed.
+  final _closedIds = new Set<int>();
+
   /// The next id to use for a local virtual channel.
   ///
   /// Ids are used to identify virtual channels. Each message is tagged with an
@@ -114,8 +122,9 @@ class _MultiChannel extends StreamChannelMixin implements MultiChannel {
   /// The trick is that each endpoint only uses odd ids for its own channels.
   /// When sending a message over a channel that was created by the remote
   /// endpoint, the channel's id plus one is used. This way each [MultiChannel]
-  /// knows that if an incoming message has an odd id, it's using the local id
-  /// scheme, but if it has an even id, it's using the remote id scheme.
+  /// knows that if an incoming message has an odd id, it's coming from a
+  /// channel that was originally created remotely, but if it has an even id,
+  /// it's coming from a channel that was originally created locally.
   var _nextId = 1;
 
   _MultiChannel(this._inner) {
@@ -128,21 +137,28 @@ class _MultiChannel extends StreamChannelMixin implements MultiChannel {
 
     _innerStreamSubscription = _inner.stream.listen((message) {
       var id = message[0];
-      var controller = _controllers[id];
 
-      // A controller might not exist if the channel was closed before an
-      // incoming message was processed.
-      if (controller == null) return;
+      // If the channel was closed before an incoming message was processed,
+      // ignore that message.
+      if (_closedIds.contains(id)) return;
+
+      var controller = _controllers.putIfAbsent(id, () {
+        // If we receive a message for a controller that doesn't have a local
+        // counterpart yet, create a controller for it to buffer incoming
+        // messages for when a local connection is created.
+        _pendingIds.add(id);
+        return new StreamChannelController(sync: true);
+      });
+
       if (message.length > 1) {
         controller.local.sink.add(message[1]);
-        return;
+      } else {
+        // A message without data indicates that the channel has been closed. We
+        // can just close the sink here without doing any more cleanup, because
+        // the sink closing will cause the stream to emit a done event which
+        // will trigger more cleanup.
+        controller.local.sink.close();
       }
-
-      // A message without data indicates that the channel has been closed. We
-      // can only close the sink here without doing any more cleanup, because
-      // the sink closing will cause the stream to emit a done event which will
-      // trigger more cleanup.
-      controller.local.sink.close();
     },
         onDone: _closeInnerChannel,
         onError: _mainController.local.sink.addError);
@@ -173,23 +189,30 @@ class _MultiChannel extends StreamChannelMixin implements MultiChannel {
           this, inputId, new Stream.empty(), new NullStreamSink());
     }
 
-    if (_controllers.containsKey(inputId)) {
+    StreamChannelController controller;
+    if (_pendingIds.remove(inputId)) {
+      // If we've already received messages for this channel, use the controller
+      // where those messages are buffered.
+      controller = _controllers[inputId];
+    } else if (_controllers.containsKey(inputId) ||
+        _closedIds.contains(inputId)) {
       throw new ArgumentError("A virtual channel with id $id already exists.");
+    } else {
+      controller = new StreamChannelController(sync: true);
+      _controllers[inputId] = controller;
     }
 
-    var controller = new StreamChannelController(sync: true);
-    _controllers[inputId] = controller;
     controller.local.stream.listen(
         (message) => _inner.sink.add([outputId, message]),
         onDone: () => _closeChannel(inputId, outputId));
-
     return new VirtualChannel._(
         this, outputId, controller.foreign.stream, controller.foreign.sink);
   }
 
   /// Closes the virtual channel for which incoming messages have [inputId] and
   /// outgoing messages have [outputId].
   void _closeChannel(int inputId, int outputId) {
+    _closedIds.add(inputId);
     var controller = _controllers.remove(inputId);
     controller.local.sink.close();
 

pubspec.yaml

@@ -1,5 +1,5 @@
 name: stream_channel
-version: 1.6.3
+version: 1.6.4
 description: An abstraction for two-way communication channels.
 author: Dart Team <[email protected]>
 homepage: https://github.com/dart-lang/stream_channel

test/multi_channel_test.dart

@@ -268,6 +268,32 @@ void main() {
     test("doesn't allow another virtual channel with the same id", () {
       expect(() => channel2.virtualChannel(virtual1.id), throwsArgumentError);
     });
+
+    test("dispatches events received before the virtual channel is created",
+        () async {
+      virtual1 = channel1.virtualChannel();
+
+      virtual1.sink.add("hello");
+      await pumpEventQueue();
+
+      virtual1.sink.add("world");
+      await pumpEventQueue();
+
+      expect(channel2.virtualChannel(virtual1.id).stream,
+          emitsInOrder(["hello", "world"]));
+    });
+
+    test(
+        "dispatches close events received before the virtual channel is "
+        "created", () async {
+      virtual1 = channel1.virtualChannel();
+
+      virtual1.sink.close();
+      await pumpEventQueue();
+
+      expect(channel2.virtualChannel(virtual1.id).stream.toList(),
+          completion(isEmpty));
+    });
   });
 
   group("when the underlying stream", () {