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

Side by Side Diff: test/dart_codegen/expect/async/broadcast_stream_controller.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
OLDNEW
1 part of dart.async; 1 part of dart.async;
2 2 class _BroadcastStream<T> extends _ControllerStream<T> {_BroadcastStream(_Strea mControllerLifecycle controller) : super(DDC$RT.cast(controller, DDC$RT.type((_S treamControllerLifecycle<dynamic> _) {
3 class _BroadcastStream<T> extends _ControllerStream<T> {
4 _BroadcastStream(_StreamControllerLifecycle controller) : super(DDC$RT.cast(
5 controller, DDC$RT.type((_StreamControllerLifecycle<dynamic> _) {}),
6 DDC$RT.type((_StreamControllerLifecycle<T> _) {}), "CastDynamic",
7 """line 8, column 67 of dart:async/broadcast_stream_controller.dart: " "",
8 controller is _StreamControllerLifecycle<T>, false));
9 bool get isBroadcast => true;
10 }
11 abstract class _BroadcastSubscriptionLink {
12 _BroadcastSubscriptionLink _next;
13 _BroadcastSubscriptionLink _previous;
14 }
15 class _BroadcastSubscription<T> extends _ControllerSubscription<T>
16 implements _BroadcastSubscriptionLink {
17 static const int _STATE_EVENT_ID = 1;
18 static const int _STATE_FIRING = 2;
19 static const int _STATE_REMOVE_AFTER_FIRING = 4;
20 int _eventState;
21 _BroadcastSubscriptionLink _next;
22 _BroadcastSubscriptionLink _previous;
23 _BroadcastSubscription(_StreamControllerLifecycle controller,
24 void onData(T data), Function onError, void onDone(), bool cancelOnError)
25 : super(DDC$RT.cast(controller,
26 DDC$RT.type((_StreamControllerLifecycle<dynamic> _) {}),
27 DDC$RT.type((_StreamControllerLifecycle<T> _) {}), "CastDynamic",
28 """line 36, column 15 of dart:async/broadcast_stream_controller.dart: """,
29 controller is _StreamControllerLifecycle<T>,
30 false), onData, onError, onDone, cancelOnError) {
31 _next = _previous = this;
32 } 3 }
33 _BroadcastStreamController get _controller => ((__x2) => DDC$RT.cast(__x2, 4 ), DDC$RT.type((_StreamControllerLifecycle<T> _) {
34 DDC$RT.type((_StreamControllerLifecycle<T> _) {}),
35 DDC$RT.type((_BroadcastStreamController<dynamic> _) {}), "CastGeneral",
36 """line 40, column 49 of dart:async/broadcast_stream_controller.dart: """,
37 __x2 is _BroadcastStreamController<dynamic>, true))(super._controller);
38 bool _expectsEvent(int eventId) => (_eventState & _STATE_EVENT_ID) == eventId;
39 void _toggleEventId() {
40 _eventState ^= _STATE_EVENT_ID;
41 } 5 }
42 bool get _isFiring => (_eventState & _STATE_FIRING) != 0; 6 ), "CastDynamic", """line 8, column 67 of dart:async/broadcast_stream_controller .dart: """, controller is _StreamControllerLifecycle<T>, false));
43 void _setRemoveAfterFiring() { 7 bool get isBroadcast => true;
44 assert(_isFiring); 8 }
45 _eventState |= _STATE_REMOVE_AFTER_FIRING; 9 abstract class _BroadcastSubscriptionLink {_BroadcastSubscriptionLink _next;
10 _BroadcastSubscriptionLink _previous;
11 }
12 class _BroadcastSubscription<T> extends _ControllerSubscription<T> implements _ BroadcastSubscriptionLink {static const int _STATE_EVENT_ID = 1;
13 static const int _STATE_FIRING = 2;
14 static const int _STATE_REMOVE_AFTER_FIRING = 4;
15 int _eventState;
16 _BroadcastSubscriptionLink _next;
17 _BroadcastSubscriptionLink _previous;
18 _BroadcastSubscription(_StreamControllerLifecycle controller, void onData(T dat a), Function onError, void onDone(), bool cancelOnError) : super(DDC$RT.cast(con troller, DDC$RT.type((_StreamControllerLifecycle<dynamic> _) {
19 }
20 ), DDC$RT.type((_StreamControllerLifecycle<T> _) {
21 }
22 ), "CastDynamic", """line 36, column 15 of dart:async/broadcast_stream_controlle r.dart: """, controller is _StreamControllerLifecycle<T>, false), onData, onErro r, onDone, cancelOnError) {
23 _next = _previous = this;
24 }
25 _BroadcastStreamController get _controller => ((__x2) => DDC$RT.cast(__x2, DDC$ RT.type((_StreamControllerLifecycle<T> _) {
26 }
27 ), DDC$RT.type((_BroadcastStreamController<dynamic> _) {
28 }
29 ), "CastGeneral", """line 40, column 49 of dart:async/broadcast_stream_controlle r.dart: """, __x2 is _BroadcastStreamController<dynamic>, true))(super._controll er);
30 bool _expectsEvent(int eventId) => (_eventState & _STATE_EVENT_ID) == eventId;
31 void _toggleEventId() {
32 _eventState ^= _STATE_EVENT_ID;
33 }
34 bool get _isFiring => (_eventState & _STATE_FIRING) != 0;
35 void _setRemoveAfterFiring() {
36 assert (_isFiring); _eventState |= _STATE_REMOVE_AFTER_FIRING;
37 }
38 bool get _removeAfterFiring => (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0;
39 void _onPause() {
40 }
41 void _onResume() {
42 }
43 }
44 abstract class _BroadcastStreamController<T> implements StreamController<T>, _S treamControllerLifecycle<T>, _BroadcastSubscriptionLink, _EventSink<T>, _EventDi spatch<T> {static const int _STATE_INITIAL = 0;
45 static const int _STATE_EVENT_ID = 1;
46 static const int _STATE_FIRING = 2;
47 static const int _STATE_CLOSED = 4;
48 static const int _STATE_ADDSTREAM = 8;
49 final _NotificationHandler _onListen;
50 final _NotificationHandler _onCancel;
51 int _state;
52 _BroadcastSubscriptionLink _next;
53 _BroadcastSubscriptionLink _previous;
54 _AddStreamState<T> _addStreamState;
55 _Future _doneFuture;
56 _BroadcastStreamController(this._onListen, this._onCancel) : _state = _STATE_IN ITIAL {
57 _next = _previous = this;
58 }
59 Stream<T> get stream => new _BroadcastStream<T>(this);
60 StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
61 bool get isClosed => (_state & _STATE_CLOSED) != 0;
62 bool get isPaused => false;
63 bool get hasListener => !_isEmpty;
64 bool get _hasOneListener {
65 assert (!_isEmpty); return identical(_next._next, this);
66 }
67 bool get _isFiring => (_state & _STATE_FIRING) != 0;
68 bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0;
69 bool get _mayAddEvent => (_state < _STATE_CLOSED);
70 _Future _ensureDoneFuture() {
71 if (_doneFuture != null) return _doneFuture;
72 return _doneFuture = new _Future();
73 }
74 bool get _isEmpty => identical(_next, this);
75 void _addListener(_BroadcastSubscription<T> subscription) {
76 assert (identical(subscription._next, subscription)); subscription._previous = _ previous;
77 subscription._next = this;
78 this._previous._next = subscription;
79 this._previous = subscription;
80 subscription._eventState = (_state & _STATE_EVENT_ID);
81 }
82 void _removeListener(_BroadcastSubscription<T> subscription) {
83 assert (identical(subscription._controller, this)); assert (!identical(subscript ion._next, subscription)); _BroadcastSubscriptionLink previous = subscription._p revious;
84 _BroadcastSubscriptionLink next = subscription._next;
85 previous._next = next;
86 next._previous = previous;
87 subscription._next = subscription._previous = subscription;
88 }
89 StreamSubscription<T> _subscribe(void onData(T data), Function onError, void on Done(), bool cancelOnError) {
90 if (isClosed) {
91 if (onDone == null) onDone = _nullDoneHandler;
92 return new _DoneStreamSubscription<T>(onDone);
93 }
94 StreamSubscription subscription = new _BroadcastSubscription<T>(this, onData, o nError, onDone, cancelOnError);
95 _addListener(DDC$RT.cast(subscription, DDC$RT.type((StreamSubscription<dynamic> _) {
96 }
97 ), DDC$RT.type((_BroadcastSubscription<T> _) {
98 }
99 ), "CastGeneral", """line 196, column 18 of dart:async/broadcast_stream_controll er.dart: """, subscription is _BroadcastSubscription<T>, false));
100 if (identical(_next, _previous)) {
101 _runGuarded(_onListen);
102 }
103 return DDC$RT.cast(subscription, DDC$RT.type((StreamSubscription<dynamic> _) {
104 }
105 ), DDC$RT.type((StreamSubscription<T> _) {
106 }
107 ), "CastDynamic", """line 201, column 12 of dart:async/broadcast_stream_controll er.dart: """, subscription is StreamSubscription<T>, false);
108 }
109 Future _recordCancel(_BroadcastSubscription<T> subscription) {
110 if (identical(subscription._next, subscription)) return null;
111 assert (!identical(subscription._next, subscription)); if (subscription._isFiri ng) {
112 subscription._setRemoveAfterFiring();
113 }
114 else {
115 assert (!identical(subscription._next, subscription)); _removeListener(subscript ion);
116 if (!_isFiring && _isEmpty) {
117 _callOnCancel();
118 }
119 }
120 return null;
121 }
122 void _recordPause(StreamSubscription<T> subscription) {
123 }
124 void _recordResume(StreamSubscription<T> subscription) {
125 }
126 Error _addEventError() {
127 if (isClosed) {
128 return new StateError("Cannot add new events after calling close");
129 }
130 assert (_isAddingStream); return new StateError("Cannot add new events while do ing an addStream");
131 }
132 void add(T data) {
133 if (!_mayAddEvent) throw _addEventError();
134 _sendData(data);
135 }
136 void addError(Object error, [StackTrace stackTrace]) {
137 error = _nonNullError(error);
138 if (!_mayAddEvent) throw _addEventError();
139 AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
140 if (replacement != null) {
141 error = _nonNullError(replacement.error);
142 stackTrace = replacement.stackTrace;
143 }
144 _sendError(error, stackTrace);
145 }
146 Future close() {
147 if (isClosed) {
148 assert (_doneFuture != null); return _doneFuture;
149 }
150 if (!_mayAddEvent) throw _addEventError();
151 _state |= _STATE_CLOSED;
152 Future doneFuture = _ensureDoneFuture();
153 _sendDone();
154 return doneFuture;
155 }
156 Future get done => _ensureDoneFuture();
157 Future addStream(Stream<T> stream, {
158 bool cancelOnError : true}
159 ) {
160 if (!_mayAddEvent) throw _addEventError();
161 _state |= _STATE_ADDSTREAM;
162 _addStreamState = ((__x3) => DDC$RT.cast(__x3, DDC$RT.type((_AddStreamState<dyn amic> _) {
163 }
164 ), DDC$RT.type((_AddStreamState<T> _) {
165 }
166 ), "CastExact", """line 268, column 23 of dart:async/broadcast_stream_controller .dart: """, __x3 is _AddStreamState<T>, false))(new _AddStreamState(this, stream , cancelOnError));
167 return _addStreamState.addStreamFuture;
168 }
169 void _add(T data) {
170 _sendData(data);
171 }
172 void _addError(Object error, StackTrace stackTrace) {
173 _sendError(error, stackTrace);
174 }
175 void _close() {
176 assert (_isAddingStream); _AddStreamState addState = _addStreamState;
177 _addStreamState = null;
178 _state &= ~_STATE_ADDSTREAM;
179 addState.complete();
180 }
181 void _forEachListener(void action(_BufferingStreamSubscription<T> subscription) ) {
182 if (_isFiring) {
183 throw new StateError("Cannot fire new event. Controller is already firing an eve nt");
184 }
185 if (_isEmpty) return; int id = (_state & _STATE_EVENT_ID);
186 _state ^= _STATE_EVENT_ID | _STATE_FIRING;
187 _BroadcastSubscriptionLink link = _next;
188 while (!identical(link, this)) {
189 _BroadcastSubscription<T> subscription = DDC$RT.cast(link, _BroadcastSubscriptio nLink, DDC$RT.type((_BroadcastSubscription<T> _) {
190 }
191 ), "CastGeneral", """line 309, column 48 of dart:async/broadcast_stream_controll er.dart: """, link is _BroadcastSubscription<T>, false);
192 if (subscription._expectsEvent(id)) {
193 subscription._eventState |= _BroadcastSubscription._STATE_FIRING;
194 action(subscription);
195 subscription._toggleEventId();
196 link = subscription._next;
197 if (subscription._removeAfterFiring) {
198 _removeListener(subscription);
46 } 199 }
47 bool get _removeAfterFiring => 200 subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING;
48 (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0; 201 }
49 void _onPause() {} 202 else {
50 void _onResume() {} 203 link = subscription._next;
51 } 204 }
52 abstract class _BroadcastStreamController<T> 205 }
53 implements StreamController<T>, _StreamControllerLifecycle<T>, _BroadcastSub scriptionLink, _EventSink<T>, _EventDispatch<T> { 206 _state &= ~_STATE_FIRING;
54 static const int _STATE_INITIAL = 0; 207 if (_isEmpty) {
55 static const int _STATE_EVENT_ID = 1; 208 _callOnCancel();
56 static const int _STATE_FIRING = 2; 209 }
57 static const int _STATE_CLOSED = 4; 210 }
58 static const int _STATE_ADDSTREAM = 8; 211 void _callOnCancel() {
59 final _NotificationHandler _onListen; 212 assert (_isEmpty); if (isClosed && _doneFuture._mayComplete) {
60 final _NotificationHandler _onCancel; 213 _doneFuture._asyncComplete(null);
61 int _state; 214 }
62 _BroadcastSubscriptionLink _next; 215 _runGuarded(_onCancel);
63 _BroadcastSubscriptionLink _previous; 216 }
64 _AddStreamState<T> _addStreamState; 217 }
65 _Future _doneFuture; 218 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> { _SyncBroadcastStreamController(void onListen(), void onCancel()) : super(onListe n, onCancel);
66 _BroadcastStreamController(this._onListen, this._onCancel) 219 void _sendData(T data) {
67 : _state = _STATE_INITIAL { 220 if (_isEmpty) return; if (_hasOneListener) {
68 _next = _previous = this; 221 _state |= _BroadcastStreamController._STATE_FIRING;
69 } 222 _BroadcastSubscription subscription = DDC$RT.cast(_next, _BroadcastSubscription Link, DDC$RT.type((_BroadcastSubscription<dynamic> _) {
70 Stream<T> get stream => new _BroadcastStream<T>(this); 223 }
71 StreamSink<T> get sink => new _StreamSinkWrapper<T>(this); 224 ), "CastGeneral", """line 350, column 45 of dart:async/broadcast_stream_controll er.dart: """, _next is _BroadcastSubscription<dynamic>, true);
72 bool get isClosed => (_state & _STATE_CLOSED) != 0; 225 subscription._add(data);
73 bool get isPaused => false; 226 _state &= ~_BroadcastStreamController._STATE_FIRING;
74 bool get hasListener => !_isEmpty; 227 if (_isEmpty) {
75 bool get _hasOneListener { 228 _callOnCancel();
76 assert(!_isEmpty); 229 }
77 return identical(_next._next, this); 230 return;}
78 } 231 _forEachListener((_BufferingStreamSubscription<T> subscription) {
79 bool get _isFiring => (_state & _STATE_FIRING) != 0; 232 subscription._add(data);
80 bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; 233 }
81 bool get _mayAddEvent => (_state < _STATE_CLOSED); 234 );
82 _Future _ensureDoneFuture() { 235 }
83 if (_doneFuture != null) return _doneFuture; 236 void _sendError(Object error, StackTrace stackTrace) {
84 return _doneFuture = new _Future(); 237 if (_isEmpty) return; _forEachListener((_BufferingStreamSubscription<T> subscrip tion) {
85 } 238 subscription._addError(error, stackTrace);
86 bool get _isEmpty => identical(_next, this); 239 }
87 void _addListener(_BroadcastSubscription<T> subscription) { 240 );
88 assert(identical(subscription._next, subscription)); 241 }
89 subscription._previous = _previous; 242 void _sendDone() {
90 subscription._next = this; 243 if (!_isEmpty) {
91 this._previous._next = subscription; 244 _forEachListener(((__x9) => DDC$RT.wrap((dynamic f(_BroadcastSubscription<T> __u 4)) {
92 this._previous = subscription; 245 dynamic c(_BroadcastSubscription<T> x0) => f(DDC$RT.cast(x0, DDC$RT.type((_Buffe ringStreamSubscription<T> _) {
93 subscription._eventState = (_state & _STATE_EVENT_ID); 246 }
94 } 247 ), DDC$RT.type((_BroadcastSubscription<T> _) {
95 void _removeListener(_BroadcastSubscription<T> subscription) { 248 }
96 assert(identical(subscription._controller, this)); 249 ), "CastParam", """line 372, column 24 of dart:async/broadcast_stream_controller .dart: """, x0 is _BroadcastSubscription<T>, false));
97 assert(!identical(subscription._next, subscription)); 250 return f == null ? null : c;
98 _BroadcastSubscriptionLink previous = subscription._previous; 251 }
99 _BroadcastSubscriptionLink next = subscription._next; 252 , __x9, DDC$RT.type((__t7<T> _) {
100 previous._next = next; 253 }
101 next._previous = previous; 254 ), DDC$RT.type((__t5<T> _) {
102 subscription._next = subscription._previous = subscription; 255 }
103 } 256 ), "WrapLiteral", """line 372, column 24 of dart:async/broadcast_stream_controll er.dart: """, __x9 is __t5<T>))((_BroadcastSubscription<T> subscription) {
104 StreamSubscription<T> _subscribe(void onData(T data), Function onError, 257 subscription._close();
105 void onDone(), bool cancelOnError) { 258 }
106 if (isClosed) { 259 ));
107 if (onDone == null) onDone = _nullDoneHandler; 260 }
108 return new _DoneStreamSubscription<T>(onDone); 261 else {
109 } 262 assert (_doneFuture != null); assert (_doneFuture._mayComplete); _doneFuture._as yncComplete(null);
110 StreamSubscription subscription = new _BroadcastSubscription<T>( 263 }
111 this, onData, onError, onDone, cancelOnError); 264 }
112 _addListener(DDC$RT.cast(subscription, 265 }
113 DDC$RT.type((StreamSubscription<dynamic> _) {}), 266 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {_AsyncBroadcastStreamController(void onListen(), void onCancel()) : super(onLis ten, onCancel);
114 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastGeneral", 267 void _sendData(T data) {
115 """line 196, column 18 of dart:async/broadcast_stream_controller.dart: " "", 268 for (_BroadcastSubscriptionLink link = _next;
116 subscription is _BroadcastSubscription<T>, false)); 269 !identical(link, this);
117 if (identical(_next, _previous)) { 270 link = link._next) {
118 _runGuarded(_onListen); 271 _BroadcastSubscription<T> subscription = DDC$RT.cast(link, _BroadcastSubscriptio nLink, DDC$RT.type((_BroadcastSubscription<T> _) {
119 } 272 }
120 return DDC$RT.cast(subscription, 273 ), "CastGeneral", """line 393, column 48 of dart:async/broadcast_stream_controll er.dart: """, link is _BroadcastSubscription<T>, false);
121 DDC$RT.type((StreamSubscription<dynamic> _) {}), 274 subscription._addPending(new _DelayedData(data));
122 DDC$RT.type((StreamSubscription<T> _) {}), "CastDynamic", 275 }
123 """line 201, column 12 of dart:async/broadcast_stream_controller.dart: " "", 276 }
124 subscription is StreamSubscription<T>, false); 277 void _sendError(Object error, StackTrace stackTrace) {
125 } 278 for (_BroadcastSubscriptionLink link = _next;
126 Future _recordCancel(_BroadcastSubscription<T> subscription) { 279 !identical(link, this);
127 if (identical(subscription._next, subscription)) return null; 280 link = link._next) {
128 assert(!identical(subscription._next, subscription)); 281 _BroadcastSubscription<T> subscription = DDC$RT.cast(link, _BroadcastSubscriptio nLink, DDC$RT.type((_BroadcastSubscription<T> _) {
129 if (subscription._isFiring) { 282 }
130 subscription._setRemoveAfterFiring(); 283 ), "CastGeneral", """line 402, column 48 of dart:async/broadcast_stream_controll er.dart: """, link is _BroadcastSubscription<T>, false);
131 } else { 284 subscription._addPending(new _DelayedError(error, stackTrace));
132 assert(!identical(subscription._next, subscription)); 285 }
133 _removeListener(subscription); 286 }
134 if (!_isFiring && _isEmpty) { 287 void _sendDone() {
135 _callOnCancel(); 288 if (!_isEmpty) {
136 } 289 for (_BroadcastSubscriptionLink link = _next;
137 } 290 !identical(link, this);
138 return null; 291 link = link._next) {
139 } 292 _BroadcastSubscription<T> subscription = DDC$RT.cast(link, _BroadcastSubscriptio nLink, DDC$RT.type((_BroadcastSubscription<T> _) {
140 void _recordPause(StreamSubscription<T> subscription) {} 293 }
141 void _recordResume(StreamSubscription<T> subscription) {} 294 ), "CastGeneral", """line 412, column 50 of dart:async/broadcast_stream_controll er.dart: """, link is _BroadcastSubscription<T>, false);
142 Error _addEventError() { 295 subscription._addPending(const _DelayedDone());
143 if (isClosed) { 296 }
144 return new StateError("Cannot add new events after calling close"); 297 }
145 } 298 else {
146 assert(_isAddingStream); 299 assert (_doneFuture != null); assert (_doneFuture._mayComplete); _doneFuture._as yncComplete(null);
147 return new StateError("Cannot add new events while doing an addStream"); 300 }
148 } 301 }
149 void add(T data) { 302 }
150 if (!_mayAddEvent) throw _addEventError(); 303 class _AsBroadcastStreamController<T> extends _SyncBroadcastStreamController<T> implements _EventDispatch<T> {_StreamImplEvents _pending;
151 _sendData(data); 304 _AsBroadcastStreamController(void onListen(), void onCancel()) : super(onListen , onCancel);
152 } 305 bool get _hasPending => _pending != null && !_pending.isEmpty;
153 void addError(Object error, [StackTrace stackTrace]) { 306 void _addPendingEvent(_DelayedEvent event) {
154 error = _nonNullError(error); 307 if (_pending == null) {
155 if (!_mayAddEvent) throw _addEventError(); 308 _pending = new _StreamImplEvents();
156 AsyncError replacement = Zone.current.errorCallback(error, stackTrace); 309 }
157 if (replacement != null) { 310 _pending.add(event);
158 error = _nonNullError(replacement.error); 311 }
159 stackTrace = replacement.stackTrace; 312 void add(T data) {
160 } 313 if (!isClosed && _isFiring) {
161 _sendError(error, stackTrace); 314 _addPendingEvent(new _DelayedData<T>(data));
162 } 315 return;}
163 Future close() { 316 super.add(data);
164 if (isClosed) { 317 while (_hasPending) {
165 assert(_doneFuture != null); 318 _pending.handleNext(this);
166 return _doneFuture; 319 }
167 } 320 }
168 if (!_mayAddEvent) throw _addEventError(); 321 void addError(Object error, [StackTrace stackTrace]) {
169 _state |= _STATE_CLOSED; 322 if (!isClosed && _isFiring) {
170 Future doneFuture = _ensureDoneFuture(); 323 _addPendingEvent(new _DelayedError(error, stackTrace));
171 _sendDone(); 324 return;}
172 return doneFuture; 325 if (!_mayAddEvent) throw _addEventError();
173 } 326 _sendError(error, stackTrace);
174 Future get done => _ensureDoneFuture(); 327 while (_hasPending) {
175 Future addStream(Stream<T> stream, {bool cancelOnError: true}) { 328 _pending.handleNext(this);
176 if (!_mayAddEvent) throw _addEventError(); 329 }
177 _state |= _STATE_ADDSTREAM; 330 }
178 _addStreamState = ((__x3) => DDC$RT.cast(__x3, 331 Future close() {
179 DDC$RT.type((_AddStreamState<dynamic> _) {}), 332 if (!isClosed && _isFiring) {
180 DDC$RT.type((_AddStreamState<T> _) {}), "CastExact", 333 _addPendingEvent(const _DelayedDone());
181 """line 268, column 23 of dart:async/broadcast_stream_controller.dart: " "", 334 _state |= _BroadcastStreamController._STATE_CLOSED;
182 __x3 is _AddStreamState<T>, 335 return super.done;
183 false))(new _AddStreamState(this, stream, cancelOnError)); 336 }
184 return _addStreamState.addStreamFuture; 337 Future result = super.close();
185 } 338 assert (!_hasPending); return result;
186 void _add(T data) { 339 }
187 _sendData(data); 340 void _callOnCancel() {
188 } 341 if (_hasPending) {
189 void _addError(Object error, StackTrace stackTrace) { 342 _pending.clear();
190 _sendError(error, stackTrace); 343 _pending = null;
191 } 344 }
192 void _close() { 345 super._callOnCancel();
193 assert(_isAddingStream); 346 }
194 _AddStreamState addState = _addStreamState; 347 }
195 _addStreamState = null; 348 class _DoneSubscription<T> implements StreamSubscription<T> {int _pauseCount = 0;
196 _state &= ~_STATE_ADDSTREAM; 349 void onData(void handleData(T data)) {
197 addState.complete(); 350 }
198 } 351 void onError(Function handleError) {
199 void _forEachListener( 352 }
200 void action(_BufferingStreamSubscription<T> subscription)) { 353 void onDone(void handleDone()) {
201 if (_isFiring) { 354 }
202 throw new StateError( 355 void pause([Future resumeSignal]) {
203 "Cannot fire new event. Controller is already firing an event"); 356 if (resumeSignal != null) resumeSignal.then(_resume);
204 } 357 _pauseCount++;
205 if (_isEmpty) return; 358 }
206 int id = (_state & _STATE_EVENT_ID); 359 void resume() {
207 _state ^= _STATE_EVENT_ID | _STATE_FIRING; 360 _resume(null);
208 _BroadcastSubscriptionLink link = _next; 361 }
209 while (!identical(link, this)) { 362 void _resume(_) {
210 _BroadcastSubscription<T> subscription = DDC$RT.cast(link, 363 if (_pauseCount > 0) _pauseCount--;
211 _BroadcastSubscriptionLink, 364 }
212 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastGeneral", 365 Future cancel() {
213 """line 309, column 48 of dart:async/broadcast_stream_controller.dart: """, 366 return new _Future.immediate(null);
214 link is _BroadcastSubscription<T>, false); 367 }
215 if (subscription._expectsEvent(id)) { 368 bool get isPaused => _pauseCount > 0;
216 subscription._eventState |= _BroadcastSubscription._STATE_FIRING; 369 Future asFuture([Object value]) => new _Future();
217 action(subscription); 370 }
218 subscription._toggleEventId(); 371 typedef void __t5<T>(_BufferingStreamSubscription<T> __u6);
219 link = subscription._next; 372 typedef dynamic __t7<T>(_BroadcastSubscription<T> __u8);
220 if (subscription._removeAfterFiring) {
221 _removeListener(subscription);
222 }
223 subscription._eventState &= ~_BroadcastSubscription._STATE_FIRING;
224 } else {
225 link = subscription._next;
226 }
227 }
228 _state &= ~_STATE_FIRING;
229 if (_isEmpty) {
230 _callOnCancel();
231 }
232 }
233 void _callOnCancel() {
234 assert(_isEmpty);
235 if (isClosed && _doneFuture._mayComplete) {
236 _doneFuture._asyncComplete(null);
237 }
238 _runGuarded(_onCancel);
239 }
240 }
241 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
242 _SyncBroadcastStreamController(void onListen(), void onCancel())
243 : super(onListen, onCancel);
244 void _sendData(T data) {
245 if (_isEmpty) return;
246 if (_hasOneListener) {
247 _state |= _BroadcastStreamController._STATE_FIRING;
248 _BroadcastSubscription subscription = DDC$RT.cast(_next,
249 _BroadcastSubscriptionLink,
250 DDC$RT.type((_BroadcastSubscription<dynamic> _) {}), "CastGeneral",
251 """line 350, column 45 of dart:async/broadcast_stream_controller.dart: """,
252 _next is _BroadcastSubscription<dynamic>, true);
253 subscription._add(data);
254 _state &= ~_BroadcastStreamController._STATE_FIRING;
255 if (_isEmpty) {
256 _callOnCancel();
257 }
258 return;
259 }
260 _forEachListener((_BufferingStreamSubscription<T> subscription) {
261 subscription._add(data);
262 });
263 }
264 void _sendError(Object error, StackTrace stackTrace) {
265 if (_isEmpty) return;
266 _forEachListener((_BufferingStreamSubscription<T> subscription) {
267 subscription._addError(error, stackTrace);
268 });
269 }
270 void _sendDone() {
271 if (!_isEmpty) {
272 _forEachListener(((__x9) => DDC$RT.wrap(
273 (dynamic f(_BroadcastSubscription<T> __u4)) {
274 dynamic c(_BroadcastSubscription<T> x0) => f(DDC$RT.cast(x0,
275 DDC$RT.type((_BufferingStreamSubscription<T> _) {}),
276 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastParam",
277 """line 372, column 24 of dart:async/broadcast_stream_controller.dar t: """,
278 x0 is _BroadcastSubscription<T>, false));
279 return f == null ? null : c;
280 }, __x9, DDC$RT.type((__t7<T> _) {}), DDC$RT.type((__t5<T> _) {}),
281 "WrapLiteral",
282 """line 372, column 24 of dart:async/broadcast_stream_controller.dart: """,
283 __x9 is __t5<T>))((_BroadcastSubscription<T> subscription) {
284 subscription._close();
285 }));
286 } else {
287 assert(_doneFuture != null);
288 assert(_doneFuture._mayComplete);
289 _doneFuture._asyncComplete(null);
290 }
291 }
292 }
293 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
294 _AsyncBroadcastStreamController(void onListen(), void onCancel())
295 : super(onListen, onCancel);
296 void _sendData(T data) {
297 for (_BroadcastSubscriptionLink link = _next;
298 !identical(link, this);
299 link = link._next) {
300 _BroadcastSubscription<T> subscription = DDC$RT.cast(link,
301 _BroadcastSubscriptionLink,
302 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastGeneral",
303 """line 393, column 48 of dart:async/broadcast_stream_controller.dart: """,
304 link is _BroadcastSubscription<T>, false);
305 subscription._addPending(new _DelayedData(data));
306 }
307 }
308 void _sendError(Object error, StackTrace stackTrace) {
309 for (_BroadcastSubscriptionLink link = _next;
310 !identical(link, this);
311 link = link._next) {
312 _BroadcastSubscription<T> subscription = DDC$RT.cast(link,
313 _BroadcastSubscriptionLink,
314 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastGeneral",
315 """line 402, column 48 of dart:async/broadcast_stream_controller.dart: """,
316 link is _BroadcastSubscription<T>, false);
317 subscription._addPending(new _DelayedError(error, stackTrace));
318 }
319 }
320 void _sendDone() {
321 if (!_isEmpty) {
322 for (_BroadcastSubscriptionLink link = _next;
323 !identical(link, this);
324 link = link._next) {
325 _BroadcastSubscription<T> subscription = DDC$RT.cast(link,
326 _BroadcastSubscriptionLink,
327 DDC$RT.type((_BroadcastSubscription<T> _) {}), "CastGeneral",
328 """line 412, column 50 of dart:async/broadcast_stream_controller.dar t: """,
329 link is _BroadcastSubscription<T>, false);
330 subscription._addPending(const _DelayedDone());
331 }
332 } else {
333 assert(_doneFuture != null);
334 assert(_doneFuture._mayComplete);
335 _doneFuture._asyncComplete(null);
336 }
337 }
338 }
339 class _AsBroadcastStreamController<T> extends _SyncBroadcastStreamController<T>
340 implements _EventDispatch<T> {
341 _StreamImplEvents _pending;
342 _AsBroadcastStreamController(void onListen(), void onCancel())
343 : super(onListen, onCancel);
344 bool get _hasPending => _pending != null && !_pending.isEmpty;
345 void _addPendingEvent(_DelayedEvent event) {
346 if (_pending == null) {
347 _pending = new _StreamImplEvents();
348 }
349 _pending.add(event);
350 }
351 void add(T data) {
352 if (!isClosed && _isFiring) {
353 _addPendingEvent(new _DelayedData<T>(data));
354 return;
355 }
356 super.add(data);
357 while (_hasPending) {
358 _pending.handleNext(this);
359 }
360 }
361 void addError(Object error, [StackTrace stackTrace]) {
362 if (!isClosed && _isFiring) {
363 _addPendingEvent(new _DelayedError(error, stackTrace));
364 return;
365 }
366 if (!_mayAddEvent) throw _addEventError();
367 _sendError(error, stackTrace);
368 while (_hasPending) {
369 _pending.handleNext(this);
370 }
371 }
372 Future close() {
373 if (!isClosed && _isFiring) {
374 _addPendingEvent(const _DelayedDone());
375 _state |= _BroadcastStreamController._STATE_CLOSED;
376 return super.done;
377 }
378 Future result = super.close();
379 assert(!_hasPending);
380 return result;
381 }
382 void _callOnCancel() {
383 if (_hasPending) {
384 _pending.clear();
385 _pending = null;
386 }
387 super._callOnCancel();
388 }
389 }
390 class _DoneSubscription<T> implements StreamSubscription<T> {
391 int _pauseCount = 0;
392 void onData(void handleData(T data)) {}
393 void onError(Function handleError) {}
394 void onDone(void handleDone()) {}
395 void pause([Future resumeSignal]) {
396 if (resumeSignal != null) resumeSignal.then(_resume);
397 _pauseCount++;
398 }
399 void resume() {
400 _resume(null);
401 }
402 void _resume(_) {
403 if (_pauseCount > 0) _pauseCount--;
404 }
405 Future cancel() {
406 return new _Future.immediate(null);
407 }
408 bool get isPaused => _pauseCount > 0;
409 Future asFuture([Object value]) => new _Future();
410 }
411 typedef void __t5<T>(_BufferingStreamSubscription<T> __u6);
412 typedef dynamic __t7<T>(_BroadcastSubscription<T> __u8);
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698