Add an IsolateChannel class. R=rnystrom@google.com Review URL: https://codereview.chromium.org//1635873002 .
diff --git a/pkgs/stream_channel/lib/src/isolate_channel.dart b/pkgs/stream_channel/lib/src/isolate_channel.dart new file mode 100644 index 0000000..c664543 --- /dev/null +++ b/pkgs/stream_channel/lib/src/isolate_channel.dart
@@ -0,0 +1,146 @@ +// Copyright (c) 2016, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; +import 'dart:isolate'; + +import '../stream_channel.dart'; + +/// A [StreamChannel] that communicates over a [ReceivePort]/[SendPort] pair, +/// presumably with another isolate. +/// +/// The remote endpoint doesn't necessarily need to be running an +/// [IsolateChannel]. This can be used with any two ports, although the +/// [StreamChannel] semantics mean that this class will treat them as being +/// paired (for example, closing the [sink] will cause the [stream] to stop +/// emitting events). +/// +/// The underlying isolate ports have no notion of closing connections. This +/// means that [stream] won't close unless [sink] is closed, and that closing +/// [sink] won't cause the remote endpoint to close. Users should take care to +/// ensure that they always close the [sink] of every [IsolateChannel] they use +/// to avoid leaving dangling [ReceivePort]s. +class IsolateChannel<T> extends StreamChannelMixin<T> { + /// The port that produces incoming messages. + /// + /// This is wrapped in a [StreamView] to produce [stream]. + final ReceivePort _receivePort; + + /// The port that sends outgoing messages. + final SendPort _sendPort; + + Stream<T> get stream => _stream; + final Stream<T> _stream; + + StreamSink<T> get sink => _sink; + _SendPortSink<T> _sink; + + /// Creates a stream channel that receives messages from [receivePort] and + /// sends them over [sendPort]. + IsolateChannel(ReceivePort receivePort, this._sendPort) + : _receivePort = receivePort, + _stream = new StreamView<T>(receivePort) { + _sink = new _SendPortSink<T>(this); + } +} + +/// The sink for [IsolateChannel]. +/// +/// [SendPort] doesn't natively implement any sink API, so this adds that API as +/// a wrapper. Closing this just closes the [ReceivePort]. +class _SendPortSink<T> implements StreamSink<T> { + /// The channel that this sink is for. + final IsolateChannel _channel; + + Future get done => _doneCompleter.future; + final _doneCompleter = new Completer(); + + /// Whether [done] has been completed. + /// + /// This is distinct from [_closed] because [done] can complete with an error + /// without the user explicitly calling [close]. + bool get _isDone => _doneCompleter.isCompleted; + + /// Whether the user has called [close]. + bool _closed = false; + + /// Whether we're currently adding a stream with [addStream]. + bool _inAddStream = false; + + _SendPortSink(this._channel); + + void add(T data) { + if (_closed) throw new StateError("Cannot add event after closing."); + if (_inAddStream) { + throw new StateError("Cannot add event while adding stream."); + } + if (_isDone) return; + + _add(data); + } + + /// A helper for [add] that doesn't check for [StateError]s. + /// + /// This is called from [addStream], so it shouldn't check [_inAddStream]. + void _add(T data) { + _channel._sendPort.send(data); + } + + void addError(error, [StackTrace stackTrace]) { + if (_closed) throw new StateError("Cannot add event after closing."); + if (_inAddStream) { + throw new StateError("Cannot add event while adding stream."); + } + + _close(error, stackTrace); + } + + Future close() { + if (_inAddStream) { + throw new StateError("Cannot close sink while adding stream."); + } + + _closed = true; + return _close(); + } + + /// A helper for [close] that doesn't check for [StateError]s. + /// + /// This is called from [addStream], so it shouldn't check [_inAddStream]. It + /// also forwards [error] and [stackTrace] to [done] if they're passed. + Future _close([error, StackTrace stackTrace]) { + if (_isDone) return done; + + _channel._receivePort.close(); + + if (error != null) { + _doneCompleter.completeError(error, stackTrace); + } else { + _doneCompleter.complete(); + } + + return done; + } + + Future addStream(Stream<T> stream) { + if (_closed) throw new StateError("Cannot add stream after closing."); + if (_inAddStream) { + throw new StateError("Cannot add stream while adding stream."); + } + if (_isDone) return; + + _inAddStream = true; + var completer = new Completer.sync(); + stream.listen(_add, + onError: (error, stackTrace) { + _close(error, stackTrace); + completer.complete(); + }, + onDone: completer.complete, + cancelOnError: true); + return completer.future.then((_) { + _inAddStream = false; + }); + } +}
diff --git a/pkgs/stream_channel/lib/stream_channel.dart b/pkgs/stream_channel/lib/stream_channel.dart index 4b3c659..ff36ec7 100644 --- a/pkgs/stream_channel/lib/stream_channel.dart +++ b/pkgs/stream_channel/lib/stream_channel.dart
@@ -5,6 +5,7 @@ import 'dart:async'; export 'src/delegating_stream_channel.dart'; +export 'src/isolate_channel.dart'; export 'src/multi_channel.dart'; export 'src/stream_channel_completer.dart';
diff --git a/pkgs/stream_channel/pubspec.yaml b/pkgs/stream_channel/pubspec.yaml index a0d8d22..97057ab 100644 --- a/pkgs/stream_channel/pubspec.yaml +++ b/pkgs/stream_channel/pubspec.yaml
@@ -4,7 +4,7 @@ author: Dart Team <misc@dartlang.org> homepage: https://github.com/dart-lang/stream_channel environment: - sdk: '>=1.0.0 <2.0.0' + sdk: '>=1.8.0 <2.0.0' dependencies: async: '^1.8.0' dev_dependencies:
diff --git a/pkgs/stream_channel/test/isolate_channel_test.dart b/pkgs/stream_channel/test/isolate_channel_test.dart new file mode 100644 index 0000000..9e4fddc --- /dev/null +++ b/pkgs/stream_channel/test/isolate_channel_test.dart
@@ -0,0 +1,126 @@ +// Copyright (c) 2016, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; +import 'dart:isolate'; + +import 'package:stream_channel/stream_channel.dart'; +import 'package:test/test.dart'; + +import 'utils.dart'; + +void main() { + var receivePort; + var sendPort; + var channel; + setUp(() { + receivePort = new ReceivePort(); + var receivePortForSend = new ReceivePort(); + sendPort = receivePortForSend.sendPort; + channel = new IsolateChannel(receivePortForSend, receivePort.sendPort); + }); + + tearDown(() { + receivePort.close(); + channel.sink.close(); + }); + + test("the channel can send messages", () { + channel.sink.add(1); + channel.sink.add(2); + channel.sink.add(3); + + expect(receivePort.take(3).toList(), completion(equals([1, 2, 3]))); + }); + + test("the channel can receive messages", () { + sendPort.send(1); + sendPort.send(2); + sendPort.send(3); + + expect(channel.stream.take(3).toList(), completion(equals([1, 2, 3]))); + }); + + test("events can't be added to an explicitly-closed sink", () { + expect(channel.sink.close(), completes); + expect(() => channel.sink.add(1), throwsStateError); + expect(() => channel.sink.addError("oh no"), throwsStateError); + expect(() => channel.sink.addStream(new Stream.fromIterable([])), + throwsStateError); + }); + + test("events can't be added while a stream is being added", () { + var controller = new StreamController(); + channel.sink.addStream(controller.stream); + + expect(() => channel.sink.add(1), throwsStateError); + expect(() => channel.sink.addError("oh no"), throwsStateError); + expect(() => channel.sink.addStream(new Stream.fromIterable([])), + throwsStateError); + expect(() => channel.sink.close(), throwsStateError); + + controller.close(); + }); + + group("stream channel rules", () { + test("closing the sink causes the stream to close before it emits any more " + "events", () { + sendPort.send(1); + sendPort.send(2); + sendPort.send(3); + sendPort.send(4); + sendPort.send(5); + + channel.stream.listen(expectAsync((message) { + expect(message, equals(1)); + channel.sink.close(); + }, count: 1)); + }); + + test("cancelling the stream's subscription has no effect on the sink", + () async { + channel.stream.listen(null).cancel(); + await pumpEventQueue(); + + channel.sink.add(1); + channel.sink.add(2); + channel.sink.add(3); + expect(receivePort.take(3).toList(), completion(equals([1, 2, 3]))); + }); + + test("the sink closes as soon as an error is added", () async { + channel.sink.addError("oh no"); + channel.sink.add(1); + expect(channel.sink.done, throwsA("oh no")); + + // Since the sink is closed, the stream should also be closed. + expect(channel.stream.isEmpty, completion(isTrue)); + + // The other end shouldn't receive the next event, since the sink was + // closed. Pump the event queue to give it a chance to. + receivePort.listen(expectAsync((_) {}, count: 0)); + await pumpEventQueue(); + }); + + test("the sink closes as soon as an error is added via addStream", + () async { + var canceled = false; + var controller = new StreamController(onCancel: () { + canceled = true; + }); + + // This future shouldn't get the error, because it's sent to [Sink.done]. + expect(channel.sink.addStream(controller.stream), completes); + + controller.addError("oh no"); + expect(channel.sink.done, throwsA("oh no")); + await pumpEventQueue(); + expect(canceled, isTrue); + + // Even though the sink is closed, this shouldn't throw an error because + // the user didn't explicitly close it. + channel.sink.add(1); + }); + }); +}