|
| 1 | +// Copyright (c) 2015, 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 | +library async.forkable_stream; |
| 6 | + |
| 7 | +import 'dart:async'; |
| 8 | + |
| 9 | +import 'stream_completer.dart'; |
| 10 | + |
| 11 | +/// A single-subscription stream from which other streams may be forked off at |
| 12 | +/// the current position. |
| 13 | +/// |
| 14 | +/// This adds an operation, [fork], which produces a new stream that |
| 15 | +/// independently emits the same events as this stream. Unlike the branches |
| 16 | +/// produced by [StreamSplitter], a fork only emits events that arrive *after* |
| 17 | +/// the call to [fork]. |
| 18 | +/// |
| 19 | +/// Each fork can be paused or canceled independently of one another and of this |
| 20 | +/// stream. The underlying stream will be listened to once any branch is |
| 21 | +/// listened to. It will be paused when all branches are paused or not yet |
| 22 | +/// listened to. It will be canceled when all branches have been listened to and |
| 23 | +/// then canceled. |
| 24 | +class ForkableStream<T> extends StreamView<T> { |
| 25 | + /// The underlying stream. |
| 26 | + final Stream _sourceStream; |
| 27 | + |
| 28 | + /// The subscription to [_sourceStream]. |
| 29 | + /// |
| 30 | + /// This will be `null` until this stream or any of its forks are listened to. |
| 31 | + StreamSubscription _subscription; |
| 32 | + |
| 33 | + /// Whether this has been cancelled and no more forks may be created. |
| 34 | + bool _isCanceled = false; |
| 35 | + |
| 36 | + /// The controllers for any branches that have not yet been canceled. |
| 37 | + /// |
| 38 | + /// This includes a controller for this stream, until that has been cancelled. |
| 39 | + final _controllers = new Set<StreamController<T>>(); |
| 40 | + |
| 41 | + /// Creates a new forkable stream wrapping [sourceStream]. |
| 42 | + ForkableStream(Stream sourceStream) |
| 43 | + // Use a completer here so that we can provide its stream to the |
| 44 | + // superclass constructor while also adding the stream controller to |
| 45 | + // [_controllers]. |
| 46 | + : this._(sourceStream, new StreamCompleter()); |
| 47 | + |
| 48 | + ForkableStream._(this._sourceStream, StreamCompleter completer) |
| 49 | + : super(completer.stream) { |
| 50 | + completer.setSourceStream(_fork(primary: true)); |
| 51 | + } |
| 52 | + |
| 53 | + /// Creates a new fork of this stream. |
| 54 | + /// |
| 55 | + /// From this point forward, the fork will emit the same events as this |
| 56 | + /// stream. It will *not* emit any events that have already been emitted by |
| 57 | + /// this stream. The fork is independent of this stream, which means each one |
| 58 | + /// may be paused or canceled without affecting the other. |
| 59 | + /// |
| 60 | + /// If this stream is done or its subscription has been canceled, this returns |
| 61 | + /// an empty stream. |
| 62 | + Stream<T> fork() => _fork(primary: false); |
| 63 | + |
| 64 | + /// Creates a stream forwarding [_sourceStream]. |
| 65 | + /// |
| 66 | + /// If [primary] is true, this is the stream underlying this object; |
| 67 | + /// otherwise, it's a fork. The only difference is that when the primary |
| 68 | + /// stream is canceled, [fork] starts throwing [StateError]s. |
| 69 | + Stream<T> _fork({bool primary: false}) { |
| 70 | + if (_isCanceled) { |
| 71 | + var controller = new StreamController<T>()..close(); |
| 72 | + return controller.stream; |
| 73 | + } |
| 74 | + |
| 75 | + var controller; |
| 76 | + controller = new StreamController<T>( |
| 77 | + onListen: () => _onListenOrResume(controller), |
| 78 | + onCancel: () => _onCancel(controller, primary: primary), |
| 79 | + onPause: () => _onPause(controller), |
| 80 | + onResume: () => _onListenOrResume(controller), |
| 81 | + sync: true); |
| 82 | + |
| 83 | + _controllers.add(controller); |
| 84 | + |
| 85 | + return controller.stream; |
| 86 | + } |
| 87 | + |
| 88 | + /// The callback called when `onListen` or `onResume` is called for the branch |
| 89 | + /// managed by [controller]. |
| 90 | + /// |
| 91 | + /// This ensures that we're subscribed to [_sourceStream] and that the |
| 92 | + /// subscription isn't paused. |
| 93 | + void _onListenOrResume(StreamController<T> controller) { |
| 94 | + if (controller.isClosed) return; |
| 95 | + if (_subscription == null) { |
| 96 | + _subscription = |
| 97 | + _sourceStream.listen(_onData, onError: _onError, onDone: _onDone); |
| 98 | + } else { |
| 99 | + _subscription.resume(); |
| 100 | + } |
| 101 | + } |
| 102 | + |
| 103 | + /// The callback called when `onCancel` is called for the branch managed by |
| 104 | + /// [controller]. |
| 105 | + /// |
| 106 | + /// This cancels or pauses the underlying subscription as necessary. If |
| 107 | + /// [primary] is true, it also ensures that future calls to [fork] throw |
| 108 | + /// [StateError]s. |
| 109 | + Future _onCancel(StreamController<T> controller, {bool primary: false}) { |
| 110 | + if (primary) _isCanceled = true; |
| 111 | + |
| 112 | + if (controller.isClosed) return null; |
| 113 | + _controllers.remove(controller); |
| 114 | + |
| 115 | + if (_controllers.isEmpty) return _subscription.cancel(); |
| 116 | + |
| 117 | + _onPause(controller); |
| 118 | + return null; |
| 119 | + } |
| 120 | + |
| 121 | + /// The callback called when `onPause` is called for the branch managed by |
| 122 | + /// [controller]. |
| 123 | + /// |
| 124 | + /// This pauses the underlying subscription if necessary. |
| 125 | + void _onPause(StreamController<T> controller) { |
| 126 | + if (controller.isClosed) return; |
| 127 | + if (_subscription.isPaused) return; |
| 128 | + if (_controllers.any((controller) => |
| 129 | + controller.hasListener && !controller.isPaused)) { |
| 130 | + return; |
| 131 | + } |
| 132 | + |
| 133 | + _subscription.pause(); |
| 134 | + } |
| 135 | + |
| 136 | + /// Forwards data events to all branches. |
| 137 | + void _onData(value) { |
| 138 | + // Don't iterate directly over the set because [controller.add] might cause |
| 139 | + // it to be modified synchronously. |
| 140 | + for (var controller in _controllers.toList()) { |
| 141 | + controller.add(value); |
| 142 | + } |
| 143 | + } |
| 144 | + |
| 145 | + /// Forwards error events to all branches. |
| 146 | + void _onError(error, StackTrace stackTrace) { |
| 147 | + // Don't iterate directly over the set because [controller.addError] might |
| 148 | + // cause it to be modified synchronously. |
| 149 | + for (var controller in _controllers.toList()) { |
| 150 | + controller.addError(error, stackTrace); |
| 151 | + } |
| 152 | + } |
| 153 | + |
| 154 | + /// Forwards close events to all branches. |
| 155 | + void _onDone() { |
| 156 | + _isCanceled = true; |
| 157 | + |
| 158 | + // Don't iterate directly over the set because [controller.close] might |
| 159 | + // cause it to be modified synchronously. |
| 160 | + for (var controller in _controllers.toList()) { |
| 161 | + controller.close(); |
| 162 | + } |
| 163 | + _controllers.clear(); |
| 164 | + } |
| 165 | +} |
| 166 | + |
0 commit comments