| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright (c) 2013, 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 scheduled_test.scheduled_stream; | |
| 6 | |
| 7 import 'dart:async'; | |
| 8 import 'dart:collection'; | |
| 9 | |
| 10 import 'package:stack_trace/stack_trace.dart'; | |
| 11 | |
| 12 import 'scheduled_test.dart'; | |
| 13 import 'src/stream_matcher.dart'; | |
| 14 import 'src/utils.dart'; | |
| 15 | |
| 16 export 'src/stream_matcher.dart'; | |
| 17 | |
| 18 /// A wrapper for streams that supports a pull-based model of retrieving values | |
| 19 /// as well as a set of [StreamMatcher]s for testing the values emitted by the | |
| 20 /// stream. | |
| 21 /// | |
| 22 /// The only method on [ScheduledStream] that's actually scheduled is [expect], | |
| 23 /// which is the method that users testing streaming code are most likely to | |
| 24 /// want to use. | |
| 25 class ScheduledStream<T> { | |
| 26 /// The underlying stream. | |
| 27 final Stream<T> _stream; | |
| 28 | |
| 29 /// The subscription to [_stream]. | |
| 30 StreamSubscription<T> _subscription; | |
| 31 | |
| 32 /// The completer for emitting a value requested by [next]. | |
| 33 /// | |
| 34 /// If this is non-null, [_pendingValues] will always be empty, since any | |
| 35 /// value coming in will be passed to this completer. | |
| 36 Completer<T> _nextCompleter; | |
| 37 | |
| 38 /// The completer for emitting a value requested by [hasNext]. | |
| 39 Completer<bool> _hasNextCompleter; | |
| 40 | |
| 41 /// The set of all streams forked from this one. | |
| 42 final _forks = new Set<ScheduledStream<T>>(); | |
| 43 final _forkControllers = new Set<StreamController<T>>(); | |
| 44 | |
| 45 /// The queue of values emitted by [_stream] but not yet emitted through | |
| 46 /// [next]. | |
| 47 final _pendingValues = new Queue<Fallible<T>>(); | |
| 48 | |
| 49 /// All values emitted by this stream so far. | |
| 50 /// | |
| 51 /// This does not include values emitted by the underlying stream but not yet | |
| 52 /// emitted through [next]. | |
| 53 List<T> get emittedValues => new UnmodifiableListView(_emittedValues); | |
| 54 final _emittedValues = new List<T>(); | |
| 55 | |
| 56 /// All values emitted by the underlying stream. | |
| 57 /// | |
| 58 /// This is intended primarily for providing debugging information. | |
| 59 List<T> get allValues { | |
| 60 var list = new List<T>.from(_emittedValues); | |
| 61 list.addAll(_pendingValues.where((value) => value.hasValue) | |
| 62 .map((value) => value.value)); | |
| 63 return new UnmodifiableListView(list); | |
| 64 } | |
| 65 | |
| 66 /// Whether the wrapped stream has been closed. | |
| 67 bool _isDone = false; | |
| 68 | |
| 69 /// Whether [next] has been called but has not yet returned. | |
| 70 /// | |
| 71 /// This is distinct from `_nextCompleter != null` when [next] is called while | |
| 72 /// there are pending values available, until the future it returns completes. | |
| 73 bool _isNextPending = false; | |
| 74 | |
| 75 /// Creates a new scheduled stream wrapping [stream]. | |
| 76 ScheduledStream(this._stream) { | |
| 77 _subscription = _stream.listen((value) { | |
| 78 for (var c in _forkControllers) { | |
| 79 c.add(value); | |
| 80 } | |
| 81 if (_hasNextCompleter != null) { | |
| 82 _hasNextCompleter.complete(true); | |
| 83 _hasNextCompleter = null; | |
| 84 } | |
| 85 | |
| 86 if (_nextCompleter != null) { | |
| 87 _nextCompleter.complete(value); | |
| 88 _emittedValues.add(value); | |
| 89 _nextCompleter = null; | |
| 90 _isNextPending = false; | |
| 91 } else { | |
| 92 _pendingValues.add(new Fallible.withValue(value)); | |
| 93 } | |
| 94 }, onError: (error, stackTrace) { | |
| 95 for (var c in _forkControllers) { | |
| 96 c.addError(error, stackTrace); | |
| 97 } | |
| 98 if (_hasNextCompleter != null) { | |
| 99 _hasNextCompleter.completeError(error, stackTrace); | |
| 100 _hasNextCompleter = null; | |
| 101 } | |
| 102 | |
| 103 if (_nextCompleter != null) { | |
| 104 _nextCompleter.completeError(error, stackTrace); | |
| 105 _nextCompleter = null; | |
| 106 } else { | |
| 107 _pendingValues.add(new Fallible.withError(error, stackTrace)); | |
| 108 } | |
| 109 }, onDone: _onDone); | |
| 110 } | |
| 111 | |
| 112 /// Enqueue an expectation that [streamMatcher] will match the value(s) | |
| 113 /// emitted by the stream at this point in the schedule. | |
| 114 /// | |
| 115 /// If [streamMatcher] is a [StreamMatcher], it will match the stream as a | |
| 116 /// whole. If it's a [Matcher] or another object, it will match the next value | |
| 117 /// emitted by the stream (as though it were a [nextValue] matcher). | |
| 118 /// | |
| 119 /// This call is scheduled; the expectation won't be added until the schedule | |
| 120 /// reaches this point, and the schedule won't continue until the matcher has | |
| 121 /// matched the stream. | |
| 122 void expect(streamMatcher) { | |
| 123 streamMatcher = new StreamMatcher.wrap(streamMatcher); | |
| 124 var description = 'stream emits $streamMatcher'; | |
| 125 schedule(() { | |
| 126 return streamMatcher.tryMatch(this).then((description) { | |
| 127 if (description == null) return; | |
| 128 | |
| 129 var expected = prefixLines(streamMatcher.toString(), | |
| 130 firstPrefix: 'Expected: ', | |
| 131 prefix: ' | '); | |
| 132 | |
| 133 var actual = prefixLines(allValues.map((value) { | |
| 134 return prefixLines(value.toString(), firstPrefix: '* '); | |
| 135 }).join('\n'), | |
| 136 firstPrefix: ' Emitted: ', | |
| 137 prefix: ' '); | |
| 138 | |
| 139 var which = ''; | |
| 140 if (description.length > 0) { | |
| 141 which = '\n' + prefixLines(description.toString(), | |
| 142 firstPrefix: ' Which: ', | |
| 143 prefix: ' | '); | |
| 144 } | |
| 145 | |
| 146 fail("$expected\n$actual$which"); | |
| 147 }); | |
| 148 }, description); | |
| 149 } | |
| 150 | |
| 151 /// Returns a Future that completes to the next value emitted by this stream. | |
| 152 /// | |
| 153 /// It's a [StateError] to call [next] when another call's Future has not yet | |
| 154 /// completed, or when the stream has no more values. The latter can be | |
| 155 /// checked using [hasNext]. | |
| 156 Future<T> next() { | |
| 157 if (_isNextPending) { | |
| 158 return new Future.error( | |
| 159 new StateError("There's already a pending call to " | |
| 160 "ScheduledStream.next."), | |
| 161 new Chain.current()); | |
| 162 } | |
| 163 | |
| 164 if (_pendingValues.isNotEmpty) { | |
| 165 _isNextPending = true; | |
| 166 | |
| 167 var valueOrError = _pendingValues.removeFirst(); | |
| 168 if (valueOrError.hasValue) { | |
| 169 _emittedValues.add(valueOrError.value); | |
| 170 } | |
| 171 return valueOrError.toFuture().whenComplete(() { | |
| 172 _isNextPending = false; | |
| 173 }); | |
| 174 } else if (_isDone) { | |
| 175 return new Future.error( | |
| 176 new StateError("ScheduledStream has no more elements."), | |
| 177 new Chain.current()); | |
| 178 } | |
| 179 | |
| 180 _isNextPending = true; | |
| 181 _nextCompleter = new Completer(); | |
| 182 return _nextCompleter.future; | |
| 183 } | |
| 184 | |
| 185 /// Returns a Future that completes to a boolean indicating whether the stream | |
| 186 /// has additional values or not. | |
| 187 Future<bool> get hasNext { | |
| 188 if (_hasNextCompleter != null) return _hasNextCompleter.future; | |
| 189 | |
| 190 if (_pendingValues.isNotEmpty) { | |
| 191 return _pendingValues.first.toFuture().then((_) => true); | |
| 192 } else if (_isDone) { | |
| 193 return new Future.value(false); | |
| 194 } | |
| 195 | |
| 196 _hasNextCompleter = new Completer(); | |
| 197 return _hasNextCompleter.future; | |
| 198 } | |
| 199 | |
| 200 /// Returns a fork of this stream. | |
| 201 /// | |
| 202 /// The fork begins at the same point [this] is at. Values can be read from it | |
| 203 /// without consuming values in [this]. If [this] is closed, the fork will be | |
| 204 /// closed at whatever point it's currently at. | |
| 205 ScheduledStream<T> fork() { | |
| 206 var controller = new StreamController<T>(); | |
| 207 for (var valueOrError in _pendingValues) { | |
| 208 if (valueOrError.hasValue) { | |
| 209 controller.add(valueOrError.value); | |
| 210 } else { | |
| 211 controller.addError(valueOrError.error, valueOrError.stackTrace); | |
| 212 } | |
| 213 } | |
| 214 if (_isDone) { | |
| 215 controller.close(); | |
| 216 } else { | |
| 217 _forkControllers.add(controller); | |
| 218 } | |
| 219 | |
| 220 var fork = new ScheduledStream<T>(controller.stream); | |
| 221 _forks.add(fork); | |
| 222 return fork; | |
| 223 } | |
| 224 | |
| 225 /// Closes this stream. | |
| 226 /// | |
| 227 /// This cancels the subscription to the underlying stream and acts as though | |
| 228 /// [this] was closed immediately after the current position, regardless of | |
| 229 /// whether the underlying stream has emitted additional events. | |
| 230 void close() { | |
| 231 _subscription.cancel(); | |
| 232 _pendingValues.clear(); | |
| 233 | |
| 234 for (var fork in _forks) { | |
| 235 fork.close(); | |
| 236 } | |
| 237 _forks.clear(); | |
| 238 | |
| 239 if (!_isDone) _onDone(); | |
| 240 } | |
| 241 | |
| 242 /// Handles a "done" event from the underlying stream, as well as [this] being | |
| 243 /// closed. | |
| 244 void _onDone() { | |
| 245 for (var c in _forkControllers) { | |
| 246 c.close(); | |
| 247 } | |
| 248 _forkControllers.clear(); | |
| 249 | |
| 250 if (_hasNextCompleter != null) { | |
| 251 _hasNextCompleter.complete(false); | |
| 252 _hasNextCompleter = null; | |
| 253 } | |
| 254 | |
| 255 if (_nextCompleter != null) { | |
| 256 _nextCompleter.completeError( | |
| 257 new StateError("ScheduledStream has no more elements."), | |
| 258 new Chain.current()); | |
| 259 _nextCompleter = null; | |
| 260 _isNextPending = false; | |
| 261 } | |
| 262 | |
| 263 _isDone = true; | |
| 264 } | |
| 265 } | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 812253002:
Delete a bunch of packages that are now on GitHub. (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart