OLD | NEW |
(Empty) | |
| 1 // Copyright (c) 2016, 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 import 'dart:async'; |
| 6 |
| 7 import '../stream_sink_transformer.dart'; |
| 8 import '../delegate/stream_sink.dart'; |
| 9 |
| 10 /// The type of the callback for handling data events. |
| 11 typedef void HandleData<S, T>(S data, EventSink<T> sink); |
| 12 |
| 13 /// The type of the callback for handling error events. |
| 14 typedef void HandleError<T>( |
| 15 Object error, StackTrace stackTrace, EventSink<T> sink); |
| 16 |
| 17 /// The type of the callback for handling done events. |
| 18 typedef void HandleDone<T>(EventSink<T> sink); |
| 19 |
| 20 /// A [StreamSinkTransformer] that delegates events to the given handlers. |
| 21 class HandlerTransformer<S, T> implements StreamSinkTransformer<S, T> { |
| 22 /// The handler for data events. |
| 23 final HandleData<S, T> _handleData; |
| 24 |
| 25 /// The handler for error events. |
| 26 final HandleError<T> _handleError; |
| 27 |
| 28 /// The handler for done events. |
| 29 final HandleDone<T> _handleDone; |
| 30 |
| 31 HandlerTransformer(this._handleData, this._handleError, this._handleDone); |
| 32 |
| 33 StreamSink<S> bind(StreamSink<T> sink) => new _HandlerSink<S, T>(this, sink); |
| 34 } |
| 35 |
| 36 /// A sink created by [HandlerTransformer]. |
| 37 class _HandlerSink<S, T> implements StreamSink<S> { |
| 38 /// The transformer that created this sink. |
| 39 final HandlerTransformer<S, T> _transformer; |
| 40 |
| 41 /// The original sink that's being transformed. |
| 42 final StreamSink<T> _inner; |
| 43 |
| 44 /// The wrapper for [_inner] whose [StreamSink.close] method can't emit |
| 45 /// errors. |
| 46 final StreamSink<T> _safeCloseInner; |
| 47 |
| 48 Future get done => _inner.done; |
| 49 |
| 50 _HandlerSink(this._transformer, StreamSink<T> inner) |
| 51 : _inner = inner, |
| 52 _safeCloseInner = new _SafeCloseSink<T>(inner); |
| 53 |
| 54 void add(S event) { |
| 55 if (_transformer._handleData == null) { |
| 56 _inner.add(event as T); |
| 57 } else { |
| 58 _transformer._handleData(event, _safeCloseInner); |
| 59 } |
| 60 } |
| 61 |
| 62 void addError(error, [StackTrace stackTrace]) { |
| 63 if (_transformer._handleError == null) { |
| 64 _inner.addError(error, stackTrace); |
| 65 } else { |
| 66 _transformer._handleError(error, stackTrace, _safeCloseInner); |
| 67 } |
| 68 } |
| 69 |
| 70 Future addStream(Stream<S> stream) { |
| 71 return _inner.addStream(stream.transform( |
| 72 new StreamTransformer<S, T>.fromHandlers( |
| 73 handleData: _transformer._handleData, |
| 74 handleError: _transformer._handleError, |
| 75 handleDone: _closeSink))); |
| 76 } |
| 77 |
| 78 Future close() { |
| 79 if (_transformer._handleDone == null) return _inner.close(); |
| 80 |
| 81 _transformer._handleDone(_safeCloseInner); |
| 82 return _inner.done; |
| 83 } |
| 84 } |
| 85 |
| 86 /// A wrapper for [StreamSink]s that swallows any errors returned by [close]. |
| 87 /// |
| 88 /// [HandlerTransformer] passes this to its handlers to ensure that when they |
| 89 /// call [close], they don't leave any dangling [Future]s behind that might emit |
| 90 /// unhandleable errors. |
| 91 class _SafeCloseSink<T> extends DelegatingStreamSink<T> { |
| 92 _SafeCloseSink(StreamSink<T> inner) : super(inner); |
| 93 |
| 94 Future close() => super.close().catchError((_) {}); |
| 95 } |
| 96 |
| 97 /// A function to pass as a [StreamTransformer]'s `handleDone` callback. |
| 98 void _closeSink(EventSink sink) { |
| 99 sink.close(); |
| 100 } |
OLD | NEW |