From 7fc981b3a687283dbab318ce1bbf7b2dbb979f08 Mon Sep 17 00:00:00 2001 From: "lrn@google.com" Date: Tue, 29 Jan 2013 08:43:15 +0000 Subject: [PATCH] Add public-facing method and class that allows intercepting stream events. This allows intercepting events at the subscription level instead of creating a new full stream and adding the events to that. Also fix a number of typos and bugs detected by analyzer. Review URL: https://codereview.chromium.org//11953103 git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@17746 260f80e4-7a28-3924-810f-c04153c831b5 --- sdk/lib/async/future.dart | 2 +- sdk/lib/async/merge_stream.dart | 11 +- sdk/lib/async/stream.dart | 210 +++++++++++++++++- sdk/lib/async/stream_impl.dart | 16 +- sdk/lib/async/stream_pipe.dart | 171 ++++++++++---- tests/lib/async/stream_controller_test.dart | 36 +++ .../async/stream_event_transform_test.dart | 77 +++++++ 7 files changed, 461 insertions(+), 62 deletions(-) create mode 100644 tests/lib/async/stream_event_transform_test.dart diff --git a/sdk/lib/async/future.dart b/sdk/lib/async/future.dart index 0c4daa08955..f2d92644073 100644 --- a/sdk/lib/async/future.dart +++ b/sdk/lib/async/future.dart @@ -113,7 +113,7 @@ abstract class Future { * See [Completer]s, for futures with values that are computed asynchronously. */ factory Future.delayed(int milliseconds, T value()) { - _FutureImpl future = new _ThenFuture((_) => value()); + _ThenFuture future = new _ThenFuture((_) => value()); new Timer(milliseconds, (_) => future._sendValue(null)); return future; } diff --git a/sdk/lib/async/merge_stream.dart b/sdk/lib/async/merge_stream.dart index 05f5b061170..4625b4f8aa9 100644 --- a/sdk/lib/async/merge_stream.dart +++ b/sdk/lib/async/merge_stream.dart @@ -159,29 +159,34 @@ class _CycleEntry { Stream source; /** The active subscription, if any. */ StreamSubscription subscription = null; + /** Whether the subscription is currently paused. */ + bool isPaused = false; /** Next entry in a linked list of entries. */ _CycleEntry next; _CycleEntry(this.stream, this.source); void cancel() { - // This method may be called event if this entry has never been activated. + // This method may be called even if this entry has never been activated. if (subscription != null) { subscription.cancel(); subscription = null; + isPaused = false; } } void pause() { ensureSubscribed(); - if (!subscription.isPaused) { + if (!isPaused) { subscription.pause(); + isPaused = true; } } void activate() { ensureSubscribed(); - if (subscription.isPaused) { + if (isPaused) { + isPaused = false; subscription.resume(); } } diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart index 382b3cc3c50..8b63b496cdd 100644 --- a/sdk/lib/async/stream.dart +++ b/sdk/lib/async/stream.dart @@ -40,12 +40,11 @@ part of dart.async; * A broadcast stream allows any number of listeners, and it fires * its events when they are ready, whether there are listeners or not. * - * Braodcast streams are used for independent events/observers. + * Broadcast streams are used for independent events/observers. * - * The default implementation of [isBroadcast] and - * [asBroadcastStream] are assuming this is a single-subscription stream - * and a broadcast stream inheriting from [Stream] must override these - * to return [:true:] and [:this:] respectively. + * The default implementation of [isBroadcast] returns false. + * A broadcast stream inheriting from [Stream] must override [isBroadcast] + * to return [:true:]. */ abstract class Stream { Stream(); @@ -93,6 +92,7 @@ abstract class Stream { * If this stream is already a broadcast stream, it is returned unmodified. */ Stream asBroadcastStream() { + if (isBroadcast) return this; return new _SingleStreamMultiplexer(this); } @@ -172,8 +172,11 @@ abstract class Stream { * If the error is intercepted, the [handle] function can decide what to do * with it. It can throw if it wants to raise a new (or the same) error, * or simply return to make the stream forget the error. + * + * If you need to transform an error into a data event, use the more generic + * [Stream.transformEvent] to handle the event by writing a data event to + * the output sink */ - // TODO(lrn): Say what to do if you want to convert the error to a value. Stream handleError(void handle(AsyncError error), { bool test(error) }) { return new _HandleErrorStream(this, handle, test); } @@ -206,6 +209,26 @@ abstract class Stream { return streamTransformer.bind(this); } + /** + * Create a new stream from this by modifying events. + * + * Subscribing on the returned stream is the same as subscribing on + * this stream, except that events are passed through the [transformer] + * before being emitted. The transformer may generate any number and + * types of events for each incoming event. Pauses on the returned + * subscription are pauses on this stream. + * + * An example that duplicates all data events: + * + * someStream.transformEvents(new StreamEventTransformer.from( + * handleData: (var value, StreamSink sink) { + * sink.add(value); + * sink.add(value); + * })); + */ + Stream transformEvents(StreamEventTransformer transformer) { + return new EventTransformStream(this, transformer); + } /** Reduces a sequence of values by repeatedly applying [combine]. */ Future reduce(var initialValue, combine(var previous, T element)) { @@ -263,11 +286,11 @@ abstract class Stream { // checked mode. http://dartbug.com/7733 (/*T*/ element) { _runUserCode( - () => match(element), + () => (element == match), (bool isMatch) { if (isMatch) { subscription.cancel(); - future._setValue(element); + future._setValue(true); } }, _cancelAndError(subscription, future) @@ -934,3 +957,174 @@ abstract class StreamTransformer { Stream bind(Stream stream); } + + +/** + * A transformer of stream events. + * + * A [StreamEventTransformer] transforms incoming Stream + * events of one kind into outgoing events of another kind. + * + * The default implementations of the "handle" methods forward + * the events unmodified. In that case the generic type [T] needs to be + * assignable to [S]. + * + * You can use a [StreamEventTransformer] to modify a Stream's events using + * the [Stream.transformEvents] method. + */ +abstract class StreamEventTransformer { + const StreamEventTransformer(); + + /** + * Create a [StreamEventTransformer] that delegates to the provided methods. + * + * The created transformer acts as if the provided functions were the + * methods of the same name. + */ + factory StreamEventTransformer.from({ + void handleData(S data, StreamSink sink), + void handleError(AsyncError error, StreamSink sink), + void handleDone(StreamSink sink) + }) { + return new _StreamEventTransformerImpl(handleData, + handleError, + handleDone); + } + + + /** + * Act on incoming data event. + * + * The method may generate any number of events on the sink, but should + * not throw. + */ + void handleData(S event, StreamSink sink) { + var data = event; + sink.add(data); + } + + /** + * Act on incoming error event. + * + * The method may generate any number of events on the sink, but should + * not throw. + */ + void handleError(AsyncError error, StreamSink sink) { + sink.signalError(error); + } + + /** + * Act on incoming done event. + * + * The method may generate any number of events on the sink, but should + * not throw. + */ + void handleDone(StreamSink sink){ + sink.close(); + } +} + +/** + * Stream that transforms another stream by intercepting and replacing events. + * + * This [Stream] is a transformation of a source stream. Listening on this + * stream is the same as listening on the source stream, except that events + * are intercepted and modified by a [StreamEventTransformer] before becoming + * events on this stream. + */ +class EventTransformStream extends Stream { + Stream _source; + StreamEventTransformer _transformer; + EventTransformStream(Stream source, + StreamEventTransformer transformer) + : _source = source, _transformer = transformer; + + StreamSubscription listen(void onData(T data), + { void onError(AsyncError error), + void onDone(), + bool unsubscribeOnError }) { + return new _EventTransformStreamSubscription(_source, _transformer, + onData, onError, onDone, + unsubscribeOnError); + } +} + +class _EventTransformStreamSubscription + extends _BaseStreamSubscription + implements _StreamOutputSink { + /** The transformer used to transform events. */ + final StreamEventTransformer _transformer; + /** Whether to unsubscribe when emitting an error. */ + final bool _unsubscribeOnError; + /** Source of incoming events. */ + StreamSubscription _subscription; + /** Cached StreamSink wrapper for this class. */ + StreamSink _sink; + + _EventTransformStreamSubscription(Stream source, + this._transformer, + void onData(T data), + void onError(AsyncError error), + void onDone(), + this._unsubscribeOnError) + : super(onData, onError, onDone) { + _sink = new _StreamOutputSinkWrapper(this); + _subscription = source.listen(_handleData, + onError: _handleError, + onDone: _handleDone); + } + + void pause([Future pauseSignal]) { + if (_subscription != null) _subscription.pause(pauseSignal); + } + + void resume() { + if (_subscription != null) _subscription.resume(); + } + + void cancel() { + if (_subscription != null) { + _subscription.cancel(); + _subscription = null; + } + } + + void _handleData(S data) { + _transformer.handleData(data, _sink); + } + + void _handleError(AsyncError error) { + _transformer.handleError(error, _sink); + } + + void _handleDone() { + _transformer.handleDone(_sink); + } + + // StreamOutputSink interface. + void _sendData(T data) { + _onData(data); + } + + void _sendError(AsyncError error) { + _onError(error); + if (_unsubscribeOnError) { + cancel(); + } + } + + void _sendDone() { + // It's ok to cancel even if we have been unsubscribed already. + cancel(); + _onDone(); + } +} + +class _StreamOutputSinkWrapper implements StreamSink { + _StreamOutputSink _sink; + _StreamOutputSinkWrapper(this._sink); + + void add(T data) => _sink._sendData(data); + void signalError(AsyncError error) => _sink._sendError(error); + void close() => _sink._sendDone(); +} diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart index 28c41b09804..caa6318f858 100644 --- a/sdk/lib/async/stream_impl.dart +++ b/sdk/lib/async/stream_impl.dart @@ -62,10 +62,10 @@ abstract class _StreamImpl extends Stream { // ------------------------------------------------------------------ // Stream interface. - StreamSubscription listen(void onData(T data), - { void onError(AsyncError error), - void onDone(), - bool unsubscribeOnError }) { + StreamSubscription listen(void onData(T data), + { void onError(AsyncError error), + void onDone(), + bool unsubscribeOnError }) { if (_isComplete) { return new _DoneSubscription(onDone); } @@ -73,7 +73,7 @@ abstract class _StreamImpl extends Stream { if (onError == null) onError = _nullErrorHandler; if (onDone == null) onDone = _nullDoneHandler; unsubscribeOnError = identical(true, unsubscribeOnError); - _StreamListener subscription = + _StreamSubscriptionImpl subscription = _createSubscription(onData, onError, onDone, unsubscribeOnError); _addListener(subscription); return subscription; @@ -1092,8 +1092,10 @@ class _DoneSubscription implements StreamSubscription { bool get _isComplete => _timer == null && _pauseCount == 0; void onData(void handleAction(T value)) {} - void onError(void handleError(StateError error)) {} - void onDone(void handleDone(T value)) { + + void onError(void handleError(AsyncError error)) {} + + void onDone(void handleDone()) { _handler = handleDone; } diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart index 59cf3829bec..895e3313b51 100644 --- a/sdk/lib/async/stream_pipe.dart +++ b/sdk/lib/async/stream_pipe.dart @@ -53,20 +53,23 @@ abstract class _ForwardingStream extends Stream { bool get isBroadcast => _source.isBroadcast; - bool asBroadcastStream() => _source.asBroadcastStream; - - StreamSubscription listen(void onData(T value), - { void onError(AsyncError error), - void onDone(), - bool unsubscribeOnError }) { + StreamSubscription listen(void onData(T value), + { void onError(AsyncError error), + void onDone(), + bool unsubscribeOnError }) { if (onData == null) onData = _nullDataHandler; if (onError == null) onError = _nullErrorHandler; if (onDone == null) onDone = _nullDoneHandler; unsubscribeOnError = identical(true, unsubscribeOnError); - StreamSubscription subscription = - new _ForwardingStreamSubscription( - this, onData, onError, onDone, unsubscribeOnError); - return subscription; + return _createSubscription(onData, onError, onDone, unsubscribeOnError); + } + + StreamSubscription _createSubscription(void onData(T value), + void onError(AsyncError error), + void onDone(), + bool unsubscribeOnError) { + return new _ForwardingStreamSubscription( + this, onData, onError, onDone, unsubscribeOnError); } // Override the following methods in subclasses to change the behavior. @@ -86,33 +89,26 @@ abstract class _ForwardingStream extends Stream { } /** - * Abstract superclass for subscriptions that forward to other subscriptions. + * Common behavior of [StreamSubscription] classes. + * + * Stores and allows updating of the event handlers of a [StreamSubscription]. */ -class _ForwardingStreamSubscription - implements StreamSubscription, _StreamOutputSink { - final _ForwardingStream _stream; +abstract class _BaseStreamSubscription implements StreamSubscription { // TODO(ahe): Restore type when feature is implemented in dart2js // checked mode. http://dartbug.com/7733 var /* _DataHandler */ _onData; _ErrorHandler _onError; _DoneHandler _onDone; - StreamSubscription _subscription; - - _ForwardingStreamSubscription(this._stream, - this._onData, - this._onError, - this._onDone, - bool unsubscribeOnError) { - _subscription = - _stream._source.listen(_handleData, - onError: _handleError, - onDone: _handleDone, - unsubscribeOnError: unsubscribeOnError); + _BaseStreamSubscription(this._onData, + this._onError, + this._onDone) { + if (_onData == null) _onData = _nullDataHandler; + if (_onError == null) _onError = _nullErrorHandler; + if (_onDone == null) _onDone = _nullDoneHandler; } // StreamSubscription interface. - void onData(void handleData(T event)) { if (handleData == null) handleData = _nullDataHandler; _onData = handleData; @@ -128,6 +124,39 @@ class _ForwardingStreamSubscription _onDone = handleDone; } + void pause([Future resumeSignal]); + + void resume(); + + void cancel(); +} + + +/** + * Abstract superclass for subscriptions that forward to other subscriptions. + */ +class _ForwardingStreamSubscription + extends _BaseStreamSubscription implements _StreamOutputSink { + final _ForwardingStream _stream; + final bool _unsubscribeOnError; + + StreamSubscription _subscription; + + _ForwardingStreamSubscription(this._stream, + void onData(T data), + void onError(AsyncError error), + void onDone(), + this._unsubscribeOnError) + : super(onData, onError, onDone) { + // Don't unsubscribe on incoming error, only if we send an error forwards. + _subscription = + _stream._source.listen(_handleData, + onError: _handleError, + onDone: _handleDone); + } + + // StreamSubscription interface. + void pause([Future resumeSignal]) { if (_subscription == null) { throw new StateError("Subscription has been unsubscribed"); @@ -158,6 +187,10 @@ class _ForwardingStreamSubscription void _sendError(AsyncError error) { _onError(error); + if (_unsubscribeOnError) { + _subscription.cancel(); + _subscription = null; + } } void _sendDone() { @@ -428,11 +461,28 @@ class _DistinctStream extends _ForwardingStream { } } +// Stream transformations and event transformations. typedef void _TransformDataHandler(S data, StreamSink sink); typedef void _TransformErrorHandler(AsyncError data, StreamSink sink); typedef void _TransformDoneHandler(StreamSink sink); +/** Default data handler forwards all data. */ +void _defaultHandleData(var data, StreamSink sink) { + sink.add(data); +} + +/** Default error handler forwards all errors. */ +void _defaultHandleError(AsyncError error, StreamSink sink) { + sink.signalError(error); +} + +/** Default done handler forwards done. */ +void _defaultHandleDone(StreamSink sink) { + sink.close(); +} + + /** * A stream transformer that intercepts all events and can generate any event as * output. @@ -467,7 +517,7 @@ class _StreamTransformerImpl implements StreamTransformer { try { _onData(data, _sink); } catch (e, s) { - _stream._signalError(_asyncError(e, s)); + _sink.signalError(_asyncError(e, s)); } } @@ -475,7 +525,7 @@ class _StreamTransformerImpl implements StreamTransformer { try { _onError(error, _sink); } catch (e, s) { - _stream._signalError(_asyncError(e, s, error)); + _sink.signalError(_asyncError(e, s, error)); } } @@ -483,22 +533,9 @@ class _StreamTransformerImpl implements StreamTransformer { try { _onDone(_sink); } catch (e, s) { - _stream._signalError(_asyncError(e, s)); + _sink.signalError(_asyncError(e, s)); } } - - /** Default data handler forwards all data. */ - static void _defaultHandleData(var data, StreamSink sink) { - sink.add(data); - } - /** Default error handler forwards all errors. */ - static void _defaultHandleError(AsyncError error, StreamSink sink) { - sink.signalError(error); - } - /** Default done handler forwards done. */ - static void _defaultHandleDone(StreamSink sink) { - sink.close(); - } } /** Creates a [StreamSink] from a [_StreamImpl]'s input methods. */ @@ -509,3 +546,51 @@ class _StreamImplSink implements StreamSink { void signalError(AsyncError error) { _target._signalError(error); } void close() { _target._close(); } } + + +/** + * A stream transformer that intercepts all events and can generate any event as + * output. + * + * Each incoming event on the source stream is passed to the corresponding + * provided event handler, along with a [StreamSink] linked to the output + * Stream. + * The handler can then decide exactly which events to send to the output. + */ +class _StreamEventTransformerImpl + implements StreamEventTransformer { + final _TransformDataHandler _handleData; + final _TransformErrorHandler _handleError; + final _TransformDoneHandler _handleDone; + + _StreamEventTransformerImpl(void onData(S data, StreamSink sink), + void onError(AsyncError data, StreamSink sink), + void onDone(StreamSink sink)) + : this._handleData = (onData == null ? _defaultHandleData : onData), + this._handleError = (onError == null ? _defaultHandleError : onError), + this._handleDone = (onDone == null ? _defaultHandleDone : onDone); + + void handleData(S data, StreamSink sink) { + try { + _handleData(data, sink); + } catch (e, s) { + sink.signalError(_asyncError(e, s)); + } + } + + void handleError(AsyncError error, StreamSink sink) { + try { + _handleError(error, sink); + } catch (e, s) { + sink.signalError(_asyncError(e, s, error)); + } + } + + void handleDone(StreamSink sink) { + try { + _handleDone(sink); + } catch (e, s) { + sink.signalError(_asyncError(e, s)); + } + } +} diff --git a/tests/lib/async/stream_controller_test.dart b/tests/lib/async/stream_controller_test.dart index 9dd7509fad0..57b99e3699b 100644 --- a/tests/lib/async/stream_controller_test.dart +++ b/tests/lib/async/stream_controller_test.dart @@ -240,6 +240,42 @@ testSingleController() { c.add(9); c.close(); + // test contains. + { + c = new StreamController(); + // Error after match is not important. + sentEvents = new Events()..add("a")..add("x")..error("FAIL")..close(); + Future contains = c.stream.contains("x"); + contains.then((var c) { + Expect.isTrue(c); + }); + sentEvents.replay(c); + } + + { + c = new StreamController(); + // Not matching is ok. + sentEvents = new Events()..add("a")..add("x")..add("b")..close(); + Future contains = c.stream.contains("y"); + contains.then((var c) { + Expect.isFalse(c); + }); + sentEvents.replay(c); + } + + { + c = new StreamController(); + // Error before match makes future err. + sentEvents = new Events()..add("a")..error("FAIL")..add("b")..close(); + Future contains = c.stream.contains("b"); + contains.then((var c) { + Expect.fail("no value expected"); + }).catchError((AsyncError e) { + Expect.equals("FAIL", e.error); + }); + sentEvents.replay(c); + } + // Test transform. c = new StreamController(); sentEvents = new Events()..add("a")..error(42)..add("b")..close(); diff --git a/tests/lib/async/stream_event_transform_test.dart b/tests/lib/async/stream_event_transform_test.dart new file mode 100644 index 00000000000..3acd3f0b15e --- /dev/null +++ b/tests/lib/async/stream_event_transform_test.dart @@ -0,0 +1,77 @@ +// Copyright (c) 2011, 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 '../../../pkg/unittest/lib/unittest.dart'; +import 'event_helper.dart'; + +void handleData(int data, StreamSink sink) { + sink.signalError(new AsyncError("$data")); + sink.add(data + 1); +} +void handleError(AsyncError e, StreamSink sink) { + String value = e.error; + int data = int.parse(value); + sink.add(data); + sink.signalError(new AsyncError("${data + 1}")); +} + +void handleDone(StreamSink sink) { + sink.add(99); + sink.close(); +} + +class EventTransformer extends StreamEventTransformer { + void handleData(int data, StreamSink sink) { + sink.signalError(new AsyncError("$data")); + sink.add(data + 1); + } + void handleError(AsyncError e, StreamSink sink) { + String value = e.error; + int data = int.parse(value); + sink.add(data); + sink.signalError(new AsyncError("${data + 1}")); + } + + void handleDone(StreamSink sink) { + sink.add(99); + sink.close(); + } +} + +main() { + { + StreamController c = new StreamController(); + Events expected = new Events()..error("0")..add(1) + ..error("1")..add(2) + ..add(3)..error("4") + ..add(99)..close(); + Events input = new Events()..add(0)..add(1)..error("3")..close(); + Events actual = new Events.capture( + c.stream.transformEvents(new EventTransformer())); + actual.onDone(() { + Expect.listEquals(expected.events, actual.events); + }); + input.replay(c); + } + + { + StreamController c = new StreamController(); + Events expected = new Events()..error("0")..add(1) + ..error("1")..add(2) + ..add(3)..error("4") + ..add(99)..close(); + Events input = new Events()..add(0)..add(1)..error("3")..close(); + Events actual = new Events.capture( + c.stream.transformEvents(new StreamEventTransformer.from( + handleData: handleData, + handleError: handleError, + handleDone: handleDone + ))); + actual.onDone(() { + Expect.listEquals(expected.events, actual.events); + }); + input.replay(c); + } +}