| OLD | NEW |
| (Empty) |
| 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 | |
| 3 // BSD-style license that can be found in the LICENSE file. | |
| 4 | |
| 5 part of dart.async; | |
| 6 | |
| 7 /** Runs user code and takes actions depending on success or failure. */ | |
| 8 _runUserCode(userCode(), | |
| 9 onSuccess(value), | |
| 10 onError(error, StackTrace stackTrace)) { | |
| 11 try { | |
| 12 onSuccess(userCode()); | |
| 13 } catch (e, s) { | |
| 14 AsyncError replacement = Zone.current.errorCallback(e, s); | |
| 15 if (replacement == null) { | |
| 16 onError(e, s); | |
| 17 } else { | |
| 18 var error = _nonNullError(replacement.error); | |
| 19 var stackTrace = replacement.stackTrace; | |
| 20 onError(error, stackTrace); | |
| 21 } | |
| 22 } | |
| 23 } | |
| 24 | |
| 25 /** Helper function to cancel a subscription and wait for the potential future, | |
| 26 before completing with an error. */ | |
| 27 void _cancelAndError(StreamSubscription subscription, | |
| 28 _Future future, | |
| 29 error, | |
| 30 StackTrace stackTrace) { | |
| 31 var cancelFuture = subscription.cancel(); | |
| 32 if (cancelFuture is Future) { | |
| 33 cancelFuture.whenComplete(() => future._completeError(error, stackTrace)); | |
| 34 } else { | |
| 35 future._completeError(error, stackTrace); | |
| 36 } | |
| 37 } | |
| 38 | |
| 39 void _cancelAndErrorWithReplacement(StreamSubscription subscription, | |
| 40 _Future future, | |
| 41 error, StackTrace stackTrace) { | |
| 42 AsyncError replacement = Zone.current.errorCallback(error, stackTrace); | |
| 43 if (replacement != null) { | |
| 44 error = _nonNullError(replacement.error); | |
| 45 stackTrace = replacement.stackTrace; | |
| 46 } | |
| 47 _cancelAndError(subscription, future, error, stackTrace); | |
| 48 } | |
| 49 | |
| 50 /** Helper function to make an onError argument to [_runUserCode]. */ | |
| 51 _cancelAndErrorClosure(StreamSubscription subscription, _Future future) => | |
| 52 ((error, StackTrace stackTrace) => _cancelAndError( | |
| 53 subscription, future, error, stackTrace)); | |
| 54 | |
| 55 /** Helper function to cancel a subscription and wait for the potential future, | |
| 56 before completing with a value. */ | |
| 57 void _cancelAndValue(StreamSubscription subscription, _Future future, value) { | |
| 58 var cancelFuture = subscription.cancel(); | |
| 59 if (cancelFuture is Future) { | |
| 60 cancelFuture.whenComplete(() => future._complete(value)); | |
| 61 } else { | |
| 62 future._complete(value); | |
| 63 } | |
| 64 } | |
| 65 | |
| 66 | |
| 67 /** | |
| 68 * A [Stream] that forwards subscriptions to another stream. | |
| 69 * | |
| 70 * This stream implements [Stream], but forwards all subscriptions | |
| 71 * to an underlying stream, and wraps the returned subscription to | |
| 72 * modify the events on the way. | |
| 73 * | |
| 74 * This class is intended for internal use only. | |
| 75 */ | |
| 76 abstract class _ForwardingStream<S, T> extends Stream<T> { | |
| 77 final Stream<S> _source; | |
| 78 | |
| 79 _ForwardingStream(this._source); | |
| 80 | |
| 81 bool get isBroadcast => _source.isBroadcast; | |
| 82 | |
| 83 StreamSubscription<T> listen(void onData(T value), | |
| 84 { Function onError, | |
| 85 void onDone(), | |
| 86 bool cancelOnError }) { | |
| 87 cancelOnError = identical(true, cancelOnError); | |
| 88 return _createSubscription(onData, onError, onDone, cancelOnError); | |
| 89 } | |
| 90 | |
| 91 StreamSubscription<T> _createSubscription( | |
| 92 void onData(T data), | |
| 93 Function onError, | |
| 94 void onDone(), | |
| 95 bool cancelOnError) { | |
| 96 return new _ForwardingStreamSubscription<S, T>( | |
| 97 this, onData, onError, onDone, cancelOnError); | |
| 98 } | |
| 99 | |
| 100 // Override the following methods in subclasses to change the behavior. | |
| 101 | |
| 102 void _handleData(S data, _EventSink<T> sink) { | |
| 103 dynamic outputData = data; | |
| 104 sink._add(outputData); | |
| 105 } | |
| 106 | |
| 107 void _handleError(error, StackTrace stackTrace, _EventSink<T> sink) { | |
| 108 sink._addError(error, stackTrace); | |
| 109 } | |
| 110 | |
| 111 void _handleDone(_EventSink<T> sink) { | |
| 112 sink._close(); | |
| 113 } | |
| 114 } | |
| 115 | |
| 116 /** | |
| 117 * Abstract superclass for subscriptions that forward to other subscriptions. | |
| 118 */ | |
| 119 class _ForwardingStreamSubscription<S, T> | |
| 120 extends _BufferingStreamSubscription<T> { | |
| 121 final _ForwardingStream<S, T> _stream; | |
| 122 | |
| 123 StreamSubscription<S> _subscription; | |
| 124 | |
| 125 _ForwardingStreamSubscription(this._stream, void onData(T data), | |
| 126 Function onError, void onDone(), | |
| 127 bool cancelOnError) | |
| 128 : super(onData, onError, onDone, cancelOnError) { | |
| 129 _subscription = _stream._source.listen(_handleData, | |
| 130 onError: _handleError, | |
| 131 onDone: _handleDone); | |
| 132 } | |
| 133 | |
| 134 // _StreamSink interface. | |
| 135 // Transformers sending more than one event have no way to know if the stream | |
| 136 // is canceled or closed after the first, so we just ignore remaining events. | |
| 137 | |
| 138 void _add(T data) { | |
| 139 if (_isClosed) return; | |
| 140 super._add(data); | |
| 141 } | |
| 142 | |
| 143 void _addError(Object error, StackTrace stackTrace) { | |
| 144 if (_isClosed) return; | |
| 145 super._addError(error, stackTrace); | |
| 146 } | |
| 147 | |
| 148 // StreamSubscription callbacks. | |
| 149 | |
| 150 void _onPause() { | |
| 151 if (_subscription == null) return; | |
| 152 _subscription.pause(); | |
| 153 } | |
| 154 | |
| 155 void _onResume() { | |
| 156 if (_subscription == null) return; | |
| 157 _subscription.resume(); | |
| 158 } | |
| 159 | |
| 160 Future _onCancel() { | |
| 161 if (_subscription != null) { | |
| 162 StreamSubscription subscription = _subscription; | |
| 163 _subscription = null; | |
| 164 subscription.cancel(); | |
| 165 } | |
| 166 return null; | |
| 167 } | |
| 168 | |
| 169 // Methods used as listener on source subscription. | |
| 170 | |
| 171 void _handleData(S data) { | |
| 172 _stream._handleData(data, this); | |
| 173 } | |
| 174 | |
| 175 void _handleError(error, StackTrace stackTrace) { | |
| 176 _stream._handleError(error, stackTrace, this); | |
| 177 } | |
| 178 | |
| 179 void _handleDone() { | |
| 180 _stream._handleDone(this); | |
| 181 } | |
| 182 } | |
| 183 | |
| 184 // ------------------------------------------------------------------- | |
| 185 // Stream transformers used by the default Stream implementation. | |
| 186 // ------------------------------------------------------------------- | |
| 187 | |
| 188 typedef bool _Predicate<T>(T value); | |
| 189 | |
| 190 void _addErrorWithReplacement(_EventSink sink, error, stackTrace) { | |
| 191 AsyncError replacement = Zone.current.errorCallback(error, stackTrace); | |
| 192 if (replacement != null) { | |
| 193 error = _nonNullError(replacement.error); | |
| 194 stackTrace = replacement.stackTrace; | |
| 195 } | |
| 196 sink._addError(error, stackTrace); | |
| 197 } | |
| 198 | |
| 199 | |
| 200 class _WhereStream<T> extends _ForwardingStream<T, T> { | |
| 201 final _Predicate<T> _test; | |
| 202 | |
| 203 _WhereStream(Stream<T> source, bool test(T value)) | |
| 204 : _test = test, super(source); | |
| 205 | |
| 206 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 207 bool satisfies; | |
| 208 try { | |
| 209 satisfies = _test(inputEvent); | |
| 210 } catch (e, s) { | |
| 211 _addErrorWithReplacement(sink, e, s); | |
| 212 return; | |
| 213 } | |
| 214 if (satisfies) { | |
| 215 sink._add(inputEvent); | |
| 216 } | |
| 217 } | |
| 218 } | |
| 219 | |
| 220 | |
| 221 typedef T _Transformation<S, T>(S value); | |
| 222 | |
| 223 /** | |
| 224 * A stream pipe that converts data events before passing them on. | |
| 225 */ | |
| 226 class _MapStream<S, T> extends _ForwardingStream<S, T> { | |
| 227 final _Transformation _transform; | |
| 228 | |
| 229 _MapStream(Stream<S> source, T transform(S event)) | |
| 230 : this._transform = transform, super(source); | |
| 231 | |
| 232 void _handleData(S inputEvent, _EventSink<T> sink) { | |
| 233 T outputEvent; | |
| 234 try { | |
| 235 outputEvent = _transform(inputEvent); | |
| 236 } catch (e, s) { | |
| 237 _addErrorWithReplacement(sink, e, s); | |
| 238 return; | |
| 239 } | |
| 240 sink._add(outputEvent); | |
| 241 } | |
| 242 } | |
| 243 | |
| 244 /** | |
| 245 * A stream pipe that converts data events before passing them on. | |
| 246 */ | |
| 247 class _ExpandStream<S, T> extends _ForwardingStream<S, T> { | |
| 248 final _Transformation<S, Iterable<T>> _expand; | |
| 249 | |
| 250 _ExpandStream(Stream<S> source, Iterable<T> expand(S event)) | |
| 251 : this._expand = expand, super(source); | |
| 252 | |
| 253 void _handleData(S inputEvent, _EventSink<T> sink) { | |
| 254 try { | |
| 255 for (T value in _expand(inputEvent)) { | |
| 256 sink._add(value); | |
| 257 } | |
| 258 } catch (e, s) { | |
| 259 // If either _expand or iterating the generated iterator throws, | |
| 260 // we abort the iteration. | |
| 261 _addErrorWithReplacement(sink, e, s); | |
| 262 } | |
| 263 } | |
| 264 } | |
| 265 | |
| 266 | |
| 267 typedef bool _ErrorTest(error); | |
| 268 | |
| 269 /** | |
| 270 * A stream pipe that converts or disposes error events | |
| 271 * before passing them on. | |
| 272 */ | |
| 273 class _HandleErrorStream<T> extends _ForwardingStream<T, T> { | |
| 274 final Function _transform; | |
| 275 final _ErrorTest _test; | |
| 276 | |
| 277 _HandleErrorStream(Stream<T> source, | |
| 278 Function onError, | |
| 279 bool test(error)) | |
| 280 : this._transform = onError, this._test = test, super(source); | |
| 281 | |
| 282 void _handleError(Object error, StackTrace stackTrace, _EventSink<T> sink) { | |
| 283 bool matches = true; | |
| 284 if (_test != null) { | |
| 285 try { | |
| 286 matches = _test(error); | |
| 287 } catch (e, s) { | |
| 288 _addErrorWithReplacement(sink, e, s); | |
| 289 return; | |
| 290 } | |
| 291 } | |
| 292 if (matches) { | |
| 293 try { | |
| 294 _invokeErrorHandler(_transform, error, stackTrace); | |
| 295 } catch (e, s) { | |
| 296 if (identical(e, error)) { | |
| 297 sink._addError(error, stackTrace); | |
| 298 } else { | |
| 299 _addErrorWithReplacement(sink, e, s); | |
| 300 } | |
| 301 return; | |
| 302 } | |
| 303 } else { | |
| 304 sink._addError(error, stackTrace); | |
| 305 } | |
| 306 } | |
| 307 } | |
| 308 | |
| 309 | |
| 310 class _TakeStream<T> extends _ForwardingStream<T, T> { | |
| 311 int _remaining; | |
| 312 | |
| 313 _TakeStream(Stream<T> source, int count) | |
| 314 : this._remaining = count, super(source) { | |
| 315 // This test is done early to avoid handling an async error | |
| 316 // in the _handleData method. | |
| 317 if (count is! int) throw new ArgumentError(count); | |
| 318 } | |
| 319 | |
| 320 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 321 if (_remaining > 0) { | |
| 322 sink._add(inputEvent); | |
| 323 _remaining -= 1; | |
| 324 if (_remaining == 0) { | |
| 325 // Closing also unsubscribes all subscribers, which unsubscribes | |
| 326 // this from source. | |
| 327 sink._close(); | |
| 328 } | |
| 329 } | |
| 330 } | |
| 331 } | |
| 332 | |
| 333 | |
| 334 class _TakeWhileStream<T> extends _ForwardingStream<T, T> { | |
| 335 final _Predicate<T> _test; | |
| 336 | |
| 337 _TakeWhileStream(Stream<T> source, bool test(T value)) | |
| 338 : this._test = test, super(source); | |
| 339 | |
| 340 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 341 bool satisfies; | |
| 342 try { | |
| 343 satisfies = _test(inputEvent); | |
| 344 } catch (e, s) { | |
| 345 _addErrorWithReplacement(sink, e, s); | |
| 346 // The test didn't say true. Didn't say false either, but we stop anyway. | |
| 347 sink._close(); | |
| 348 return; | |
| 349 } | |
| 350 if (satisfies) { | |
| 351 sink._add(inputEvent); | |
| 352 } else { | |
| 353 sink._close(); | |
| 354 } | |
| 355 } | |
| 356 } | |
| 357 | |
| 358 class _SkipStream<T> extends _ForwardingStream<T, T> { | |
| 359 int _remaining; | |
| 360 | |
| 361 _SkipStream(Stream<T> source, int count) | |
| 362 : this._remaining = count, super(source) { | |
| 363 // This test is done early to avoid handling an async error | |
| 364 // in the _handleData method. | |
| 365 if (count is! int || count < 0) throw new ArgumentError(count); | |
| 366 } | |
| 367 | |
| 368 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 369 if (_remaining > 0) { | |
| 370 _remaining--; | |
| 371 return; | |
| 372 } | |
| 373 sink._add(inputEvent); | |
| 374 } | |
| 375 } | |
| 376 | |
| 377 class _SkipWhileStream<T> extends _ForwardingStream<T, T> { | |
| 378 final _Predicate<T> _test; | |
| 379 bool _hasFailed = false; | |
| 380 | |
| 381 _SkipWhileStream(Stream<T> source, bool test(T value)) | |
| 382 : this._test = test, super(source); | |
| 383 | |
| 384 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 385 if (_hasFailed) { | |
| 386 sink._add(inputEvent); | |
| 387 return; | |
| 388 } | |
| 389 bool satisfies; | |
| 390 try { | |
| 391 satisfies = _test(inputEvent); | |
| 392 } catch (e, s) { | |
| 393 _addErrorWithReplacement(sink, e, s); | |
| 394 // A failure to return a boolean is considered "not matching". | |
| 395 _hasFailed = true; | |
| 396 return; | |
| 397 } | |
| 398 if (!satisfies) { | |
| 399 _hasFailed = true; | |
| 400 sink._add(inputEvent); | |
| 401 } | |
| 402 } | |
| 403 } | |
| 404 | |
| 405 typedef bool _Equality<T>(T a, T b); | |
| 406 | |
| 407 class _DistinctStream<T> extends _ForwardingStream<T, T> { | |
| 408 static var _SENTINEL = new Object(); | |
| 409 | |
| 410 _Equality<T> _equals; | |
| 411 var _previous = _SENTINEL; | |
| 412 | |
| 413 _DistinctStream(Stream<T> source, bool equals(T a, T b)) | |
| 414 : _equals = equals, super(source); | |
| 415 | |
| 416 void _handleData(T inputEvent, _EventSink<T> sink) { | |
| 417 if (identical(_previous, _SENTINEL)) { | |
| 418 _previous = inputEvent; | |
| 419 return sink._add(inputEvent); | |
| 420 } else { | |
| 421 bool isEqual; | |
| 422 try { | |
| 423 if (_equals == null) { | |
| 424 isEqual = (_previous == inputEvent); | |
| 425 } else { | |
| 426 isEqual = _equals(_previous, inputEvent); | |
| 427 } | |
| 428 } catch (e, s) { | |
| 429 _addErrorWithReplacement(sink, e, s); | |
| 430 return null; | |
| 431 } | |
| 432 if (!isEqual) { | |
| 433 sink._add(inputEvent); | |
| 434 _previous = inputEvent; | |
| 435 } | |
| 436 } | |
| 437 } | |
| 438 } | |
| OLD | NEW |