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

Side by Side Diff: test/generated_sdk/lib/async/stream_transformers.dart

Issue 1162723007: remove generated_sdk from checked in code (Closed) Base URL: git@github.com:dart-lang/dev_compiler.git@master
Patch Set: Created 5 years, 6 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
« no previous file with comments | « test/generated_sdk/lib/async/stream_pipe.dart ('k') | test/generated_sdk/lib/async/timer.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
(Empty)
1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
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.
4
5 part of dart.async;
6
7 /**
8 * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
9 */
10 class _EventSinkWrapper<T> implements EventSink<T> {
11 _EventSink _sink;
12 _EventSinkWrapper(this._sink);
13
14 void add(T data) { _sink._add(data); }
15 void addError(error, [StackTrace stackTrace]) {
16 _sink._addError(error, stackTrace);
17 }
18 void close() { _sink._close(); }
19 }
20
21 /**
22 * A StreamSubscription that pipes data through a sink.
23 *
24 * The constructor of this class takes a [_SinkMapper] which maps from
25 * [EventSink] to [EventSink]. The input to the mapper is the output of
26 * the transformation. The returned sink is the transformation's input.
27 */
28 class _SinkTransformerStreamSubscription<S, T>
29 extends _BufferingStreamSubscription<T> {
30 /// The transformer's input sink.
31 EventSink _transformerSink;
32
33 /// The subscription to the input stream.
34 StreamSubscription<S> _subscription;
35
36 _SinkTransformerStreamSubscription(Stream<S> source,
37 _SinkMapper<S, T> mapper,
38 void onData(T data),
39 Function onError,
40 void onDone(),
41 bool cancelOnError)
42 // We set the adapter's target only when the user is allowed to send data.
43 : super(onData, onError, onDone, cancelOnError) {
44 _EventSinkWrapper<T> eventSink = new _EventSinkWrapper<T>(this);
45 _transformerSink = mapper(eventSink);
46 _subscription = source.listen(_handleData,
47 onError: _handleError,
48 onDone: _handleDone);
49 }
50
51 /** Whether this subscription is still subscribed to its source. */
52 bool get _isSubscribed => _subscription != null;
53
54 // _EventSink interface.
55
56 /**
57 * Adds an event to this subscriptions.
58 *
59 * Contrary to normal [_BufferingStreamSubscription]s we may receive
60 * events when the stream is already closed. Report them as state
61 * error.
62 */
63 void _add(T data) {
64 if (_isClosed) {
65 throw new StateError("Stream is already closed");
66 }
67 super._add(data);
68 }
69
70 /**
71 * Adds an error event to this subscriptions.
72 *
73 * Contrary to normal [_BufferingStreamSubscription]s we may receive
74 * events when the stream is already closed. Report them as state
75 * error.
76 */
77 void _addError(Object error, StackTrace stackTrace) {
78 if (_isClosed) {
79 throw new StateError("Stream is already closed");
80 }
81 super._addError(error, stackTrace);
82 }
83
84 /**
85 * Adds a close event to this subscriptions.
86 *
87 * Contrary to normal [_BufferingStreamSubscription]s we may receive
88 * events when the stream is already closed. Report them as state
89 * error.
90 */
91 void _close() {
92 if (_isClosed) {
93 throw new StateError("Stream is already closed");
94 }
95 super._close();
96 }
97
98 // _BufferingStreamSubscription hooks.
99
100 void _onPause() {
101 if (_isSubscribed) _subscription.pause();
102 }
103
104 void _onResume() {
105 if (_isSubscribed) _subscription.resume();
106 }
107
108 Future _onCancel() {
109 if (_isSubscribed) {
110 StreamSubscription subscription = _subscription;
111 _subscription = null;
112 subscription.cancel();
113 }
114 return null;
115 }
116
117 void _handleData(S data) {
118 try {
119 _transformerSink.add(data);
120 } catch (e, s) {
121 _addError(e, s);
122 }
123 }
124
125 void _handleError(error, [stackTrace]) {
126 try {
127 _transformerSink.addError(error, stackTrace);
128 } catch (e, s) {
129 if (identical(e, error)) {
130 _addError(error, stackTrace);
131 } else {
132 _addError(e, s);
133 }
134 }
135 }
136
137 void _handleDone() {
138 try {
139 _subscription = null;
140 _transformerSink.close();
141 } catch (e, s) {
142 _addError(e, s);
143 }
144 }
145 }
146
147
148 typedef EventSink<S> _SinkMapper<S, T>(EventSink<T> output);
149
150 /**
151 * A StreamTransformer for Sink-mappers.
152 *
153 * A Sink-mapper takes an [EventSink] (its output) and returns another
154 * EventSink (its input).
155 *
156 * Note that this class can be `const`.
157 */
158 class _StreamSinkTransformer<S, T> implements StreamTransformer<S, T> {
159 final _SinkMapper<S, T> _sinkMapper;
160 const _StreamSinkTransformer(this._sinkMapper);
161
162 Stream<T> bind(Stream<S> stream)
163 => new _BoundSinkStream<S, T>(stream, _sinkMapper);
164 }
165
166 /**
167 * The result of binding a StreamTransformer for Sink-mappers.
168 *
169 * It contains the bound Stream and the sink-mapper. Only when the user starts
170 * listening to this stream is the sink-mapper invoked. The result is used
171 * to create a StreamSubscription that transforms events.
172 */
173 class _BoundSinkStream<S, T> extends Stream<T> {
174 final _SinkMapper<S, T> _sinkMapper;
175 final Stream<S> _stream;
176
177 bool get isBroadcast => _stream.isBroadcast;
178
179 _BoundSinkStream(this._stream, this._sinkMapper);
180
181 StreamSubscription<T> listen(void onData(T event),
182 { Function onError,
183 void onDone(),
184 bool cancelOnError }) {
185 cancelOnError = identical(true, cancelOnError);
186 StreamSubscription<T> subscription = new _SinkTransformerStreamSubscription(
187 _stream, _sinkMapper, onData, onError, onDone, cancelOnError);
188 return subscription;
189 }
190 }
191
192 /// Data-handler coming from [StreamTransformer.fromHandlers].
193 typedef void _TransformDataHandler<S, T>(S data, EventSink<T> sink);
194 /// Error-handler coming from [StreamTransformer.fromHandlers].
195 typedef void _TransformErrorHandler<T>(
196 Object error, StackTrace stackTrace, EventSink<T> sink);
197 /// Done-handler coming from [StreamTransformer.fromHandlers].
198 typedef void _TransformDoneHandler<T>(EventSink<T> sink);
199
200 /**
201 * Wraps handlers (from [StreamTransformer.fromHandlers]) into an `EventSink`.
202 *
203 * This way we can reuse the code from [_StreamSinkTransformer].
204 */
205 class _HandlerEventSink<S, T> implements EventSink<S> {
206 final _TransformDataHandler<S, T> _handleData;
207 final _TransformErrorHandler<T> _handleError;
208 final _TransformDoneHandler<T> _handleDone;
209
210 /// The output sink where the handlers should send their data into.
211 final EventSink<T> _sink;
212
213 _HandlerEventSink(this._handleData, this._handleError, this._handleDone,
214 this._sink);
215
216 void add(S data) => _handleData(data, _sink);
217 void addError(Object error, [StackTrace stackTrace])
218 => _handleError(error, stackTrace, _sink);
219 void close() => _handleDone(_sink);
220 }
221
222 /**
223 * A StreamTransformer that transformers events with the given handlers.
224 *
225 * Note that this transformer can only be used once.
226 */
227 class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> {
228
229 _StreamHandlerTransformer({
230 void handleData(S data, EventSink<T> sink),
231 void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
232 void handleDone(EventSink<T> sink)})
233 : super((EventSink<T> outputSink) {
234 if (handleData == null) handleData = _defaultHandleData;
235 if (handleError == null) handleError = _defaultHandleError;
236 if (handleDone == null) handleDone = _defaultHandleDone;
237 return new _HandlerEventSink<S, T>(
238 handleData, handleError, handleDone, outputSink);
239 });
240
241 Stream<T> bind(Stream<S> stream) {
242 return super.bind(stream);
243 }
244
245 /** Default data handler forwards all data. */
246 static void _defaultHandleData(var data, EventSink sink) {
247 sink.add(data);
248 }
249
250 /** Default error handler forwards all errors. */
251 static void _defaultHandleError(error, StackTrace stackTrace,
252 EventSink sink) {
253 sink.addError(error);
254 }
255
256 /** Default done handler forwards done. */
257 static void _defaultHandleDone(EventSink sink) {
258 sink.close();
259 }
260 }
261
262 /// A closure mapping a stream and cancelOnError to a StreamSubscription.
263 typedef StreamSubscription<T> _SubscriptionTransformer<S, T>(
264 Stream<S> stream, bool cancelOnError);
265
266 /**
267 * A [StreamTransformer] that minimizes the number of additional classes.
268 *
269 * Instead of implementing three classes: a [StreamTransformer], a [Stream]
270 * (as the result of a `bind` call) and a [StreamSubscription] (which does the
271 * actual work), this class only requires a fincution that is invoked when the
272 * last bit (the subscription) of the transformer-workflow is needed.
273 *
274 * The given transformer function maps from Stream and cancelOnError to a
275 * `StreamSubscription`. As such it can also act on `cancel` events, making it
276 * fully general.
277 */
278 class _StreamSubscriptionTransformer<S, T> implements StreamTransformer<S, T> {
279 final _SubscriptionTransformer<S, T> _transformer;
280
281 const _StreamSubscriptionTransformer(this._transformer);
282
283 Stream<T> bind(Stream<S> stream) =>
284 new _BoundSubscriptionStream<S, T>(stream, _transformer);
285 }
286
287 /**
288 * A stream transformed by a [_StreamSubscriptionTransformer].
289 *
290 * When this stream is listened to it invokes the [_transformer] function with
291 * the stored [_stream]. Usually the transformer starts listening at this
292 * moment.
293 */
294 class _BoundSubscriptionStream<S, T> extends Stream<T> {
295 final _SubscriptionTransformer<S, T> _transformer;
296 final Stream<S> _stream;
297
298 _BoundSubscriptionStream(this._stream, this._transformer);
299
300 StreamSubscription<T> listen(void onData(T event),
301 { Function onError,
302 void onDone(),
303 bool cancelOnError }) {
304 cancelOnError = identical(true, cancelOnError);
305 StreamSubscription<T> result = _transformer(_stream, cancelOnError);
306 result.onData(onData);
307 result.onError(onError);
308 result.onDone(onDone);
309 return result;
310 }
311 }
OLDNEW
« no previous file with comments | « test/generated_sdk/lib/async/stream_pipe.dart ('k') | test/generated_sdk/lib/async/timer.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698