Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(3)

Issues Search
    My Issues | Starred     Open | Closed | All

Side by Side Diff: test/dart_codegen/expect/async/stream_impl.dart

Issue 963593002: Disable formatting and add new-lines to make tests faster. (Closed) Base URL: git@github.com:dart-lang/dev_compiler.git@master
Patch Set: Created 5 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« test/dart_codegen/expect/_internal/iterable.dart ('K') | « test/dart_codegen/expect/async/stream_controller.dart ('k') | test/dart_codegen/expect/async/stream_pipe.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 part of dart.async; 1 part of dart.async;
2 2 abstract class _EventSink<T> {void _add(T data);
3 abstract class _EventSink<T> { 3 void _addError(Object error, StackTrace stackTrace);
4 void _add(T data); 4 void _close();
5 void _addError(Object error, StackTrace stackTrace); 5 }
6 void _close(); 6 abstract class _EventDispatch<T> {void _sendData(T data);
7 } 7 void _sendError(Object error, StackTrace stackTrace);
8 abstract class _EventDispatch<T> { 8 void _sendDone();
9 void _sendData(T data); 9 }
10 void _sendError(Object error, StackTrace stackTrace); 10 class _BufferingStreamSubscription<T> implements StreamSubscription<T>, _EventS ink<T>, _EventDispatch<T> {static const int _STATE_CANCEL_ON_ERROR = 1;
11 void _sendDone(); 11 static const int _STATE_CLOSED = 2;
12 } 12 static const int _STATE_INPUT_PAUSED = 4;
13 class _BufferingStreamSubscription<T> 13 static const int _STATE_CANCELED = 8;
14 implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T> { 14 static const int _STATE_WAIT_FOR_CANCEL = 16;
15 static const int _STATE_CANCEL_ON_ERROR = 1; 15 static const int _STATE_IN_CALLBACK = 32;
16 static const int _STATE_CLOSED = 2; 16 static const int _STATE_HAS_PENDING = 64;
17 static const int _STATE_INPUT_PAUSED = 4; 17 static const int _STATE_PAUSE_COUNT = 128;
18 static const int _STATE_CANCELED = 8; 18 static const int _STATE_PAUSE_COUNT_SHIFT = 7;
19 static const int _STATE_WAIT_FOR_CANCEL = 16; 19 _DataHandler<T> _onData;
20 static const int _STATE_IN_CALLBACK = 32; 20 Function _onError;
21 static const int _STATE_HAS_PENDING = 64; 21 _DoneHandler _onDone;
22 static const int _STATE_PAUSE_COUNT = 128; 22 final Zone _zone = Zone.current;
23 static const int _STATE_PAUSE_COUNT_SHIFT = 7; 23 int _state;
24 _DataHandler<T> _onData; 24 Future _cancelFuture;
25 Function _onError; 25 _PendingEvents _pending;
26 _DoneHandler _onDone; 26 _BufferingStreamSubscription(void onData(T data), Function onError, void onDone (), bool cancelOnError) : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
27 final Zone _zone = Zone.current; 27 this.onData(onData);
28 int _state; 28 this.onError(onError);
29 Future _cancelFuture; 29 this.onDone(onDone);
30 _PendingEvents _pending; 30 }
31 _BufferingStreamSubscription( 31 void _setPendingEvents(_PendingEvents pendingEvents) {
32 void onData(T data), Function onError, void onDone(), bool cancelOnError) 32 assert (_pending == null); if (pendingEvents == null) return; _pending = pending Events;
33 : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) { 33 if (!pendingEvents.isEmpty) {
34 this.onData(onData); 34 _state |= _STATE_HAS_PENDING;
35 this.onError(onError); 35 _pending.schedule(this);
36 this.onDone(onDone); 36 }
37 } 37 }
38 void _setPendingEvents(_PendingEvents pendingEvents) { 38 _PendingEvents _extractPending() {
39 assert(_pending == null); 39 assert (_isCanceled); _PendingEvents events = _pending;
40 if (pendingEvents == null) return; 40 _pending = null;
41 _pending = pendingEvents; 41 return events;
42 if (!pendingEvents.isEmpty) { 42 }
43 _state |= _STATE_HAS_PENDING; 43 void onData(void handleData(T event)) {
44 _pending.schedule(this); 44 if (handleData == null) handleData = _nullDataHandler;
45 _onData = _zone.registerUnaryCallback(DDC$RT.wrap((void f(T __u99)) {
46 void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam", """line 154, column 4 3 of dart:async/stream_impl.dart: """, x0 is T, false));
47 return f == null ? null : c;
48 }
49 , handleData, DDC$RT.type((__t102<T> _) {
50 }
51 ), __t100, "Wrap", """line 154, column 43 of dart:async/stream_impl.dart: """, h andleData is __t100));
52 }
53 void onError(Function handleError) {
54 if (handleError == null) handleError = _nullErrorHandler;
55 _onError = _registerErrorHandler(handleError, _zone);
56 }
57 void onDone(void handleDone()) {
58 if (handleDone == null) handleDone = _nullDoneHandler;
59 _onDone = _zone.registerCallback(handleDone);
60 }
61 void pause([Future resumeSignal]) {
62 if (_isCanceled) return; bool wasPaused = _isPaused;
63 bool wasInputPaused = _isInputPaused;
64 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
65 if (resumeSignal != null) resumeSignal.whenComplete(resume);
66 if (!wasPaused && _pending != null) _pending.cancelSchedule();
67 if (!wasInputPaused && !_inCallback) _guardCallback(_onPause);
68 }
69 void resume() {
70 if (_isCanceled) return; if (_isPaused) {
71 _decrementPauseCount();
72 if (!_isPaused) {
73 if (_hasPending && !_pending.isEmpty) {
74 _pending.schedule(this);
45 } 75 }
46 } 76 else {
47 _PendingEvents _extractPending() { 77 assert (_mayResumeInput); _state &= ~_STATE_INPUT_PAUSED;
48 assert(_isCanceled); 78 if (!_inCallback) _guardCallback(_onResume);
49 _PendingEvents events = _pending;
50 _pending = null;
51 return events;
52 }
53 void onData(void handleData(T event)) {
54 if (handleData == null) handleData = _nullDataHandler;
55 _onData = _zone.registerUnaryCallback(DDC$RT.wrap((void f(T __u99)) {
56 void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam",
57 """line 154, column 43 of dart:async/stream_impl.dart: """, x0 is T,
58 false));
59 return f == null ? null : c;
60 }, handleData, DDC$RT.type((__t102<T> _) {}), __t100, "Wrap",
61 """line 154, column 43 of dart:async/stream_impl.dart: """,
62 handleData is __t100));
63 }
64 void onError(Function handleError) {
65 if (handleError == null) handleError = _nullErrorHandler;
66 _onError = _registerErrorHandler(handleError, _zone);
67 }
68 void onDone(void handleDone()) {
69 if (handleDone == null) handleDone = _nullDoneHandler;
70 _onDone = _zone.registerCallback(handleDone);
71 }
72 void pause([Future resumeSignal]) {
73 if (_isCanceled) return;
74 bool wasPaused = _isPaused;
75 bool wasInputPaused = _isInputPaused;
76 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
77 if (resumeSignal != null) resumeSignal.whenComplete(resume);
78 if (!wasPaused && _pending != null) _pending.cancelSchedule();
79 if (!wasInputPaused && !_inCallback) _guardCallback(_onPause);
80 }
81 void resume() {
82 if (_isCanceled) return;
83 if (_isPaused) {
84 _decrementPauseCount();
85 if (!_isPaused) {
86 if (_hasPending && !_pending.isEmpty) {
87 _pending.schedule(this);
88 } else {
89 assert(_mayResumeInput);
90 _state &= ~_STATE_INPUT_PAUSED;
91 if (!_inCallback) _guardCallback(_onResume);
92 }
93 }
94 } 79 }
95 } 80 }
96 Future cancel() { 81 }
97 _state &= ~_STATE_WAIT_FOR_CANCEL; 82 }
98 if (_isCanceled) return _cancelFuture; 83 Future cancel() {
99 _cancel(); 84 _state &= ~_STATE_WAIT_FOR_CANCEL;
100 return _cancelFuture; 85 if (_isCanceled) return _cancelFuture;
101 } 86 _cancel();
102 Future asFuture([var futureValue]) { 87 return _cancelFuture;
103 _Future<T> result = new _Future<T>(); 88 }
104 _onDone = () { 89 Future asFuture([var futureValue]) {
105 result._complete(futureValue); 90 _Future<T> result = new _Future<T>();
106 }; 91 _onDone = () {
107 _onError = (error, stackTrace) { 92 result._complete(futureValue);
108 cancel(); 93 }
109 result._completeError(error, DDC$RT.cast(stackTrace, dynamic, StackTrace, 94 ;
110 "CastGeneral", 95 _onError = (error, stackTrace) {
111 """line 212, column 36 of dart:async/stream_impl.dart: """, 96 cancel();
112 stackTrace is StackTrace, true)); 97 result._completeError(error, DDC$RT.cast(stackTrace, dynamic, StackTrace, "Cast General", """line 212, column 36 of dart:async/stream_impl.dart: """, stackTrace is StackTrace, true));
113 }; 98 }
114 return result; 99 ;
115 } 100 return result;
116 bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0; 101 }
117 bool get _isClosed => (_state & _STATE_CLOSED) != 0; 102 bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
118 bool get _isCanceled => (_state & _STATE_CANCELED) != 0; 103 bool get _isClosed => (_state & _STATE_CLOSED) != 0;
119 bool get _waitsForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0; 104 bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
120 bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0; 105 bool get _waitsForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0;
121 bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0; 106 bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0;
122 bool get _isPaused => _state >= _STATE_PAUSE_COUNT; 107 bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0;
123 bool get _canFire => _state < _STATE_IN_CALLBACK; 108 bool get _isPaused => _state >= _STATE_PAUSE_COUNT;
124 bool get _mayResumeInput => 109 bool get _canFire => _state < _STATE_IN_CALLBACK;
125 !_isPaused && (_pending == null || _pending.isEmpty); 110 bool get _mayResumeInput => !_isPaused && (_pending == null || _pending.isEmpty );
126 bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0; 111 bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0;
127 bool get isPaused => _isPaused; 112 bool get isPaused => _isPaused;
128 void _cancel() { 113 void _cancel() {
129 _state |= _STATE_CANCELED; 114 _state |= _STATE_CANCELED;
130 if (_hasPending) { 115 if (_hasPending) {
131 _pending.cancelSchedule(); 116 _pending.cancelSchedule();
132 } 117 }
133 if (!_inCallback) _pending = null; 118 if (!_inCallback) _pending = null;
134 _cancelFuture = _onCancel(); 119 _cancelFuture = _onCancel();
135 } 120 }
136 void _incrementPauseCount() { 121 void _incrementPauseCount() {
137 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED; 122 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED;
138 } 123 }
139 void _decrementPauseCount() { 124 void _decrementPauseCount() {
140 assert(_isPaused); 125 assert (_isPaused); _state -= _STATE_PAUSE_COUNT;
141 _state -= _STATE_PAUSE_COUNT; 126 }
142 } 127 void _add(T data) {
143 void _add(T data) { 128 assert (!_isClosed); if (_isCanceled) return; if (_canFire) {
144 assert(!_isClosed); 129 _sendData(data);
145 if (_isCanceled) return; 130 }
146 if (_canFire) { 131 else {
147 _sendData(data); 132 _addPending(new _DelayedData(data));
148 } else { 133 }
149 _addPending(new _DelayedData(data)); 134 }
150 } 135 void _addError(Object error, StackTrace stackTrace) {
151 } 136 if (_isCanceled) return; if (_canFire) {
152 void _addError(Object error, StackTrace stackTrace) { 137 _sendError(error, stackTrace);
153 if (_isCanceled) return; 138 }
154 if (_canFire) { 139 else {
155 _sendError(error, stackTrace); 140 _addPending(new _DelayedError(error, stackTrace));
156 } else { 141 }
157 _addPending(new _DelayedError(error, stackTrace)); 142 }
158 } 143 void _close() {
159 } 144 assert (!_isClosed); if (_isCanceled) return; _state |= _STATE_CLOSED;
160 void _close() { 145 if (_canFire) {
161 assert(!_isClosed); 146 _sendDone();
162 if (_isCanceled) return; 147 }
163 _state |= _STATE_CLOSED; 148 else {
164 if (_canFire) { 149 _addPending(const _DelayedDone());
165 _sendDone(); 150 }
166 } else { 151 }
167 _addPending(const _DelayedDone()); 152 void _onPause() {
168 } 153 assert (_isInputPaused);}
169 } 154 void _onResume() {
170 void _onPause() { 155 assert (!_isInputPaused);}
171 assert(_isInputPaused); 156 Future _onCancel() {
172 } 157 assert (_isCanceled); return null;
173 void _onResume() { 158 }
174 assert(!_isInputPaused); 159 void _addPending(_DelayedEvent event) {
175 } 160 _StreamImplEvents pending = DDC$RT.cast(_pending, _PendingEvents, _StreamImplEve nts, "CastGeneral", """line 322, column 33 of dart:async/stream_impl.dart: """, _pending is _StreamImplEvents, true);
176 Future _onCancel() { 161 if (_pending == null) pending = _pending = new _StreamImplEvents();
177 assert(_isCanceled); 162 pending.add(event);
178 return null; 163 if (!_hasPending) {
179 } 164 _state |= _STATE_HAS_PENDING;
180 void _addPending(_DelayedEvent event) { 165 if (!_isPaused) {
181 _StreamImplEvents pending = DDC$RT.cast(_pending, _PendingEvents, 166 _pending.schedule(this);
182 _StreamImplEvents, "CastGeneral", 167 }
183 """line 322, column 33 of dart:async/stream_impl.dart: """, 168 }
184 _pending is _StreamImplEvents, true); 169 }
185 if (_pending == null) pending = _pending = new _StreamImplEvents(); 170 void _sendData(T data) {
186 pending.add(event); 171 assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); bool wasInput Paused = _isInputPaused;
187 if (!_hasPending) { 172 _state |= _STATE_IN_CALLBACK;
188 _state |= _STATE_HAS_PENDING; 173 _zone.runUnaryGuarded(DDC$RT.wrap((void f(T __u104)) {
189 if (!_isPaused) { 174 void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam", """line 341, column 2 7 of dart:async/stream_impl.dart: """, x0 is T, false));
190 _pending.schedule(this); 175 return f == null ? null : c;
191 } 176 }
192 } 177 , _onData, DDC$RT.type((__t102<T> _) {
193 } 178 }
194 void _sendData(T data) { 179 ), __t100, "Wrap", """line 341, column 27 of dart:async/stream_impl.dart: """, _ onData is __t100), data);
195 assert(!_isCanceled); 180 _state &= ~_STATE_IN_CALLBACK;
196 assert(!_isPaused); 181 _checkState(wasInputPaused);
197 assert(!_inCallback); 182 }
198 bool wasInputPaused = _isInputPaused; 183 void _sendError(var error, StackTrace stackTrace) {
199 _state |= _STATE_IN_CALLBACK; 184 assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); bool wasInput Paused = _isInputPaused;
200 _zone.runUnaryGuarded(DDC$RT.wrap((void f(T __u104)) { 185 void sendError() {
201 void c(T x0) => f(DDC$RT.cast(x0, dynamic, T, "CastParam", 186 if (_isCanceled && !_waitsForCancel) return; _state |= _STATE_IN_CALLBACK;
202 """line 341, column 27 of dart:async/stream_impl.dart: """, x0 is T, 187 if (_onError is ZoneBinaryCallback) {
203 false)); 188 _zone.runBinaryGuarded(DDC$RT.cast(_onError, Function, __t105, "CastGeneral", """line 358, column 32 of dart:async/stream_impl.dart: """, _onError is __t105, false), error, stackTrace);
204 return f == null ? null : c; 189 }
205 }, _onData, DDC$RT.type((__t102<T> _) {}), __t100, "Wrap", 190 else {
206 """line 341, column 27 of dart:async/stream_impl.dart: """, 191 _zone.runUnaryGuarded(DDC$RT.cast(_onError, Function, __t100, "CastGeneral", " ""line 360, column 31 of dart:async/stream_impl.dart: """, _onError is __t100, f alse), error);
207 _onData is __t100), data); 192 }
208 _state &= ~_STATE_IN_CALLBACK; 193 _state &= ~_STATE_IN_CALLBACK;
209 _checkState(wasInputPaused); 194 }
210 } 195 if (_cancelOnError) {
211 void _sendError(var error, StackTrace stackTrace) { 196 _state |= _STATE_WAIT_FOR_CANCEL;
212 assert(!_isCanceled); 197 _cancel();
213 assert(!_isPaused); 198 if (_cancelFuture is Future) {
214 assert(!_inCallback); 199 _cancelFuture.whenComplete(sendError);
215 bool wasInputPaused = _isInputPaused; 200 }
216 void sendError() { 201 else {
217 if (_isCanceled && !_waitsForCancel) return; 202 sendError();
218 _state |= _STATE_IN_CALLBACK; 203 }
219 if (_onError is ZoneBinaryCallback) { 204 }
220 _zone.runBinaryGuarded(DDC$RT.cast(_onError, Function, __t105, 205 else {
221 "CastGeneral", 206 sendError();
222 """line 358, column 32 of dart:async/stream_impl.dart: """, 207 _checkState(wasInputPaused);
223 _onError is __t105, false), error, stackTrace); 208 }
224 } else { 209 }
225 _zone.runUnaryGuarded(DDC$RT.cast(_onError, Function, __t100, 210 void _sendDone() {
226 "CastGeneral", 211 assert (!_isCanceled); assert (!_isPaused); assert (!_inCallback); void sendDone () {
227 """line 360, column 31 of dart:async/stream_impl.dart: """, 212 if (!_waitsForCancel) return; _state |= (_STATE_CANCELED | _STATE_CLOSED | _STAT E_IN_CALLBACK);
228 _onError is __t100, false), error); 213 _zone.runGuarded(_onDone);
229 } 214 _state &= ~_STATE_IN_CALLBACK;
230 _state &= ~_STATE_IN_CALLBACK; 215 }
231 } 216 _cancel();
232 if (_cancelOnError) { 217 _state |= _STATE_WAIT_FOR_CANCEL;
233 _state |= _STATE_WAIT_FOR_CANCEL; 218 if (_cancelFuture is Future) {
234 _cancel(); 219 _cancelFuture.whenComplete(sendDone);
235 if (_cancelFuture is Future) { 220 }
236 _cancelFuture.whenComplete(sendError); 221 else {
237 } else { 222 sendDone();
238 sendError(); 223 }
239 } 224 }
240 } else { 225 void _guardCallback(callback) {
241 sendError(); 226 assert (!_inCallback); bool wasInputPaused = _isInputPaused;
242 _checkState(wasInputPaused); 227 _state |= _STATE_IN_CALLBACK;
243 } 228 callback();
244 } 229 _state &= ~_STATE_IN_CALLBACK;
245 void _sendDone() { 230 _checkState(wasInputPaused);
246 assert(!_isCanceled); 231 }
247 assert(!_isPaused); 232 void _checkState(bool wasInputPaused) {
248 assert(!_inCallback); 233 assert (!_inCallback); if (_hasPending && _pending.isEmpty) {
249 void sendDone() { 234 _state &= ~_STATE_HAS_PENDING;
250 if (!_waitsForCancel) return; 235 if (_isInputPaused && _mayResumeInput) {
251 _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK); 236 _state &= ~_STATE_INPUT_PAUSED;
252 _zone.runGuarded(_onDone); 237 }
253 _state &= ~_STATE_IN_CALLBACK; 238 }
254 } 239 while (true) {
255 _cancel(); 240 if (_isCanceled) {
256 _state |= _STATE_WAIT_FOR_CANCEL; 241 _pending = null;
257 if (_cancelFuture is Future) { 242 return;}
258 _cancelFuture.whenComplete(sendDone); 243 bool isInputPaused = _isInputPaused;
259 } else { 244 if (wasInputPaused == isInputPaused) break;
260 sendDone(); 245 _state ^= _STATE_IN_CALLBACK;
261 } 246 if (isInputPaused) {
262 } 247 _onPause();
263 void _guardCallback(callback) { 248 }
264 assert(!_inCallback); 249 else {
265 bool wasInputPaused = _isInputPaused; 250 _onResume();
266 _state |= _STATE_IN_CALLBACK; 251 }
267 callback(); 252 _state &= ~_STATE_IN_CALLBACK;
268 _state &= ~_STATE_IN_CALLBACK; 253 wasInputPaused = isInputPaused;
269 _checkState(wasInputPaused); 254 }
270 } 255 if (_hasPending && !_isPaused) {
271 void _checkState(bool wasInputPaused) { 256 _pending.schedule(this);
272 assert(!_inCallback); 257 }
273 if (_hasPending && _pending.isEmpty) { 258 }
274 _state &= ~_STATE_HAS_PENDING; 259 }
275 if (_isInputPaused && _mayResumeInput) { 260 abstract class _StreamImpl<T> extends Stream<T> {StreamSubscription<T> listen(v oid onData(T data), {
276 _state &= ~_STATE_INPUT_PAUSED; 261 Function onError, void onDone(), bool cancelOnError}
277 } 262 ) {
278 } 263 cancelOnError = identical(true, cancelOnError);
279 while (true) { 264 StreamSubscription subscription = _createSubscription(onData, onError, onDone, cancelOnError);
280 if (_isCanceled) { 265 _onListen(subscription);
281 _pending = null; 266 return DDC$RT.cast(subscription, DDC$RT.type((StreamSubscription<dynamic> _) {
282 return; 267 }
283 } 268 ), DDC$RT.type((StreamSubscription<T> _) {
284 bool isInputPaused = _isInputPaused; 269 }
285 if (wasInputPaused == isInputPaused) break; 270 ), "CastDynamic", """line 476, column 12 of dart:async/stream_impl.dart: """, su bscription is StreamSubscription<T>, false);
286 _state ^= _STATE_IN_CALLBACK; 271 }
287 if (isInputPaused) { 272 _BufferingStreamSubscription<T> _createSubscription(void onData(T data), Functi on onError, void onDone(), bool cancelOnError) {
288 _onPause(); 273 return new _BufferingStreamSubscription<T>(onData, onError, onDone, cancelOnErro r);
289 } else { 274 }
290 _onResume(); 275 void _onListen(StreamSubscription subscription) {
291 } 276 }
292 _state &= ~_STATE_IN_CALLBACK; 277 }
293 wasInputPaused = isInputPaused; 278 typedef _PendingEvents _EventGenerator();
294 } 279 class _GeneratedStreamImpl<T> extends _StreamImpl<T> {final _EventGenerator _pe nding;
295 if (_hasPending && !_isPaused) { 280 bool _isUsed = false;
296 _pending.schedule(this); 281 _GeneratedStreamImpl(this._pending);
297 } 282 StreamSubscription _createSubscription(void onData(T data), Function onError, v oid onDone(), bool cancelOnError) {
298 } 283 if (_isUsed) throw new StateError("Stream has already been listened to.");
299 } 284 _isUsed = true;
300 abstract class _StreamImpl<T> extends Stream<T> { 285 return new _BufferingStreamSubscription(onData, onError, onDone, cancelOnError) .._setPendingEvents(_pending());
301 StreamSubscription<T> listen(void onData(T data), 286 }
302 {Function onError, void onDone(), bool cancelOnError}) { 287 }
303 cancelOnError = identical(true, cancelOnError); 288 class _IterablePendingEvents<T> extends _PendingEvents {Iterator<T> _iterator;
304 StreamSubscription subscription = 289 _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator;
305 _createSubscription(onData, onError, onDone, cancelOnError); 290 bool get isEmpty => _iterator == null;
306 _onListen(subscription); 291 void handleNext(_EventDispatch dispatch) {
307 return DDC$RT.cast(subscription, 292 if (_iterator == null) {
308 DDC$RT.type((StreamSubscription<dynamic> _) {}), 293 throw new StateError("No events pending.");
309 DDC$RT.type((StreamSubscription<T> _) {}), "CastDynamic", 294 }
310 """line 476, column 12 of dart:async/stream_impl.dart: """, 295 bool isDone;
311 subscription is StreamSubscription<T>, false); 296 try {
312 } 297 isDone = !_iterator.moveNext();
313 _BufferingStreamSubscription<T> _createSubscription(void onData(T data), 298 }
314 Function onError, void onDone(), bool cancelOnError) { 299 catch (e, s) {
315 return new _BufferingStreamSubscription<T>( 300 _iterator = null;
316 onData, onError, onDone, cancelOnError); 301 dispatch._sendError(e, s);
317 } 302 return;}
318 void _onListen(StreamSubscription subscription) {} 303 if (!isDone) {
319 } 304 dispatch._sendData(_iterator.current);
320 typedef _PendingEvents _EventGenerator(); 305 }
321 class _GeneratedStreamImpl<T> extends _StreamImpl<T> { 306 else {
322 final _EventGenerator _pending; 307 _iterator = null;
323 bool _isUsed = false; 308 dispatch._sendDone();
324 _GeneratedStreamImpl(this._pending); 309 }
325 StreamSubscription _createSubscription(void onData(T data), Function onError, 310 }
326 void onDone(), bool cancelOnError) { 311 void clear() {
327 if (_isUsed) throw new StateError("Stream has already been listened to."); 312 if (isScheduled) cancelSchedule();
328 _isUsed = true; 313 _iterator = null;
329 return new _BufferingStreamSubscription( 314 }
330 onData, onError, onDone, cancelOnError).._setPendingEvents(_pending()); 315 }
331 } 316 typedef void _DataHandler<T>(T value);
332 } 317 typedef void _DoneHandler();
333 class _IterablePendingEvents<T> extends _PendingEvents { 318 void _nullDataHandler(var value) {
334 Iterator<T> _iterator; 319 }
335 _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator; 320 void _nullErrorHandler(error, [StackTrace stackTrace]) {
336 bool get isEmpty => _iterator == null; 321 Zone.current.handleUncaughtError(error, stackTrace);
337 void handleNext(_EventDispatch dispatch) { 322 }
338 if (_iterator == null) { 323 void _nullDoneHandler() {
339 throw new StateError("No events pending."); 324 }
340 } 325 abstract class _DelayedEvent {_DelayedEvent next;
341 bool isDone; 326 void perform(_EventDispatch dispatch);
342 try { 327 }
343 isDone = !_iterator.moveNext(); 328 class _DelayedData<T> extends _DelayedEvent {final T value;
344 } catch (e, s) { 329 _DelayedData(this.value);
345 _iterator = null; 330 void perform(_EventDispatch<T> dispatch) {
346 dispatch._sendError(e, s); 331 dispatch._sendData(value);
347 return; 332 }
348 } 333 }
349 if (!isDone) { 334 class _DelayedError extends _DelayedEvent {final error;
350 dispatch._sendData(_iterator.current); 335 final StackTrace stackTrace;
351 } else { 336 _DelayedError(this.error, this.stackTrace);
352 _iterator = null; 337 void perform(_EventDispatch dispatch) {
353 dispatch._sendDone(); 338 dispatch._sendError(error, stackTrace);
354 } 339 }
355 } 340 }
356 void clear() { 341 class _DelayedDone implements _DelayedEvent {const _DelayedDone();
357 if (isScheduled) cancelSchedule(); 342 void perform(_EventDispatch dispatch) {
358 _iterator = null; 343 dispatch._sendDone();
359 } 344 }
360 } 345 _DelayedEvent get next => null;
361 typedef void _DataHandler<T>(T value); 346 void set next(_DelayedEvent _) {
362 typedef void _DoneHandler(); 347 throw new StateError("No events after a done.");
363 void _nullDataHandler(var value) {} 348 }
364 void _nullErrorHandler(error, [StackTrace stackTrace]) { 349 }
365 Zone.current.handleUncaughtError(error, stackTrace); 350 abstract class _PendingEvents {static const int _STATE_UNSCHEDULED = 0;
366 } 351 static const int _STATE_SCHEDULED = 1;
367 void _nullDoneHandler() {} 352 static const int _STATE_CANCELED = 3;
368 abstract class _DelayedEvent { 353 int _state = _STATE_UNSCHEDULED;
369 _DelayedEvent next; 354 bool get isEmpty;
370 void perform(_EventDispatch dispatch); 355 bool get isScheduled => _state == _STATE_SCHEDULED;
371 } 356 bool get _eventScheduled => _state >= _STATE_SCHEDULED;
372 class _DelayedData<T> extends _DelayedEvent { 357 void schedule(_EventDispatch dispatch) {
373 final T value; 358 if (isScheduled) return; assert (!isEmpty); if (_eventScheduled) {
374 _DelayedData(this.value); 359 assert (_state == _STATE_CANCELED); _state = _STATE_SCHEDULED;
375 void perform(_EventDispatch<T> dispatch) { 360 return;}
376 dispatch._sendData(value); 361 scheduleMicrotask(() {
377 } 362 int oldState = _state;
378 } 363 _state = _STATE_UNSCHEDULED;
379 class _DelayedError extends _DelayedEvent { 364 if (oldState == _STATE_CANCELED) return; handleNext(dispatch);
380 final error; 365 }
381 final StackTrace stackTrace; 366 );
382 _DelayedError(this.error, this.stackTrace); 367 _state = _STATE_SCHEDULED;
383 void perform(_EventDispatch dispatch) { 368 }
384 dispatch._sendError(error, stackTrace); 369 void cancelSchedule() {
385 } 370 if (isScheduled) _state = _STATE_CANCELED;
386 } 371 }
387 class _DelayedDone implements _DelayedEvent { 372 void handleNext(_EventDispatch dispatch);
388 const _DelayedDone(); 373 void clear();
389 void perform(_EventDispatch dispatch) { 374 }
390 dispatch._sendDone(); 375 class _StreamImplEvents extends _PendingEvents {_DelayedEvent firstPendingEvent = null;
391 } 376 _DelayedEvent lastPendingEvent = null;
392 _DelayedEvent get next => null; 377 bool get isEmpty => lastPendingEvent == null;
393 void set next(_DelayedEvent _) { 378 void add(_DelayedEvent event) {
394 throw new StateError("No events after a done."); 379 if (lastPendingEvent == null) {
395 } 380 firstPendingEvent = lastPendingEvent = event;
396 } 381 }
397 abstract class _PendingEvents { 382 else {
398 static const int _STATE_UNSCHEDULED = 0; 383 lastPendingEvent = lastPendingEvent.next = event;
399 static const int _STATE_SCHEDULED = 1; 384 }
400 static const int _STATE_CANCELED = 3; 385 }
401 int _state = _STATE_UNSCHEDULED; 386 void handleNext(_EventDispatch dispatch) {
402 bool get isEmpty; 387 assert (!isScheduled); _DelayedEvent event = firstPendingEvent;
403 bool get isScheduled => _state == _STATE_SCHEDULED; 388 firstPendingEvent = event.next;
404 bool get _eventScheduled => _state >= _STATE_SCHEDULED; 389 if (firstPendingEvent == null) {
405 void schedule(_EventDispatch dispatch) { 390 lastPendingEvent = null;
406 if (isScheduled) return; 391 }
407 assert(!isEmpty); 392 event.perform(dispatch);
408 if (_eventScheduled) { 393 }
409 assert(_state == _STATE_CANCELED); 394 void clear() {
410 _state = _STATE_SCHEDULED; 395 if (isScheduled) cancelSchedule();
411 return; 396 firstPendingEvent = lastPendingEvent = null;
412 } 397 }
413 scheduleMicrotask(() { 398 }
414 int oldState = _state; 399 class _BroadcastLinkedList {_BroadcastLinkedList _next;
415 _state = _STATE_UNSCHEDULED; 400 _BroadcastLinkedList _previous;
416 if (oldState == _STATE_CANCELED) return; 401 void _unlink() {
417 handleNext(dispatch); 402 _previous._next = _next;
418 }); 403 _next._previous = _previous;
419 _state = _STATE_SCHEDULED; 404 _next = _previous = this;
420 } 405 }
421 void cancelSchedule() { 406 void _insertBefore(_BroadcastLinkedList newNext) {
422 if (isScheduled) _state = _STATE_CANCELED; 407 _BroadcastLinkedList newPrevious = newNext._previous;
423 } 408 newPrevious._next = this;
424 void handleNext(_EventDispatch dispatch); 409 newNext._previous = _previous;
425 void clear(); 410 _previous._next = newNext;
426 } 411 _previous = newPrevious;
427 class _StreamImplEvents extends _PendingEvents { 412 }
428 _DelayedEvent firstPendingEvent = null; 413 }
429 _DelayedEvent lastPendingEvent = null; 414 typedef void _broadcastCallback(StreamSubscription subscription);
430 bool get isEmpty => lastPendingEvent == null; 415 class _DoneStreamSubscription<T> implements StreamSubscription<T> {static const int _DONE_SENT = 1;
431 void add(_DelayedEvent event) { 416 static const int _SCHEDULED = 2;
432 if (lastPendingEvent == null) { 417 static const int _PAUSED = 4;
433 firstPendingEvent = lastPendingEvent = event; 418 final Zone _zone;
434 } else { 419 int _state = 0;
435 lastPendingEvent = lastPendingEvent.next = event; 420 _DoneHandler _onDone;
436 } 421 _DoneStreamSubscription(this._onDone) : _zone = Zone.current {
437 } 422 _schedule();
438 void handleNext(_EventDispatch dispatch) { 423 }
439 assert(!isScheduled); 424 bool get _isSent => (_state & _DONE_SENT) != 0;
440 _DelayedEvent event = firstPendingEvent; 425 bool get _isScheduled => (_state & _SCHEDULED) != 0;
441 firstPendingEvent = event.next; 426 bool get isPaused => _state >= _PAUSED;
442 if (firstPendingEvent == null) { 427 void _schedule() {
443 lastPendingEvent = null; 428 if (_isScheduled) return; _zone.scheduleMicrotask(_sendDone);
444 } 429 _state |= _SCHEDULED;
445 event.perform(dispatch); 430 }
446 } 431 void onData(void handleData(T data)) {
447 void clear() { 432 }
448 if (isScheduled) cancelSchedule(); 433 void onError(Function handleError) {
449 firstPendingEvent = lastPendingEvent = null; 434 }
450 } 435 void onDone(void handleDone()) {
451 } 436 _onDone = handleDone;
452 class _BroadcastLinkedList { 437 }
453 _BroadcastLinkedList _next; 438 void pause([Future resumeSignal]) {
454 _BroadcastLinkedList _previous; 439 _state += _PAUSED;
455 void _unlink() { 440 if (resumeSignal != null) resumeSignal.whenComplete(resume);
456 _previous._next = _next; 441 }
457 _next._previous = _previous; 442 void resume() {
458 _next = _previous = this; 443 if (isPaused) {
459 } 444 _state -= _PAUSED;
460 void _insertBefore(_BroadcastLinkedList newNext) { 445 if (!isPaused && !_isSent) {
461 _BroadcastLinkedList newPrevious = newNext._previous; 446 _schedule();
462 newPrevious._next = this; 447 }
463 newNext._previous = _previous; 448 }
464 _previous._next = newNext; 449 }
465 _previous = newPrevious; 450 Future cancel() => null;
466 } 451 Future asFuture([futureValue]) {
467 } 452 _Future result = new _Future();
468 typedef void _broadcastCallback(StreamSubscription subscription); 453 _onDone = () {
469 class _DoneStreamSubscription<T> implements StreamSubscription<T> { 454 result._completeWithValue(null);
470 static const int _DONE_SENT = 1; 455 }
471 static const int _SCHEDULED = 2; 456 ;
472 static const int _PAUSED = 4; 457 return result;
473 final Zone _zone; 458 }
474 int _state = 0; 459 void _sendDone() {
475 _DoneHandler _onDone; 460 _state &= ~_SCHEDULED;
476 _DoneStreamSubscription(this._onDone) : _zone = Zone.current { 461 if (isPaused) return; _state |= _DONE_SENT;
477 _schedule(); 462 if (_onDone != null) _zone.runGuarded(_onDone);
478 } 463 }
479 bool get _isSent => (_state & _DONE_SENT) != 0; 464 }
480 bool get _isScheduled => (_state & _SCHEDULED) != 0; 465 class _AsBroadcastStream<T> extends Stream<T> {final Stream<T> _source;
481 bool get isPaused => _state >= _PAUSED; 466 final _broadcastCallback _onListenHandler;
482 void _schedule() { 467 final _broadcastCallback _onCancelHandler;
483 if (_isScheduled) return; 468 final Zone _zone;
484 _zone.scheduleMicrotask(_sendDone); 469 _AsBroadcastStreamController<T> _controller;
485 _state |= _SCHEDULED; 470 StreamSubscription<T> _subscription;
486 } 471 _AsBroadcastStream(this._source, void onListenHandler(StreamSubscription subscr iption), void onCancelHandler(StreamSubscription subscription)) : _onListenHandl er = Zone.current.registerUnaryCallback(DDC$RT.wrap((void f(StreamSubscription<d ynamic> __u108)) {
487 void onData(void handleData(T data)) {} 472 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type ((StreamSubscription<dynamic> _) {
488 void onError(Function handleError) {} 473 }
489 void onDone(void handleDone()) { 474 ), "CastParam", """line 813, column 63 of dart:async/stream_impl.dart: """, x0 i s StreamSubscription<dynamic>, true));
490 _onDone = handleDone; 475 return f == null ? null : c;
491 } 476 }
492 void pause([Future resumeSignal]) { 477 , onListenHandler, __t109, __t100, "Wrap", """line 813, column 63 of dart:async/ stream_impl.dart: """, onListenHandler is __t100)), _onCancelHandler = Zone.curr ent.registerUnaryCallback(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u111 )) {
493 _state += _PAUSED; 478 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type ((StreamSubscription<dynamic> _) {
494 if (resumeSignal != null) resumeSignal.whenComplete(resume); 479 }
495 } 480 ), "CastParam", """line 814, column 63 of dart:async/stream_impl.dart: """, x0 i s StreamSubscription<dynamic>, true));
496 void resume() { 481 return f == null ? null : c;
497 if (isPaused) { 482 }
498 _state -= _PAUSED; 483 , onCancelHandler, __t109, __t100, "Wrap", """line 814, column 63 of dart:async/ stream_impl.dart: """, onCancelHandler is __t100)), _zone = Zone.current {
499 if (!isPaused && !_isSent) { 484 _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
500 _schedule(); 485 }
501 } 486 bool get isBroadcast => true;
502 } 487 StreamSubscription<T> listen(void onData(T data), {
503 } 488 Function onError, void onDone(), bool cancelOnError}
504 Future cancel() => null; 489 ) {
505 Future asFuture([futureValue]) { 490 if (_controller == null || _controller.isClosed) {
506 _Future result = new _Future(); 491 return new _DoneStreamSubscription<T>(onDone);
507 _onDone = () { 492 }
508 result._completeWithValue(null); 493 if (_subscription == null) {
509 }; 494 _subscription = _source.listen(_controller.add, onError: _controller.addError, o nDone: _controller.close);
510 return result; 495 }
511 } 496 cancelOnError = identical(true, cancelOnError);
512 void _sendDone() { 497 return _controller._subscribe(onData, onError, onDone, cancelOnError);
513 _state &= ~_SCHEDULED; 498 }
514 if (isPaused) return; 499 void _onCancel() {
515 _state |= _DONE_SENT; 500 bool shutdown = (_controller == null) || _controller.isClosed;
516 if (_onDone != null) _zone.runGuarded(_onDone); 501 if (_onCancelHandler != null) {
517 } 502 _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u112)) {
518 } 503 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type ((StreamSubscription<dynamic> _) {
519 class _AsBroadcastStream<T> extends Stream<T> { 504 }
520 final Stream<T> _source; 505 ), "CastParam", """line 842, column 22 of dart:async/stream_impl.dart: """, x0 i s StreamSubscription<dynamic>, true));
521 final _broadcastCallback _onListenHandler; 506 return f == null ? null : c;
522 final _broadcastCallback _onCancelHandler; 507 }
523 final Zone _zone; 508 , _onCancelHandler, __t109, __t100, "Wrap", """line 842, column 22 of dart:async /stream_impl.dart: """, _onCancelHandler is __t100), new _BroadcastSubscriptionW rapper(this));
524 _AsBroadcastStreamController<T> _controller; 509 }
525 StreamSubscription<T> _subscription; 510 if (shutdown) {
526 _AsBroadcastStream(this._source, 511 if (_subscription != null) {
527 void onListenHandler(StreamSubscription subscription), 512 _subscription.cancel();
528 void onCancelHandler(StreamSubscription subscription)) 513 _subscription = null;
529 : _onListenHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap( 514 }
530 (void f(StreamSubscription<dynamic> __u108)) { 515 }
531 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, 516 }
532 DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam", 517 void _onListen() {
533 """line 813, column 63 of dart:async/stream_impl.dart: """, 518 if (_onListenHandler != null) {
534 x0 is StreamSubscription<dynamic>, true)); 519 _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u113)) {
535 return f == null ? null : c; 520 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, DDC$RT.type ((StreamSubscription<dynamic> _) {
536 }, onListenHandler, __t109, __t100, "Wrap", 521 }
537 """line 813, column 63 of dart:async/stream_impl.dart: """, 522 ), "CastParam", """line 854, column 22 of dart:async/stream_impl.dart: """, x0 i s StreamSubscription<dynamic>, true));
538 onListenHandler is __t100)), 523 return f == null ? null : c;
539 _onCancelHandler = Zone.current.registerUnaryCallback(DDC$RT.wrap( 524 }
540 (void f(StreamSubscription<dynamic> __u111)) { 525 , _onListenHandler, __t109, __t100, "Wrap", """line 854, column 22 of dart:async /stream_impl.dart: """, _onListenHandler is __t100), new _BroadcastSubscriptionW rapper(this));
541 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, 526 }
542 DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam", 527 }
543 """line 814, column 63 of dart:async/stream_impl.dart: """, 528 void _cancelSubscription() {
544 x0 is StreamSubscription<dynamic>, true)); 529 if (_subscription == null) return; StreamSubscription subscription = _subscripti on;
545 return f == null ? null : c; 530 _subscription = null;
546 }, onCancelHandler, __t109, __t100, "Wrap", 531 _controller = null;
547 """line 814, column 63 of dart:async/stream_impl.dart: """, 532 subscription.cancel();
548 onCancelHandler is __t100)), 533 }
549 _zone = Zone.current { 534 void _pauseSubscription(Future resumeSignal) {
550 _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel); 535 if (_subscription == null) return; _subscription.pause(resumeSignal);
551 } 536 }
552 bool get isBroadcast => true; 537 void _resumeSubscription() {
553 StreamSubscription<T> listen(void onData(T data), 538 if (_subscription == null) return; _subscription.resume();
554 {Function onError, void onDone(), bool cancelOnError}) { 539 }
555 if (_controller == null || _controller.isClosed) { 540 bool get _isSubscriptionPaused {
556 return new _DoneStreamSubscription<T>(onDone); 541 if (_subscription == null) return false;
557 } 542 return _subscription.isPaused;
558 if (_subscription == null) { 543 }
559 _subscription = _source.listen(_controller.add, 544 }
560 onError: _controller.addError, onDone: _controller.close); 545 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> {final _AsBroadcastStream _stream;
561 } 546 _BroadcastSubscriptionWrapper(this._stream);
562 cancelOnError = identical(true, cancelOnError); 547 void onData(void handleData(T data)) {
563 return _controller._subscribe(onData, onError, onDone, cancelOnError); 548 throw new UnsupportedError("Cannot change handlers of asBroadcastStream source s ubscription.");
564 } 549 }
565 void _onCancel() { 550 void onError(void handleError(Object data)) {
566 bool shutdown = (_controller == null) || _controller.isClosed; 551 throw new UnsupportedError("Cannot change handlers of asBroadcastStream source s ubscription.");
567 if (_onCancelHandler != null) { 552 }
568 _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u112)) { 553 void onDone(void handleDone()) {
569 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, 554 throw new UnsupportedError("Cannot change handlers of asBroadcastStream source s ubscription.");
570 DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam", 555 }
571 """line 842, column 22 of dart:async/stream_impl.dart: """, 556 void pause([Future resumeSignal]) {
572 x0 is StreamSubscription<dynamic>, true)); 557 _stream._pauseSubscription(resumeSignal);
573 return f == null ? null : c; 558 }
574 }, _onCancelHandler, __t109, __t100, "Wrap", 559 void resume() {
575 """line 842, column 22 of dart:async/stream_impl.dart: """, 560 _stream._resumeSubscription();
576 _onCancelHandler is __t100), new _BroadcastSubscriptionWrapper(this)); 561 }
577 } 562 Future cancel() {
578 if (shutdown) { 563 _stream._cancelSubscription();
579 if (_subscription != null) { 564 return null;
580 _subscription.cancel(); 565 }
581 _subscription = null; 566 bool get isPaused {
582 } 567 return _stream._isSubscriptionPaused;
583 } 568 }
584 } 569 Future asFuture([var futureValue]) {
585 void _onListen() { 570 throw new UnsupportedError("Cannot change handlers of asBroadcastStream source s ubscription.");
586 if (_onListenHandler != null) { 571 }
587 _zone.runUnary(DDC$RT.wrap((void f(StreamSubscription<dynamic> __u113)) { 572 }
588 void c(StreamSubscription<dynamic> x0) => f(DDC$RT.cast(x0, dynamic, 573 class _StreamIteratorImpl<T> implements StreamIterator<T> {static const int _ST ATE_FOUND = 0;
589 DDC$RT.type((StreamSubscription<dynamic> _) {}), "CastParam", 574 static const int _STATE_DONE = 1;
590 """line 854, column 22 of dart:async/stream_impl.dart: """, 575 static const int _STATE_MOVING = 2;
591 x0 is StreamSubscription<dynamic>, true)); 576 static const int _STATE_EXTRA_DATA = 3;
592 return f == null ? null : c; 577 static const int _STATE_EXTRA_ERROR = 4;
593 }, _onListenHandler, __t109, __t100, "Wrap", 578 static const int _STATE_EXTRA_DONE = 5;
594 """line 854, column 22 of dart:async/stream_impl.dart: """, 579 StreamSubscription _subscription;
595 _onListenHandler is __t100), new _BroadcastSubscriptionWrapper(this)); 580 T _current = ((__x114) => DDC$RT.cast(__x114, Null, T, "CastLiteral", """line 9 68, column 16 of dart:async/stream_impl.dart: """, __x114 is T, false))(null);
596 } 581 var _futureOrPrefetch = null;
597 } 582 int _state = _STATE_FOUND;
598 void _cancelSubscription() { 583 _StreamIteratorImpl(final Stream<T> stream) {
599 if (_subscription == null) return; 584 _subscription = stream.listen(_onData, onError: _onError, onDone: _onDone, cance lOnError: true);
600 StreamSubscription subscription = _subscription; 585 }
601 _subscription = null; 586 T get current => _current;
602 _controller = null; 587 Future<bool> moveNext() {
603 subscription.cancel(); 588 if (_state == _STATE_DONE) {
604 } 589 return new _Future<bool>.immediate(false);
605 void _pauseSubscription(Future resumeSignal) { 590 }
606 if (_subscription == null) return; 591 if (_state == _STATE_MOVING) {
607 _subscription.pause(resumeSignal); 592 throw new StateError("Already waiting for next.");
608 } 593 }
609 void _resumeSubscription() { 594 if (_state == _STATE_FOUND) {
610 if (_subscription == null) return; 595 _state = _STATE_MOVING;
611 _subscription.resume(); 596 _current = ((__x115) => DDC$RT.cast(__x115, Null, T, "CastLiteral", """line 998 , column 18 of dart:async/stream_impl.dart: """, __x115 is T, false))(null);
612 } 597 _futureOrPrefetch = new _Future<bool>();
613 bool get _isSubscriptionPaused { 598 return DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((Future<bool> _) {
614 if (_subscription == null) return false; 599 }
615 return _subscription.isPaused; 600 ), "CastGeneral", """line 1000, column 14 of dart:async/stream_impl.dart: """, _ futureOrPrefetch is Future<bool>, false);
616 } 601 }
617 } 602 else {
618 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> { 603 assert (_state >= _STATE_EXTRA_DATA); switch (_state) {case _STATE_EXTRA_DATA: _ state = _STATE_FOUND;
619 final _AsBroadcastStream _stream; 604 _current = DDC$RT.cast(_futureOrPrefetch, dynamic, T, "CastGeneral", """line 10 06, column 22 of dart:async/stream_impl.dart: """, _futureOrPrefetch is T, false );
620 _BroadcastSubscriptionWrapper(this._stream); 605 _futureOrPrefetch = null;
621 void onData(void handleData(T data)) { 606 _subscription.resume();
622 throw new UnsupportedError( 607 return new _Future<bool>.immediate(true);
623 "Cannot change handlers of asBroadcastStream source subscription."); 608 case _STATE_EXTRA_ERROR: AsyncError prefetch = DDC$RT.cast(_futureOrPrefetch, d ynamic, AsyncError, "CastGeneral", """line 1011, column 33 of dart:async/stream_ impl.dart: """, _futureOrPrefetch is AsyncError, true);
624 } 609 _clear();
625 void onError(void handleError(Object data)) { 610 return new _Future<bool>.immediateError(prefetch.error, prefetch.stackTrace);
626 throw new UnsupportedError( 611 case _STATE_EXTRA_DONE: _clear();
627 "Cannot change handlers of asBroadcastStream source subscription."); 612 return new _Future<bool>.immediate(false);
628 } 613 }
629 void onDone(void handleDone()) { 614 }
630 throw new UnsupportedError( 615 }
631 "Cannot change handlers of asBroadcastStream source subscription."); 616 void _clear() {
632 } 617 _subscription = null;
633 void pause([Future resumeSignal]) { 618 _futureOrPrefetch = null;
634 _stream._pauseSubscription(resumeSignal); 619 _current = ((__x116) => DDC$RT.cast(__x116, Null, T, "CastLiteral", """line 102 6, column 16 of dart:async/stream_impl.dart: """, __x116 is T, false))(null);
635 } 620 _state = _STATE_DONE;
636 void resume() { 621 }
637 _stream._resumeSubscription(); 622 Future cancel() {
638 } 623 StreamSubscription subscription = _subscription;
639 Future cancel() { 624 if (_state == _STATE_MOVING) {
640 _stream._cancelSubscription(); 625 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Fu ture<bool> _) {
641 return null; 626 }
642 } 627 ), "CastGeneral", """line 1033, column 31 of dart:async/stream_impl.dart: """, _ futureOrPrefetch is _Future<bool>, false);
643 bool get isPaused { 628 _clear();
644 return _stream._isSubscriptionPaused; 629 hasNext._complete(false);
645 } 630 }
646 Future asFuture([var futureValue]) { 631 else {
647 throw new UnsupportedError( 632 _clear();
648 "Cannot change handlers of asBroadcastStream source subscription."); 633 }
649 } 634 return subscription.cancel();
650 } 635 }
651 class _StreamIteratorImpl<T> implements StreamIterator<T> { 636 void _onData(T data) {
652 static const int _STATE_FOUND = 0; 637 if (_state == _STATE_MOVING) {
653 static const int _STATE_DONE = 1; 638 _current = data;
654 static const int _STATE_MOVING = 2; 639 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_F uture<bool> _) {
655 static const int _STATE_EXTRA_DATA = 3; 640 }
656 static const int _STATE_EXTRA_ERROR = 4; 641 ), "CastGeneral", """line 1045, column 31 of dart:async/stream_impl.dart: """, _ futureOrPrefetch is _Future<bool>, false);
657 static const int _STATE_EXTRA_DONE = 5; 642 _futureOrPrefetch = null;
658 StreamSubscription _subscription; 643 _state = _STATE_FOUND;
659 T _current = ((__x114) => DDC$RT.cast(__x114, Null, T, "CastLiteral", 644 hasNext._complete(true);
660 """line 968, column 16 of dart:async/stream_impl.dart: """, __x114 is T, 645 return;}
661 false))(null); 646 _subscription.pause();
662 var _futureOrPrefetch = null; 647 assert (_futureOrPrefetch == null); _futureOrPrefetch = data;
663 int _state = _STATE_FOUND; 648 _state = _STATE_EXTRA_DATA;
664 _StreamIteratorImpl(final Stream<T> stream) { 649 }
665 _subscription = stream.listen(_onData, 650 void _onError(Object error, [StackTrace stackTrace]) {
666 onError: _onError, onDone: _onDone, cancelOnError: true); 651 if (_state == _STATE_MOVING) {
667 } 652 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Fu ture<bool> _) {
668 T get current => _current; 653 }
669 Future<bool> moveNext() { 654 ), "CastGeneral", """line 1059, column 31 of dart:async/stream_impl.dart: """, _ futureOrPrefetch is _Future<bool>, false);
670 if (_state == _STATE_DONE) { 655 _clear();
671 return new _Future<bool>.immediate(false); 656 hasNext._completeError(error, stackTrace);
672 } 657 return;}
673 if (_state == _STATE_MOVING) { 658 _subscription.pause();
674 throw new StateError("Already waiting for next."); 659 assert (_futureOrPrefetch == null); _futureOrPrefetch = new AsyncError(error, s tackTrace);
675 } 660 _state = _STATE_EXTRA_ERROR;
676 if (_state == _STATE_FOUND) { 661 }
677 _state = _STATE_MOVING; 662 void _onDone() {
678 _current = ((__x115) => DDC$RT.cast(__x115, Null, T, "CastLiteral", 663 if (_state == _STATE_MOVING) {
679 """line 998, column 18 of dart:async/stream_impl.dart: """, 664 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic, DDC$RT.type((_Fu ture<bool> _) {
680 __x115 is T, false))(null); 665 }
681 _futureOrPrefetch = new _Future<bool>(); 666 ), "CastGeneral", """line 1073, column 31 of dart:async/stream_impl.dart: """, _ futureOrPrefetch is _Future<bool>, false);
682 return DDC$RT.cast(_futureOrPrefetch, dynamic, 667 _clear();
683 DDC$RT.type((Future<bool> _) {}), "CastGeneral", 668 hasNext._complete(false);
684 """line 1000, column 14 of dart:async/stream_impl.dart: """, 669 return;}
685 _futureOrPrefetch is Future<bool>, false); 670 _subscription.pause();
686 } else { 671 _futureOrPrefetch = null;
687 assert(_state >= _STATE_EXTRA_DATA); 672 _state = _STATE_EXTRA_DONE;
688 switch (_state) { 673 }
689 case _STATE_EXTRA_DATA: 674 }
690 _state = _STATE_FOUND; 675 typedef dynamic __t100(dynamic __u101);
691 _current = DDC$RT.cast(_futureOrPrefetch, dynamic, T, "CastGeneral", 676 typedef void __t102<T>(T __u103);
692 """line 1006, column 22 of dart:async/stream_impl.dart: """, 677 typedef dynamic __t105(dynamic __u106, dynamic __u107);
693 _futureOrPrefetch is T, false); 678 typedef void __t109(StreamSubscription<dynamic> __u110);
694 _futureOrPrefetch = null;
695 _subscription.resume();
696 return new _Future<bool>.immediate(true);
697 case _STATE_EXTRA_ERROR:
698 AsyncError prefetch = DDC$RT.cast(_futureOrPrefetch, dynamic,
699 AsyncError, "CastGeneral",
700 """line 1011, column 33 of dart:async/stream_impl.dart: """,
701 _futureOrPrefetch is AsyncError, true);
702 _clear();
703 return new _Future<bool>.immediateError(
704 prefetch.error, prefetch.stackTrace);
705 case _STATE_EXTRA_DONE:
706 _clear();
707 return new _Future<bool>.immediate(false);
708 }
709 }
710 }
711 void _clear() {
712 _subscription = null;
713 _futureOrPrefetch = null;
714 _current = ((__x116) => DDC$RT.cast(__x116, Null, T, "CastLiteral",
715 """line 1026, column 16 of dart:async/stream_impl.dart: """,
716 __x116 is T, false))(null);
717 _state = _STATE_DONE;
718 }
719 Future cancel() {
720 StreamSubscription subscription = _subscription;
721 if (_state == _STATE_MOVING) {
722 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
723 DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
724 """line 1033, column 31 of dart:async/stream_impl.dart: """,
725 _futureOrPrefetch is _Future<bool>, false);
726 _clear();
727 hasNext._complete(false);
728 } else {
729 _clear();
730 }
731 return subscription.cancel();
732 }
733 void _onData(T data) {
734 if (_state == _STATE_MOVING) {
735 _current = data;
736 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
737 DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
738 """line 1045, column 31 of dart:async/stream_impl.dart: """,
739 _futureOrPrefetch is _Future<bool>, false);
740 _futureOrPrefetch = null;
741 _state = _STATE_FOUND;
742 hasNext._complete(true);
743 return;
744 }
745 _subscription.pause();
746 assert(_futureOrPrefetch == null);
747 _futureOrPrefetch = data;
748 _state = _STATE_EXTRA_DATA;
749 }
750 void _onError(Object error, [StackTrace stackTrace]) {
751 if (_state == _STATE_MOVING) {
752 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
753 DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
754 """line 1059, column 31 of dart:async/stream_impl.dart: """,
755 _futureOrPrefetch is _Future<bool>, false);
756 _clear();
757 hasNext._completeError(error, stackTrace);
758 return;
759 }
760 _subscription.pause();
761 assert(_futureOrPrefetch == null);
762 _futureOrPrefetch = new AsyncError(error, stackTrace);
763 _state = _STATE_EXTRA_ERROR;
764 }
765 void _onDone() {
766 if (_state == _STATE_MOVING) {
767 _Future<bool> hasNext = DDC$RT.cast(_futureOrPrefetch, dynamic,
768 DDC$RT.type((_Future<bool> _) {}), "CastGeneral",
769 """line 1073, column 31 of dart:async/stream_impl.dart: """,
770 _futureOrPrefetch is _Future<bool>, false);
771 _clear();
772 hasNext._complete(false);
773 return;
774 }
775 _subscription.pause();
776 _futureOrPrefetch = null;
777 _state = _STATE_EXTRA_DONE;
778 }
779 }
780 typedef dynamic __t100(dynamic __u101);
781 typedef void __t102<T>(T __u103);
782 typedef dynamic __t105(dynamic __u106, dynamic __u107);
783 typedef void __t109(StreamSubscription<dynamic> __u110);
OLDNEW
« test/dart_codegen/expect/_internal/iterable.dart ('K') | « test/dart_codegen/expect/async/stream_controller.dart ('k') | test/dart_codegen/expect/async/stream_pipe.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698