Modernize the package's style. This moves the package to new-style doc comments, deprecates separate top-level libraries, and removes library tags. It also deprecates some top-level classes in favor of static const fields. There's more that could be done, but this fixes most of the low-hanging fruit. R=lrn@google.com Review URL: https://codereview.chromium.org//1777453002 .
diff --git a/CHANGELOG.md b/CHANGELOG.md index 15570d7..946d3a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md
@@ -1,3 +1,16 @@ +## 1.9.0 + +* Deprecate top-level libraries other than `package:async/async.dart`, which + exports these libraries' interfaces. + +* Add `Result.captureStreamTransformer`, `Result.releaseStreamTransformer`, + `Result.captureSinkTransformer`, and `Result.releaseSinkTransformer`. + +* Deprecate `CaptureStreamTransformer`, `ReleaseStreamTransformer`, + `CaptureSink`, and `ReleaseSink`. `Result.captureStreamTransformer`, + `Result.releaseStreamTransformer`, `Result.captureSinkTransformer`, and + `Result.releaseSinkTransformer` should be used instead. + ## 1.8.0 - Added `StreamSinkCompleter`, for creating a `StreamSink` now and providing its
diff --git a/README.md b/README.md index 3820f06..16cd55f 100644 --- a/README.md +++ b/README.md
@@ -1,28 +1,15 @@ -# Async utilities package - -Contains tools to work with asynchronous computations. - -The package contains `Stream` and `Future` related functionality, -as well as sub-libraries with different utilities. +Contains utility classes in the style of `dart:async` to work with asynchronous +computations. ### Zipping streams -The "stream_zip.dart" sub-library contains functionality -to combine several streams of events into a single stream of tuples of events. +The `StreamZip` class can combine several streams of events into a single stream +of tuples of events. ### Results -The "result.dart" sub-library introduces a `Result` class that can hold either -a value or an error. -It allows capturing an asynchronous computation which can give either a value -or an error, into an asynchronous computation that always gives a `Result` -value, where errors can be treated as data. -It also allows releasing the `Result` back into an asynchronous computation. -### History. -This package is unrelated to the discontinued `async` package with version 0.1.7. - -## Features and bugs - -Please file feature requests and bugs at the [issue tracker][tracker]. - -[tracker]: https://github.com/dart-lang/async/issues +The package introduces a `Result` class that can hold either a value or an +error. It allows capturing an asynchronous computation which can give either a +value or an error, into an asynchronous computation that always gives a `Result` +value, where errors can be treated as data. It also allows releasing the +`Result` back into an asynchronous computation.
diff --git a/lib/async.dart b/lib/async.dart index ef1ac04..0d3e051 100644 --- a/lib/async.dart +++ b/lib/async.dart
@@ -2,9 +2,6 @@ // 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. -library dart.pkg.async; - -export "result.dart"; export "src/async_memoizer.dart"; export "src/cancelable_operation.dart"; export "src/delegate/event_sink.dart"; @@ -17,7 +14,12 @@ export "src/lazy_stream.dart"; export "src/null_stream_sink.dart"; export "src/restartable_timer.dart"; -export "src/result_future.dart"; +export "src/result.dart"; +export "src/result/capture_transformer.dart"; +export "src/result/error.dart"; +export "src/result/future.dart"; +export "src/result/release_transformer.dart"; +export "src/result/value.dart"; export "src/single_subscription_transformer.dart"; export "src/stream_completer.dart"; export "src/stream_group.dart"; @@ -25,5 +27,5 @@ export "src/stream_sink_completer.dart"; export "src/stream_sink_transformer.dart"; export "src/stream_splitter.dart"; +export "src/stream_zip.dart"; export "src/subscription_stream.dart"; -export "stream_zip.dart";
diff --git a/lib/result.dart b/lib/result.dart index 13e3dd9..627f200 100644 --- a/lib/result.dart +++ b/lib/result.dart
@@ -2,255 +2,12 @@ // 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. -/// Capture asynchronous results into synchronous values. -/// -/// Capturing a result (either a returned value or a thrown error) -/// means converting it into a [Result] - -/// either a [ValueResult] or an [ErrorResult]. -/// -/// This value can release itself by writing itself either to a -/// [EventSink] or a [Completer], or by becoming a [Future]. +/// Import `async.dart` instead. +@Deprecated("Will be removed in async 2.0.0.") library dart.pkg.async.results; -import "dart:async"; - -/// The result of a computation. -abstract class Result<T> { - /// Create a `Result` with the result of calling [computation]. - /// - /// This generates either a [ValueResult] with the value returned by - /// calling `computation`, or an [ErrorResult] with an error thrown by - /// the call. - factory Result(T computation()) { - try { - return new ValueResult(computation()); - } catch (e, s) { - return new ErrorResult(e, s); - } - } - - /// Create a `Result` holding a value. - /// - /// Alias for [ValueResult.ValueResult]. - factory Result.value(T value) = ValueResult<T>; - - /// Create a `Result` holding an error. - /// - /// Alias for [ErrorResult.ErrorResult]. - factory Result.error(Object error, [StackTrace stackTrace]) => - new ErrorResult(error, stackTrace); - - // Helper functions. - static _captureValue(value) => new ValueResult(value); - static _captureError(error, stack) => new ErrorResult(error, stack); - static _release(Result v) { - if (v.isValue) return v.asValue.value; // Avoid wrapping in future. - return v.asFuture; - } - - /// Capture the result of a future into a `Result` future. - /// - /// The resulting future will never have an error. - /// Errors have been converted to an [ErrorResult] value. - static Future<Result> capture(Future future) { - return future.then(_captureValue, onError: _captureError); - } - - /// Release the result of a captured future. - /// - /// Converts the [Result] value of the given [future] to a value or error - /// completion of the returned future. - /// - /// If [future] completes with an error, the returned future completes with - /// the same error. - static Future release(Future<Result> future) { - return future.then(_release); - } - - /// Capture the results of a stream into a stream of [Result] values. - /// - /// The returned stream will not have any error events. - /// Errors from the source stream have been converted to [ErrorResult]s. - /// - /// Shorthand for transforming the stream using [CaptureStreamTransformer]. - static Stream<Result> captureStream(Stream source) { - return source.transform(const CaptureStreamTransformer()); - } - - /// Release a stream of [result] values into a stream of the results. - /// - /// `Result` values of the source stream become value or error events in - /// the returned stream as appropriate. - /// Errors from the source stream become errors in the returned stream. - /// - /// Shorthand for transforming the stream using [ReleaseStreamTransformer]. - static Stream releaseStream(Stream<Result> source) { - return source.transform(const ReleaseStreamTransformer()); - } - - /// Converts a result of a result to a single result. - /// - /// If the result is an error, or it is a `Result` value - /// which is then an error, then a result with that error is returned. - /// Otherwise both levels of results are value results, and a single - /// result with the value is returned. - static Result flatten(Result<Result> result) { - if (result.isError) return result; - return result.asValue.value; - } - - /// Whether this result is a value result. - /// - /// Always the opposite of [isError]. - bool get isValue; - - /// Whether this result is an error result. - /// - /// Always the opposite of [isValue]. - bool get isError; - - /// If this is a value result, return itself. - /// - /// Otherwise return `null`. - ValueResult<T> get asValue; - - /// If this is an error result, return itself. - /// - /// Otherwise return `null`. - ErrorResult get asError; - - /// Complete a completer with this result. - void complete(Completer<T> completer); - - /// Add this result to an [EventSink]. - /// - /// Calls the sink's `add` or `addError` method as appropriate. - void addTo(EventSink<T> sink); - - /// Creates a future completed with this result as a value or an error. - Future<T> get asFuture; -} - -/// A result representing a returned value. -class ValueResult<T> implements Result<T> { - final T value; - ValueResult(this.value); - bool get isValue => true; - bool get isError => false; - ValueResult<T> get asValue => this; - ErrorResult get asError => null; - void complete(Completer<T> completer) { - completer.complete(value); - } - void addTo(EventSink<T> sink) { - sink.add(value); - } - Future<T> get asFuture => new Future.value(value); -} - -/// A result representing a thrown error. -class ErrorResult implements Result { - final error; - final StackTrace stackTrace; - ErrorResult(this.error, this.stackTrace); - bool get isValue => false; - bool get isError => true; - ValueResult get asValue => null; - ErrorResult get asError => this; - void complete(Completer completer) { - completer.completeError(error, stackTrace); - } - void addTo(EventSink sink) { - sink.addError(error, stackTrace); - } - Future get asFuture => new Future.error(error, stackTrace); - - /// Calls an error handler with the error and stacktrace. - /// - /// An async error handler function is either a function expecting two - /// arguments, which will be called with the error and the stack trace, - /// or it has to be a function expecting only one argument, - /// which will be called with only the error. - void handle(Function errorHandler) { - if (errorHandler is ZoneBinaryCallback) { - errorHandler(error, stackTrace); - } else { - errorHandler(error); - } - } -} - -/// A stream transformer that captures a stream of events into [Result]s. -/// -/// The result of the transformation is a stream of [Result] values and -/// no error events. -class CaptureStreamTransformer<T> implements StreamTransformer<T, Result<T>> { - const CaptureStreamTransformer(); - - Stream<Result<T>> bind(Stream<T> source) { - return new Stream<Result<T>>.eventTransformed(source, _createSink); - } - - static EventSink _createSink(EventSink<Result> sink) { - return new CaptureSink(sink); - } -} - -/// A stream transformer that releases a stream of result events. -/// -/// The result of the transformation is a stream of values and -/// error events. -class ReleaseStreamTransformer<T> implements StreamTransformer<Result<T>, T> { - const ReleaseStreamTransformer(); - - Stream<T> bind(Stream<Result<T>> source) { - return new Stream<T>.eventTransformed(source, _createSink); - } - - static EventSink<Result> _createSink(EventSink sink) { - return new ReleaseSink(sink); - } -} - -/// An event sink wrapper that captures the incoming events. -/// -/// Wraps an [EventSink] that expects [Result] values. -/// Accepts any value and error result, -/// and passes them to the wrapped sink as [Result] values. -/// -/// The wrapped sink will never receive an error event. -class CaptureSink<T> implements EventSink<T> { - final EventSink _sink; - - CaptureSink(EventSink<Result<T>> sink) : _sink = sink; - void add(T value) { _sink.add(new ValueResult(value)); } - void addError(Object error, [StackTrace stackTrace]) { - _sink.add(new ErrorResult(error, stackTrace)); - } - void close() { _sink.close(); } -} - -/// An event sink wrapper that releases the incoming result events. -/// -/// Wraps an output [EventSink] that expects any result. -/// Accepts [Result] values, and puts the result value or error into the -/// corresponding output sink add method. -class ReleaseSink<T> implements EventSink<Result<T>> { - final EventSink _sink; - ReleaseSink(EventSink<T> sink) : _sink = sink; - void add(Result<T> result) { - if (result.isValue) { - _sink.add(result.asValue.value); - } else { - ErrorResult error = result.asError; - _sink.addError(error.error, error.stackTrace); - } - } - void addError(Object error, [StackTrace stackTrace]) { - // Errors may be added by intermediate processing, even if it is never - // added by CaptureSink. - _sink.addError(error, stackTrace); - } - - void close() { _sink.close(); } -} +export "src/result.dart"; +export "src/result/capture_transformer.dart"; +export "src/result/error.dart"; +export "src/result/release_transformer.dart"; +export "src/result/value.dart";
diff --git a/lib/src/async_memoizer.dart b/lib/src/async_memoizer.dart index 41c3dae..3612a7d 100644 --- a/lib/src/async_memoizer.dart +++ b/lib/src/async_memoizer.dart
@@ -2,8 +2,6 @@ // 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. -library async.async_memoizer; - import 'dart:async'; /// A class for running an asynchronous function exactly once and caching its @@ -35,12 +33,12 @@ Future<T> get future => _completer.future; final _completer = new Completer(); - /// Whether [run] has been called yet. + /// Whether [runOnce] has been called yet. bool get hasRun => _completer.isCompleted; /// Runs the function, [computation], if it hasn't been run before. /// - /// If [run] has already been called, this returns the original result. + /// If [runOnce] has already been called, this returns the original result. Future<T> runOnce(computation()) { if (!hasRun) _completer.complete(new Future.sync(computation)); return future;
diff --git a/lib/src/cancelable_operation.dart b/lib/src/cancelable_operation.dart index a48c94f..1ba6b3c 100644 --- a/lib/src/cancelable_operation.dart +++ b/lib/src/cancelable_operation.dart
@@ -2,8 +2,6 @@ // 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. -library async.cancelable_operation; - import 'dart:async'; import 'package:async/async.dart';
diff --git a/lib/src/delegate/event_sink.dart b/lib/src/delegate/event_sink.dart index 337e7a8..5a525df 100644 --- a/lib/src/delegate/event_sink.dart +++ b/lib/src/delegate/event_sink.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.event_sink; - import 'dart:async'; /// Simple delegating wrapper around an [EventSink].
diff --git a/lib/src/delegate/future.dart b/lib/src/delegate/future.dart index 34f6158..5e84e4f 100644 --- a/lib/src/delegate/future.dart +++ b/lib/src/delegate/future.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.future; - import 'dart:async'; /// A wrapper that forwards calls to a [Future].
diff --git a/lib/src/delegate/sink.dart b/lib/src/delegate/sink.dart index bb50da3..cee2937 100644 --- a/lib/src/delegate/sink.dart +++ b/lib/src/delegate/sink.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.sink; - /// Simple delegating wrapper around a [Sink]. /// /// Subclasses can override individual methods, or use this to expose only the
diff --git a/lib/src/delegate/stream_consumer.dart b/lib/src/delegate/stream_consumer.dart index 8162db6..dcaf0c2 100644 --- a/lib/src/delegate/stream_consumer.dart +++ b/lib/src/delegate/stream_consumer.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.stream_consumer; - import 'dart:async'; /// Simple delegating wrapper around a [StreamConsumer].
diff --git a/lib/src/delegate/stream_sink.dart b/lib/src/delegate/stream_sink.dart index b6ace65..e06afc1 100644 --- a/lib/src/delegate/stream_sink.dart +++ b/lib/src/delegate/stream_sink.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.stream_sink; - import 'dart:async'; /// Simple delegating wrapper around a [StreamSink].
diff --git a/lib/src/delegate/stream_subscription.dart b/lib/src/delegate/stream_subscription.dart index ff9b665..e7153b2 100644 --- a/lib/src/delegate/stream_subscription.dart +++ b/lib/src/delegate/stream_subscription.dart
@@ -2,8 +2,6 @@ // 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. -library async.delegate.stream_subscription; - import 'dart:async'; /// Simple delegating wrapper around a [StreamSubscription].
diff --git a/lib/src/future_group.dart b/lib/src/future_group.dart index 02ff185..114c133 100644 --- a/lib/src/future_group.dart +++ b/lib/src/future_group.dart
@@ -2,8 +2,6 @@ // 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. -library async.future_group; - import 'dart:async'; /// A collection of futures waits until all added [Future]s complete. @@ -35,7 +33,7 @@ Future<List<T>> get future => _completer.future; final _completer = new Completer<List<T>>(); - /// Whether this group is waiting on any futures. + /// Whether this group has no pending futures. bool get isIdle => _pending == 0; /// A broadcast stream that emits a `null` event whenever the last pending
diff --git a/lib/src/lazy_stream.dart b/lib/src/lazy_stream.dart index 1878d4c..3ba4953 100644 --- a/lib/src/lazy_stream.dart +++ b/lib/src/lazy_stream.dart
@@ -2,8 +2,6 @@ // 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. -library async.lazy_stream; - import "dart:async"; import "stream_completer.dart";
diff --git a/lib/src/null_stream_sink.dart b/lib/src/null_stream_sink.dart index aa85924..c83790c 100644 --- a/lib/src/null_stream_sink.dart +++ b/lib/src/null_stream_sink.dart
@@ -2,8 +2,6 @@ // 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. -library async.null_stream_sink; - import 'dart:async'; /// A [StreamSink] that discards all events.
diff --git a/lib/src/restartable_timer.dart b/lib/src/restartable_timer.dart index 05196d2..eed51e6 100644 --- a/lib/src/restartable_timer.dart +++ b/lib/src/restartable_timer.dart
@@ -2,8 +2,6 @@ // 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. -library async.restartable_timer; - import 'dart:async'; /// A non-periodic timer that can be restarted any number of times.
diff --git a/lib/src/result.dart b/lib/src/result.dart new file mode 100644 index 0000000..482cf9b --- /dev/null +++ b/lib/src/result.dart
@@ -0,0 +1,164 @@ +// 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 'result/capture_transformer.dart'; +import 'result/error.dart'; +import 'result/release_transformer.dart'; +import 'result/value.dart'; +import 'stream_sink_transformer.dart'; + +/// The result of a computation. +/// +/// Capturing a result (either a returned value or a thrown error) means +/// converting it into a [Result] - either a [ValueResult] or an [ErrorResult]. +/// +/// This value can release itself by writing itself either to a [EventSink] or a +/// [Completer], or by becoming a [Future]. +abstract class Result<T> { + /// A stream transformer that captures a stream of events into [Result]s. + /// + /// The result of the transformation is a stream of [Result] values and no + /// error events. This is the transformer used by [captureStream]. + static const StreamTransformer<Object, Result> captureStreamTransformer = + const CaptureStreamTransformer(); + + /// A stream transformer that releases a stream of result events. + /// + /// The result of the transformation is a stream of values and error events. + /// This is the transformer used by [releaseStream]. + static const StreamTransformer<Object, Result> releaseStreamTransformer = + const ReleaseStreamTransformer(); + + /// A sink transformer that captures events into [Result]s. + /// + /// The result of the transformation is a sink that only forwards [Result] + /// values and no error events. + static const StreamSinkTransformer<Object, Result> captureSinkTransformer = + const StreamSinkTransformer.fromStreamTransformer( + const CaptureStreamTransformer()); + + /// A sink transformer that releases result events. + /// + /// The result of the transformation is a sink that forwards of values and + /// error events. + static const StreamSinkTransformer<Object, Result> releaseSinkTransformer = + const StreamSinkTransformer.fromStreamTransformer( + const ReleaseStreamTransformer()); + + /// Create a `Result` with the result of calling [computation]. + /// + /// This generates either a [ValueResult] with the value returned by + /// calling `computation`, or an [ErrorResult] with an error thrown by + /// the call. + factory Result(T computation()) { + try { + return new ValueResult(computation()); + } catch (e, s) { + return new ErrorResult(e, s); + } + } + + /// Create a `Result` holding a value. + /// + /// Alias for [ValueResult.ValueResult]. + factory Result.value(T value) = ValueResult<T>; + + /// Create a `Result` holding an error. + /// + /// Alias for [ErrorResult.ErrorResult]. + factory Result.error(Object error, [StackTrace stackTrace]) => + new ErrorResult(error, stackTrace); + + // Helper functions. + static _captureValue(value) => new ValueResult(value); + static _captureError(error, stack) => new ErrorResult(error, stack); + static _release(Result v) { + if (v.isValue) return v.asValue.value; // Avoid wrapping in future. + return v.asFuture; + } + + /// Capture the result of a future into a `Result` future. + /// + /// The resulting future will never have an error. + /// Errors have been converted to an [ErrorResult] value. + static Future<Result> capture(Future future) { + return future.then(_captureValue, onError: _captureError); + } + + /// Release the result of a captured future. + /// + /// Converts the [Result] value of the given [future] to a value or error + /// completion of the returned future. + /// + /// If [future] completes with an error, the returned future completes with + /// the same error. + static Future release(Future<Result> future) { + return future.then(_release); + } + + /// Capture the results of a stream into a stream of [Result] values. + /// + /// The returned stream will not have any error events. + /// Errors from the source stream have been converted to [ErrorResult]s. + /// + /// Shorthand for transforming the stream using [captureStreamTransformer]. + static Stream<Result> captureStream(Stream source) { + return source.transform(captureStreamTransformer); + } + + /// Release a stream of [result] values into a stream of the results. + /// + /// `Result` values of the source stream become value or error events in + /// the returned stream as appropriate. + /// Errors from the source stream become errors in the returned stream. + /// + /// Shorthand for transforming the stream using [releaseStreamTransformer]. + static Stream releaseStream(Stream<Result> source) { + return source.transform(releaseStreamTransformer); + } + + /// Converts a result of a result to a single result. + /// + /// If the result is an error, or it is a `Result` value + /// which is then an error, then a result with that error is returned. + /// Otherwise both levels of results are value results, and a single + /// result with the value is returned. + static Result flatten(Result<Result> result) { + if (result.isError) return result; + return result.asValue.value; + } + + /// Whether this result is a value result. + /// + /// Always the opposite of [isError]. + bool get isValue; + + /// Whether this result is an error result. + /// + /// Always the opposite of [isValue]. + bool get isError; + + /// If this is a value result, return itself. + /// + /// Otherwise return `null`. + ValueResult<T> get asValue; + + /// If this is an error result, return itself. + /// + /// Otherwise return `null`. + ErrorResult get asError; + + /// Complete a completer with this result. + void complete(Completer<T> completer); + + /// Add this result to an [EventSink]. + /// + /// Calls the sink's `add` or `addError` method as appropriate. + void addTo(EventSink<T> sink); + + /// Creates a future completed with this result as a value or an error. + Future<T> get asFuture; +}
diff --git a/lib/src/result/capture_sink.dart b/lib/src/result/capture_sink.dart new file mode 100644 index 0000000..7474525 --- /dev/null +++ b/lib/src/result/capture_sink.dart
@@ -0,0 +1,27 @@ +// 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 '../result.dart'; + +/// Use [Result.captureSinkTransformer]. +@Deprecated("Will be removed in async 2.0.0.") +class CaptureSink<T> implements EventSink<T> { + final EventSink _sink; + + CaptureSink(EventSink<Result<T>> sink) : _sink = sink; + + void add(T value) { + _sink.add(new Result.value(value)); + } + + void addError(Object error, [StackTrace stackTrace]) { + _sink.add(new Result.error(error, stackTrace)); + } + + void close() { + _sink.close(); + } +}
diff --git a/lib/src/result/capture_transformer.dart b/lib/src/result/capture_transformer.dart new file mode 100644 index 0000000..ca4d0d3 --- /dev/null +++ b/lib/src/result/capture_transformer.dart
@@ -0,0 +1,22 @@ +// 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 '../result.dart'; +import 'capture_sink.dart'; + +/// Use [Result.captureTransformer] instead. +@Deprecated("Will be removed in async 2.0.0.") +class CaptureStreamTransformer<T> implements StreamTransformer<T, Result<T>> { + const CaptureStreamTransformer(); + + Stream<Result<T>> bind(Stream<T> source) { + return new Stream<Result<T>>.eventTransformed(source, _createSink); + } + + static EventSink _createSink(EventSink<Result> sink) { + return new CaptureSink(sink); + } +}
diff --git a/lib/src/result/error.dart b/lib/src/result/error.dart new file mode 100644 index 0000000..eecf68e --- /dev/null +++ b/lib/src/result/error.dart
@@ -0,0 +1,45 @@ +// 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 '../result.dart'; +import 'value.dart'; + +/// A result representing a thrown error. +class ErrorResult implements Result { + final error; + final StackTrace stackTrace; + + bool get isValue => false; + bool get isError => true; + ValueResult get asValue => null; + ErrorResult get asError => this; + + ErrorResult(this.error, this.stackTrace); + + void complete(Completer completer) { + completer.completeError(error, stackTrace); + } + + void addTo(EventSink sink) { + sink.addError(error, stackTrace); + } + + Future get asFuture => new Future.error(error, stackTrace); + + /// Calls an error handler with the error and stacktrace. + /// + /// An async error handler function is either a function expecting two + /// arguments, which will be called with the error and the stack trace, or it + /// has to be a function expecting only one argument, which will be called + /// with only the error. + void handle(Function errorHandler) { + if (errorHandler is ZoneBinaryCallback) { + errorHandler(error, stackTrace); + } else { + errorHandler(error); + } + } +}
diff --git a/lib/src/result_future.dart b/lib/src/result/future.dart similarity index 93% rename from lib/src/result_future.dart rename to lib/src/result/future.dart index 311e83f..db9dd82 100644 --- a/lib/src/result_future.dart +++ b/lib/src/result/future.dart
@@ -2,12 +2,10 @@ // 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. -library async.result_future; - import 'dart:async'; +import '../delegate/future.dart'; import '../result.dart'; -import 'delegate/future.dart'; /// A [Future] wrapper that provides synchronous access to the result of the /// wrapped [Future] once it's completed.
diff --git a/lib/src/result/release_sink.dart b/lib/src/result/release_sink.dart new file mode 100644 index 0000000..114981b --- /dev/null +++ b/lib/src/result/release_sink.dart
@@ -0,0 +1,34 @@ +// 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 '../result.dart'; + +/// Use [Result.captureSinkTransformer]. +@Deprecated("Will be removed in async 2.0.0.") +class ReleaseSink<T> implements EventSink<Result<T>> { + final EventSink _sink; + + ReleaseSink(EventSink<T> sink) : _sink = sink; + + void add(Result<T> result) { + if (result.isValue) { + _sink.add(result.asValue.value); + } else { + var error = result.asError; + _sink.addError(error.error, error.stackTrace); + } + } + + void addError(Object error, [StackTrace stackTrace]) { + // Errors may be added by intermediate processing, even if it is never + // added by CaptureSink. + _sink.addError(error, stackTrace); + } + + void close() { + _sink.close(); + } +}
diff --git a/lib/src/result/release_transformer.dart b/lib/src/result/release_transformer.dart new file mode 100644 index 0000000..456ed0a --- /dev/null +++ b/lib/src/result/release_transformer.dart
@@ -0,0 +1,22 @@ +// 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 '../result.dart'; +import 'release_sink.dart'; + +/// Use [Result.releaseTransformer] instead. +@Deprecated("Will be removed in async 2.0.0.") +class ReleaseStreamTransformer<T> implements StreamTransformer<Result<T>, T> { + const ReleaseStreamTransformer(); + + Stream<T> bind(Stream<Result<T>> source) { + return new Stream<T>.eventTransformed(source, _createSink); + } + + static EventSink<Result> _createSink(EventSink sink) { + return new ReleaseSink(sink); + } +}
diff --git a/lib/src/result/value.dart b/lib/src/result/value.dart new file mode 100644 index 0000000..f065ab6 --- /dev/null +++ b/lib/src/result/value.dart
@@ -0,0 +1,30 @@ +// 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 '../result.dart'; +import 'error.dart'; + +/// A result representing a returned value. +class ValueResult<T> implements Result<T> { + final T value; + + bool get isValue => true; + bool get isError => false; + ValueResult<T> get asValue => this; + ErrorResult get asError => null; + + ValueResult(this.value); + + void complete(Completer<T> completer) { + completer.complete(value); + } + + void addTo(EventSink<T> sink) { + sink.add(value); + } + + Future<T> get asFuture => new Future.value(value); +}
diff --git a/lib/src/single_subscription_transformer.dart b/lib/src/single_subscription_transformer.dart index 28fd512..4255175 100644 --- a/lib/src/single_subscription_transformer.dart +++ b/lib/src/single_subscription_transformer.dart
@@ -2,8 +2,6 @@ // 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. -library async.single_subscription_transformer; - import 'dart:async'; /// A transformer that converts a broadcast stream into a single-subscription
diff --git a/lib/src/stream_completer.dart b/lib/src/stream_completer.dart index a6260dc..73557b2 100644 --- a/lib/src/stream_completer.dart +++ b/lib/src/stream_completer.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_completer; - import "dart:async"; /// A single-subscription [stream] where the contents are provided later.
diff --git a/lib/src/stream_group.dart b/lib/src/stream_group.dart index d99f515..6239fc2 100644 --- a/lib/src/stream_group.dart +++ b/lib/src/stream_group.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_group; - import 'dart:async'; /// A collection of streams whose events are unified and sent through a central
diff --git a/lib/src/stream_queue.dart b/lib/src/stream_queue.dart index 09b3a75..8482e0c 100644 --- a/lib/src/stream_queue.dart +++ b/lib/src/stream_queue.dart
@@ -2,14 +2,12 @@ // 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. -library async.stream_events; - import 'dart:async'; import 'dart:collection'; +import "result.dart"; import "subscription_stream.dart"; import "stream_completer.dart"; -import "../result.dart"; /// An asynchronous pull-based interface for accessing stream events. ///
diff --git a/lib/src/stream_sink_completer.dart b/lib/src/stream_sink_completer.dart index 0b6dc46..b53a8d2 100644 --- a/lib/src/stream_sink_completer.dart +++ b/lib/src/stream_sink_completer.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_sink_completer; - import 'dart:async'; import 'null_stream_sink.dart';
diff --git a/lib/src/stream_sink_transformer.dart b/lib/src/stream_sink_transformer.dart index a8410ab..6cde04a 100644 --- a/lib/src/stream_sink_transformer.dart +++ b/lib/src/stream_sink_transformer.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_sink_transformer; - import 'dart:async'; import 'stream_sink_transformer/handler_transformer.dart';
diff --git a/lib/src/stream_sink_transformer/handler_transformer.dart b/lib/src/stream_sink_transformer/handler_transformer.dart index e5d8e3c..a20a3fa 100644 --- a/lib/src/stream_sink_transformer/handler_transformer.dart +++ b/lib/src/stream_sink_transformer/handler_transformer.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_sink_transformer.handler_transformer; - import 'dart:async'; import '../stream_sink_transformer.dart';
diff --git a/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart b/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart index b53f208..83d2b19 100644 --- a/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart +++ b/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart
@@ -2,8 +2,6 @@ // 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. -library async.stream_sink_transformer.stream_transformer_wrapper; - import 'dart:async'; import '../stream_sink_transformer.dart';
diff --git a/lib/src/stream_splitter.dart b/lib/src/stream_splitter.dart index ac401a7..ec648aa 100644 --- a/lib/src/stream_splitter.dart +++ b/lib/src/stream_splitter.dart
@@ -2,12 +2,10 @@ // 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. -library async.stream_splitter; - import 'dart:async'; -import '../result.dart'; import 'future_group.dart'; +import 'result.dart'; /// A class that splits a single source stream into an arbitrary number of /// (single-subscription) streams (called "branch") that emit the same events.
diff --git a/lib/src/stream_zip.dart b/lib/src/stream_zip.dart new file mode 100644 index 0000000..432cc22 --- /dev/null +++ b/lib/src/stream_zip.dart
@@ -0,0 +1,120 @@ +// 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"; + +/// A stream that combines the values of other streams. +/// +/// This emits lists of collected values from each input stream. The first list +/// contains the first value emitted by each stream, the second contrains the +/// second value, and so on. The lists have the same ordering as the iterable +/// passed to [new StreamZip]. +/// +/// Any errors from any of the streams are forwarded directly to this stream. +class StreamZip<T> extends Stream<List<T>> { + final Iterable<Stream<T>> _streams; + + StreamZip(Iterable<Stream<T>> streams) : _streams = streams; + + StreamSubscription<List<T>> listen(void onData(List data), { + Function onError, + void onDone(), + bool cancelOnError}) { + cancelOnError = identical(true, cancelOnError); + List<StreamSubscription> subscriptions = <StreamSubscription>[]; + StreamController controller; + List current; + int dataCount = 0; + + /// Called for each data from a subscription in [subscriptions]. + void handleData(int index, data) { + current[index] = data; + dataCount++; + if (dataCount == subscriptions.length) { + List data = current; + current = new List(subscriptions.length); + dataCount = 0; + for (int i = 0; i < subscriptions.length; i++) { + if (i != index) subscriptions[i].resume(); + } + controller.add(data); + } else { + subscriptions[index].pause(); + } + } + + /// Called for each error from a subscription in [subscriptions]. + /// Except if [cancelOnError] is true, in which case the function below + /// is used instead. + void handleError(Object error, StackTrace stackTrace) { + controller.addError(error, stackTrace); + } + + /// Called when a subscription has an error and [cancelOnError] is true. + /// + /// Prematurely cancels all subscriptions since we know that we won't + /// be needing any more values. + void handleErrorCancel(Object error, StackTrace stackTrace) { + for (int i = 0; i < subscriptions.length; i++) { + subscriptions[i].cancel(); + } + controller.addError(error, stackTrace); + } + + void handleDone() { + for (int i = 0; i < subscriptions.length; i++) { + subscriptions[i].cancel(); + } + controller.close(); + } + + try { + for (Stream stream in _streams) { + int index = subscriptions.length; + subscriptions.add(stream.listen( + (data) { handleData(index, data); }, + onError: cancelOnError ? handleError : handleErrorCancel, + onDone: handleDone, + cancelOnError: cancelOnError)); + } + } catch (e) { + for (int i = subscriptions.length - 1; i >= 0; i--) { + subscriptions[i].cancel(); + } + rethrow; + } + + current = new List(subscriptions.length); + + controller = new StreamController<List>( + onPause: () { + for (int i = 0; i < subscriptions.length; i++) { + // This may pause some subscriptions more than once. + // These will not be resumed by onResume below, but must wait for the + // next round. + subscriptions[i].pause(); + } + }, + onResume: () { + for (int i = 0; i < subscriptions.length; i++) { + subscriptions[i].resume(); + } + }, + onCancel: () { + for (int i = 0; i < subscriptions.length; i++) { + // Canceling more than once is safe. + subscriptions[i].cancel(); + } + } + ); + + if (subscriptions.isEmpty) { + controller.close(); + } + return controller.stream.listen(onData, + onError: onError, + onDone: onDone, + cancelOnError: cancelOnError); + } +}
diff --git a/lib/src/subscription_stream.dart b/lib/src/subscription_stream.dart index bdf329b..4f58c76 100644 --- a/lib/src/subscription_stream.dart +++ b/lib/src/subscription_stream.dart
@@ -2,8 +2,6 @@ // 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. -library async.subscription_stream; - import 'dart:async'; import "delegate/stream_subscription.dart";
diff --git a/lib/stream_zip.dart b/lib/stream_zip.dart index 055489d..cdeb5c9 100644 --- a/lib/stream_zip.dart +++ b/lib/stream_zip.dart
@@ -2,118 +2,8 @@ // 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. -/** - * Help for combining multiple streams into a single stream. - */ +/// Import `async.dart` instead. +@Deprecated("Will be removed in async 2.0.0.") library dart.pkg.async.stream_zip; -import "dart:async"; - -/** - * A stream that combines the values of other streams. - */ -class StreamZip extends Stream<List> { - final Iterable<Stream> _streams; - StreamZip(Iterable<Stream> streams) : _streams = streams; - - StreamSubscription<List> listen(void onData(List data), { - Function onError, - void onDone(), - bool cancelOnError}) { - cancelOnError = identical(true, cancelOnError); - List<StreamSubscription> subscriptions = <StreamSubscription>[]; - StreamController controller; - List current; - int dataCount = 0; - - /// Called for each data from a subscription in [subscriptions]. - void handleData(int index, data) { - current[index] = data; - dataCount++; - if (dataCount == subscriptions.length) { - List data = current; - current = new List(subscriptions.length); - dataCount = 0; - for (int i = 0; i < subscriptions.length; i++) { - if (i != index) subscriptions[i].resume(); - } - controller.add(data); - } else { - subscriptions[index].pause(); - } - } - - /// Called for each error from a subscription in [subscriptions]. - /// Except if [cancelOnError] is true, in which case the function below - /// is used instead. - void handleError(Object error, StackTrace stackTrace) { - controller.addError(error, stackTrace); - } - - /// Called when a subscription has an error and [cancelOnError] is true. - /// - /// Prematurely cancels all subscriptions since we know that we won't - /// be needing any more values. - void handleErrorCancel(Object error, StackTrace stackTrace) { - for (int i = 0; i < subscriptions.length; i++) { - subscriptions[i].cancel(); - } - controller.addError(error, stackTrace); - } - - void handleDone() { - for (int i = 0; i < subscriptions.length; i++) { - subscriptions[i].cancel(); - } - controller.close(); - } - - try { - for (Stream stream in _streams) { - int index = subscriptions.length; - subscriptions.add(stream.listen( - (data) { handleData(index, data); }, - onError: cancelOnError ? handleError : handleErrorCancel, - onDone: handleDone, - cancelOnError: cancelOnError)); - } - } catch (e) { - for (int i = subscriptions.length - 1; i >= 0; i--) { - subscriptions[i].cancel(); - } - rethrow; - } - - current = new List(subscriptions.length); - - controller = new StreamController<List>( - onPause: () { - for (int i = 0; i < subscriptions.length; i++) { - // This may pause some subscriptions more than once. - // These will not be resumed by onResume below, but must wait for the - // next round. - subscriptions[i].pause(); - } - }, - onResume: () { - for (int i = 0; i < subscriptions.length; i++) { - subscriptions[i].resume(); - } - }, - onCancel: () { - for (int i = 0; i < subscriptions.length; i++) { - // Canceling more than once is safe. - subscriptions[i].cancel(); - } - } - ); - - if (subscriptions.isEmpty) { - controller.close(); - } - return controller.stream.listen(onData, - onError: onError, - onDone: onDone, - cancelOnError: cancelOnError); - } -} +export "src/stream_zip.dart";
diff --git a/pubspec.yaml b/pubspec.yaml index 7560836..2430d75 100644 --- a/pubspec.yaml +++ b/pubspec.yaml
@@ -1,5 +1,5 @@ name: async -version: 1.8.0 +version: 1.9.0 author: Dart Team <misc@dartlang.org> description: Utility functions and classes related to the 'dart:async' library. homepage: https://www.github.com/dart-lang/async
diff --git a/test/result_test.dart b/test/result_test.dart index 848c455..7af3fcf 100644 --- a/test/result_test.dart +++ b/test/result_test.dart
@@ -5,7 +5,7 @@ import "dart:async"; import "dart:collection"; -import "package:async/result.dart"; +import "package:async/async.dart"; import "package:stack_trace/stack_trace.dart"; import "package:test/test.dart";
diff --git a/test/stream_group_test.dart b/test/stream_group_test.dart index db6c938..5078dad 100644 --- a/test/stream_group_test.dart +++ b/test/stream_group_test.dart
@@ -2,8 +2,6 @@ // 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. -library async.test.stream_group_test; - import 'dart:async'; import 'package:async/async.dart';
diff --git a/test/stream_zip_test.dart b/test/stream_zip_test.dart index 35ace7d..47f3d8d 100644 --- a/test/stream_zip_test.dart +++ b/test/stream_zip_test.dart
@@ -3,7 +3,8 @@ // BSD-style license that can be found in the LICENSE file. import "dart:async"; -import "package:async/stream_zip.dart"; + +import "package:async/async.dart"; import "package:test/test.dart"; /// Create an error with the same values as [base], except that it throwsA
diff --git a/test/utils.dart b/test/utils.dart index 445b9fc..c32ea2c 100644 --- a/test/utils.dart +++ b/test/utils.dart
@@ -3,8 +3,6 @@ // BSD-style license that can be found in the LICENSE file. /// Helper utilities for testing. -library async.test.util; - import "dart:async"; import "package:async/async.dart";