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 'package:async/async.dart'; |
| 8 |
| 9 import '../stream_channel.dart'; |
| 10 |
| 11 /// A [channel] where the source and destination are provided later. |
| 12 /// |
| 13 /// The [channel] is a normal channel that can be listened to and that events |
| 14 /// can be added to immediately, but until [setChannel] is called it won't emit |
| 15 /// any events and all events added to it will be buffered. |
| 16 class StreamChannelCompleter<T> { |
| 17 /// The completer for this channel's stream. |
| 18 final _streamCompleter = new StreamCompleter<T>(); |
| 19 |
| 20 /// The completer for this channel's sink. |
| 21 final _sinkCompleter = new StreamSinkCompleter<T>(); |
| 22 |
| 23 /// The channel for this completer. |
| 24 StreamChannel<T> get channel => _channel; |
| 25 StreamChannel<T> _channel; |
| 26 |
| 27 /// Whether [setChannel] has been called. |
| 28 bool _set = false; |
| 29 |
| 30 /// Convert a `Future<StreamChannel>` to a `StreamChannel`. |
| 31 /// |
| 32 /// This creates a channel using a channel completer, and sets the source |
| 33 /// channel to the result of the future when the future completes. |
| 34 /// |
| 35 /// If the future completes with an error, the returned channel's stream will |
| 36 /// instead contain just that error. The sink will silently discard all |
| 37 /// events. |
| 38 static StreamChannel fromFuture(Future<StreamChannel> channelFuture) { |
| 39 var completer = new StreamChannelCompleter(); |
| 40 channelFuture.then(completer.setChannel, onError: completer.setError); |
| 41 return completer.channel; |
| 42 } |
| 43 |
| 44 StreamChannelCompleter() { |
| 45 _channel = new StreamChannel<T>( |
| 46 _streamCompleter.stream, _sinkCompleter.sink); |
| 47 } |
| 48 |
| 49 /// Set a channel as the source and destination for [channel]. |
| 50 /// |
| 51 /// A channel may be set at most once. |
| 52 /// |
| 53 /// Either [setChannel] or [setError] may be called at most once. Trying to |
| 54 /// call either of them again will fail. |
| 55 void setChannel(StreamChannel<T> channel) { |
| 56 if (_set) throw new StateError("The channel has already been set."); |
| 57 _set = true; |
| 58 |
| 59 _streamCompleter.setSourceStream(channel.stream); |
| 60 _sinkCompleter.setDestinationSink(channel.sink); |
| 61 } |
| 62 |
| 63 /// Indicates that there was an error connecting the channel. |
| 64 /// |
| 65 /// This makes the stream emit [error] and close. It makes the sink discard |
| 66 /// all its events. |
| 67 /// |
| 68 /// Either [setChannel] or [setError] may be called at most once. Trying to |
| 69 /// call either of them again will fail. |
| 70 void setError(error, [StackTrace stackTrace]) { |
| 71 if (_set) throw new StateError("The channel has already been set."); |
| 72 _set = true; |
| 73 |
| 74 _streamCompleter.setError(error, stackTrace); |
| 75 _sinkCompleter.setDestinationSink(new NullStreamSink()); |
| 76 } |
| 77 } |
OLD | NEW |