| OLD | NEW |
| (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 } | |
| OLD | NEW |