Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(201)

Unified Diff: lib/src/isolate_channel/send_port_sink.dart

Issue 1671763002: Make IsolateChannel use StreamChannelCompleter. (Closed) Base URL: git@github.com:dart-lang/stream_channel.git@master
Patch Set: Created 4 years, 10 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « lib/src/isolate_channel.dart ('k') | lib/src/stream_channel_controller.dart » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: lib/src/isolate_channel/send_port_sink.dart
diff --git a/lib/src/isolate_channel/send_port_sink.dart b/lib/src/isolate_channel/send_port_sink.dart
deleted file mode 100644
index d98f1da65db04e2ee798622c6d32c1ed5ebef707..0000000000000000000000000000000000000000
--- a/lib/src/isolate_channel/send_port_sink.dart
+++ /dev/null
@@ -1,111 +0,0 @@
-// 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';
-
-/// 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 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;
-
- 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._receivePort, this._sendPort);
-
- 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) {
- _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;
-
- _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 new Future.value();
-
- _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;
- });
- }
-}
« no previous file with comments | « lib/src/isolate_channel.dart ('k') | lib/src/stream_channel_controller.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698