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

Side by Side Diff: sdk/lib/async/stream_controller.dart

Issue 12033019: Make StreamController take on{Pause,Subscription]StateChange as arguments. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 11 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 unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « no previous file | sdk/lib/async/stream_impl.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 part of dart.async; 5 part of dart.async;
6 6
7 // ------------------------------------------------------------------- 7 // -------------------------------------------------------------------
8 // Default implementation of a stream with a controller for adding 8 // Default implementation of a stream with a controller for adding
9 // events to the stream. 9 // events to the stream.
10 // ------------------------------------------------------------------- 10 // -------------------------------------------------------------------
11 11
12 /** 12 /**
13 * A controller and the stream it controls. 13 * A controller and the stream it controls.
14 * 14 *
15 * This controller allows sending data, error and done events on 15 * This controller allows sending data, error and done events on
16 * its [stream]. 16 * its [stream].
17 * This class can be used to create a simple stream that others 17 * This class can be used to create a simple stream that others
18 * can listen on, and to push events to that stream. 18 * can listen on, and to push events to that stream.
19 * 19 *
20 * It's possible to check whether the stream is paused or not, and whether 20 * It's possible to check whether the stream is paused or not, and whether
21 * it has subscribers or not, as well as getting a callback when either of 21 * it has subscribers or not, as well as getting a callback when either of
22 * these change. 22 * these change.
23 */ 23 */
24 class StreamController<T> extends Stream<T> implements StreamSink<T> { 24 class StreamController<T> extends Stream<T> implements StreamSink<T> {
25 _StreamImpl<T> _stream; 25 _StreamImpl<T> _stream;
26 Stream<T> get stream => _stream; 26 Stream<T> get stream => _stream;
27 27
28 /** 28 /**
29 * A controller with a [stream] that supports multiple subscribers. 29 * A controller with a [stream] that supports multiple subscribers.
30 *
31 * The [onPauseStateChange] function is called when the stream becomes
32 * paused or resumes after being paused. The current pause state can
33 * be read from [isPaused]. Ignored if [:null:].
34 *
35 * The [onSubscriptionStateChange] function is called when the stream
36 * receives its first listener or loses its last. The current subscription
37 * state can be read from [hasSubscribers]. Ignored if [:null:].
30 */ 38 */
31 StreamController.multiSubscription() { 39 StreamController.multiSubscription({void onPauseStateChange(),
40 void onSubscriptionStateChange()}) {
32 _stream = new _MultiControllerStream<T>(onSubscriptionStateChange, 41 _stream = new _MultiControllerStream<T>(onSubscriptionStateChange,
33 onPauseStateChange); 42 onPauseStateChange);
34 } 43 }
35 /** 44 /**
36 * A controller with a [stream] that supports only one single subscriber. 45 * A controller with a [stream] that supports only one single subscriber.
37 * The controller will buffer all incoming events until the subscriber is 46 * The controller will buffer all incoming events until the subscriber is
38 * registered. 47 * registered.
48 *
49 * The [onPauseStateChange] function is called when the stream becomes
50 * paused or resumes after being paused. The current pause state can
51 * be read from [isPaused]. Ignored if [:null:].
52 *
53 * The [onSubscriptionStateChange] function is called when the stream
54 * receives its first listener or loses its last. The current subscription
55 * state can be read from [hasSubscribers]. Ignored if [:null:].
39 */ 56 */
40 StreamController() { 57 StreamController({void onPauseStateChange(),
58 void onSubscriptionStateChange()}) {
41 _stream = new _SingleControllerStream<T>(onSubscriptionStateChange, 59 _stream = new _SingleControllerStream<T>(onSubscriptionStateChange,
42 onPauseStateChange); 60 onPauseStateChange);
43 } 61 }
44 62
45 bool get isSingleSubscription => _stream.isSingleSubscription; 63 bool get isSingleSubscription => _stream.isSingleSubscription;
46 64
47 Stream<T> asMultiSubscriptionStream() => _stream.asMultiSubscriptionStream(); 65 Stream<T> asMultiSubscriptionStream() => _stream.asMultiSubscriptionStream();
48 66
49 StreamSubscription listen(void onData(T data), 67 StreamSubscription listen(void onData(T data),
50 { void onError(AsyncError error), 68 { void onError(AsyncError error),
(...skipping 44 matching lines...) Expand 10 before | Expand all | Expand 10 after
95 } 113 }
96 114
97 /** 115 /**
98 * Send or enqueue a "done" message. 116 * Send or enqueue a "done" message.
99 * 117 *
100 * The "done" message should be sent at most once by a stream, and it 118 * The "done" message should be sent at most once by a stream, and it
101 * should be the last message sent. 119 * should be the last message sent.
102 */ 120 */
103 void close() { _stream._close(); } 121 void close() { _stream._close(); }
104 122
105 /**
106 * Called when the first subscriber requests a pause or the last a resume.
107 *
108 * Read [isPaused] to see the new state.
109 */
110 void onPauseStateChange() {}
111
112 /**
113 * Called when the first listener subscribes or the last unsubscribes.
114 *
115 * Read [hasSubscribers] to see what the new state is.
116 */
117 void onSubscriptionStateChange() {}
118
119 void forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription)) { 123 void forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription)) {
120 _stream._forEachSubscriber(() { 124 _stream._forEachSubscriber(() {
121 try { 125 try {
122 action(); 126 action();
123 } on AsyncError catch (e) { 127 } on AsyncError catch (e) {
124 e.throwDelayed(); 128 e.throwDelayed();
125 } catch (e, s) { 129 } catch (e, s) {
126 new AsyncError(e, s).throwDelayed(); 130 new AsyncError(e, s).throwDelayed();
127 } 131 }
128 }); 132 });
129 } 133 }
130 } 134 }
131 135
132 typedef void _NotificationHandler(); 136 typedef void _NotificationHandler();
133 137
134 class _MultiControllerStream<T> extends _MultiStreamImpl<T> { 138 class _MultiControllerStream<T> extends _MultiStreamImpl<T> {
135 _NotificationHandler _subscriptionHandler; 139 _NotificationHandler _subscriptionHandler;
136 _NotificationHandler _pauseHandler; 140 _NotificationHandler _pauseHandler;
137 141
138 _MultiControllerStream(this._subscriptionHandler, this._pauseHandler); 142 _MultiControllerStream(this._subscriptionHandler, this._pauseHandler);
139 143
140 void _onSubscriptionStateChange() { 144 void _onSubscriptionStateChange() {
141 _subscriptionHandler(); 145 if (_subscriptionHandler != null) _subscriptionHandler();
142 } 146 }
143 147
144 void _onPauseStateChange() { 148 void _onPauseStateChange() {
145 _pauseHandler(); 149 if (_pauseHandler != null) _pauseHandler();
146 } 150 }
147 } 151 }
148 152
149 class _SingleControllerStream<T> extends _SingleStreamImpl<T> { 153 class _SingleControllerStream<T> extends _SingleStreamImpl<T> {
150 _NotificationHandler _subscriptionHandler; 154 _NotificationHandler _subscriptionHandler;
151 _NotificationHandler _pauseHandler; 155 _NotificationHandler _pauseHandler;
152 156
153 _SingleControllerStream(this._subscriptionHandler, this._pauseHandler); 157 _SingleControllerStream(this._subscriptionHandler, this._pauseHandler);
154 158
155 void _onSubscriptionStateChange() { 159 void _onSubscriptionStateChange() {
156 _subscriptionHandler(); 160 if (_subscriptionHandler != null) _subscriptionHandler();
157 } 161 }
158 162
159 void _onPauseStateChange() { 163 void _onPauseStateChange() {
160 _pauseHandler(); 164 if (_pauseHandler != null) _pauseHandler();
161 } 165 }
162 } 166 }
OLDNEW
« no previous file with comments | « no previous file | sdk/lib/async/stream_impl.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698