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 [StreamChannel] that specifically enforces the second stream channel | |
12 /// guarantee. | |
Bob Nystrom
2016/06/06 22:54:53
To someone reading this doc cold, it's not clear w
nweiz
2016/06/06 23:25:30
Done.
| |
13 /// | |
14 /// This is exposed via [new StreamChannel.withCloseGuarantee]. | |
15 class CloseGuaranteeChannel<T> extends StreamChannelMixin<T> { | |
16 Stream<T> get stream => _stream; | |
17 _CloseGuaranteeStream<T> _stream; | |
18 | |
19 StreamSink<T> get sink => _sink; | |
20 _CloseGuaranteeSink<T> _sink; | |
21 | |
22 /// The subscription to the inner stream. | |
23 StreamSubscription<T> _subscription; | |
24 | |
25 /// Whether the sink has closed, causing the underlying channel to disconnect. | |
26 bool _disconnected = false; | |
27 | |
28 CloseGuaranteeChannel(Stream<T> innerStream, StreamSink<T> innerSink) { | |
29 _sink = new _CloseGuaranteeSink<T>(innerSink, this); | |
30 _stream = new _CloseGuaranteeStream<T>(innerStream, this); | |
31 } | |
32 } | |
33 | |
34 /// The stream for [CloseGuaranteeChannel]. | |
35 /// | |
36 /// This wraps the inner stream to save the subscription on the channel when | |
37 /// [listen] is called. | |
38 class _CloseGuaranteeStream<T> extends Stream<T> { | |
39 /// The inner stream this is delegating to. | |
40 final Stream<T> _inner; | |
41 | |
42 /// The [CloseGuaranteeChannel] this belongs to. | |
43 final CloseGuaranteeChannel<T> _channel; | |
44 | |
45 _CloseGuaranteeStream(this._inner, this._channel); | |
46 | |
47 StreamSubscription<T> listen(void onData(T event), | |
48 {Function onError, | |
49 void onDone(), | |
50 bool cancelOnError}) { | |
Bob Nystrom
2016/06/06 22:54:53
Run dartfmt?
nweiz
2016/06/06 23:25:30
Done.
| |
51 // If the channel is already disconnected, we shouldn't dispatch anything | |
52 // but a done event. | |
53 if (_channel._disconnected) { | |
54 onData = null; | |
55 onError = null; | |
56 } | |
57 | |
58 var subscription = _inner.listen(onData, | |
59 onError: onError, onDone: onDone, cancelOnError: cancelOnError); | |
60 if (!_channel._disconnected) { | |
61 _channel._subscription = subscription; | |
62 } | |
63 return subscription; | |
64 } | |
65 } | |
66 | |
67 /// The sink for [CloseGuaranteeChannel]. | |
68 /// | |
69 /// This wraps the inner sink to cancel the stream subscription when the sink is | |
70 /// canceled. | |
71 class _CloseGuaranteeSink<T> extends DelegatingStreamSink<T> { | |
72 /// The [CloseGuaranteeChannel] this belongs to. | |
73 final CloseGuaranteeChannel<T> _channel; | |
74 | |
75 _CloseGuaranteeSink(StreamSink<T> inner, this._channel) : super(inner); | |
76 | |
77 Future close() { | |
78 var done = super.close(); | |
79 _channel._disconnected = true; | |
80 if (_channel._subscription != null) { | |
81 // Don't dispatch anything but a done event. | |
82 _channel._subscription.onData(null); | |
83 _channel._subscription.onError(null); | |
84 } | |
85 return done; | |
86 } | |
87 } | |
OLD | NEW |