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

Side by Side Diff: sdk/lib/async/stream_impl.dart

Issue 23926011: Rewrite Futures. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Remove chained future cycle test. Created 7 years, 3 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 | Annotate | Revision Log
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 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. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 part of dart.async; 5 part of dart.async;
6 6
7 /** Abstract and private interface for a place to put events. */ 7 /** Abstract and private interface for a place to put events. */
8 abstract class _EventSink<T> { 8 abstract class _EventSink<T> {
9 void _add(T data); 9 void _add(T data);
10 void _addError(Object error); 10 void _addError(Object error);
(...skipping 172 matching lines...) Expand 10 before | Expand all | Expand 10 after
183 if (!_inCallback) { 183 if (!_inCallback) {
184 // otherwise checkState will be called after firing or callback completes. 184 // otherwise checkState will be called after firing or callback completes.
185 _state |= _STATE_IN_CALLBACK; 185 _state |= _STATE_IN_CALLBACK;
186 _onCancel(); 186 _onCancel();
187 _pending = null; 187 _pending = null;
188 _state &= ~_STATE_IN_CALLBACK; 188 _state &= ~_STATE_IN_CALLBACK;
189 } 189 }
190 } 190 }
191 191
192 Future asFuture([var futureValue]) { 192 Future asFuture([var futureValue]) {
193 _FutureImpl<T> result = new _FutureImpl<T>(); 193 _Future<T> result = new _Future<T>();
194 194
195 // Overwrite the onDone and onError handlers. 195 // Overwrite the onDone and onError handlers.
196 _onDone = () { result._setValue(futureValue); }; 196 _onDone = () { result._complete(futureValue); };
197 _onError = (error) { 197 _onError = (error) {
198 cancel(); 198 cancel();
199 result._setError(error); 199 result._completeError(error);
200 }; 200 };
201 201
202 return result; 202 return result;
203 } 203 }
204 204
205 // State management. 205 // State management.
206 206
207 bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0; 207 bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
208 bool get _isClosed => (_state & _STATE_CLOSED) != 0; 208 bool get _isClosed => (_state & _STATE_CLOSED) != 0;
209 bool get _isCanceled => (_state & _STATE_CANCELED) != 0; 209 bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
(...skipping 499 matching lines...) Expand 10 before | Expand all | Expand 10 after
709 void pause([Future resumeSignal]) { 709 void pause([Future resumeSignal]) {
710 _pauseCounter++; 710 _pauseCounter++;
711 if (resumeSignal != null) resumeSignal.then((_) { resume(); }); 711 if (resumeSignal != null) resumeSignal.then((_) { resume(); });
712 } 712 }
713 void resume() { 713 void resume() {
714 if (_pauseCounter > 0) _pauseCounter--; 714 if (_pauseCounter > 0) _pauseCounter--;
715 } 715 }
716 void cancel() {} 716 void cancel() {}
717 bool get isPaused => _pauseCounter > 0; 717 bool get isPaused => _pauseCounter > 0;
718 718
719 Future asFuture([futureValue]) => new _FutureImpl(); 719 Future asFuture([futureValue]) => new _Future();
720 } 720 }
721 721
722 class _AsBroadcastStream<T> extends Stream<T> { 722 class _AsBroadcastStream<T> extends Stream<T> {
723 final Stream<T> _source; 723 final Stream<T> _source;
724 final _broadcastCallback _onListenHandler; 724 final _broadcastCallback _onListenHandler;
725 final _broadcastCallback _onCancelHandler; 725 final _broadcastCallback _onCancelHandler;
726 final _Zone _zone; 726 final _Zone _zone;
727 727
728 _AsBroadcastStreamController<T> _controller; 728 _AsBroadcastStreamController<T> _controller;
729 StreamSubscription<T> _subscription; 729 StreamSubscription<T> _subscription;
(...skipping 180 matching lines...) Expand 10 before | Expand all | Expand 10 after
910 _subscription = stream.listen(_onData, 910 _subscription = stream.listen(_onData,
911 onError: _onError, 911 onError: _onError,
912 onDone: _onDone, 912 onDone: _onDone,
913 cancelOnError: true); 913 cancelOnError: true);
914 } 914 }
915 915
916 T get current => _current; 916 T get current => _current;
917 917
918 Future<bool> moveNext() { 918 Future<bool> moveNext() {
919 if (_state == _STATE_DONE) { 919 if (_state == _STATE_DONE) {
920 return new _FutureImpl<bool>.immediate(false); 920 return new _Future<bool>.immediate(false);
921 } 921 }
922 if (_state == _STATE_MOVING) { 922 if (_state == _STATE_MOVING) {
923 throw new StateError("Already waiting for next."); 923 throw new StateError("Already waiting for next.");
924 } 924 }
925 if (_state == _STATE_FOUND) { 925 if (_state == _STATE_FOUND) {
926 _state = _STATE_MOVING; 926 _state = _STATE_MOVING;
927 _futureOrPrefetch = new _FutureImpl<bool>(); 927 _futureOrPrefetch = new _Future<bool>();
928 return _futureOrPrefetch; 928 return _futureOrPrefetch;
929 } else { 929 } else {
930 assert(_state >= _STATE_EXTRA_DATA); 930 assert(_state >= _STATE_EXTRA_DATA);
931 switch (_state) { 931 switch (_state) {
932 case _STATE_EXTRA_DATA: 932 case _STATE_EXTRA_DATA:
933 _state = _STATE_FOUND; 933 _state = _STATE_FOUND;
934 _current = _futureOrPrefetch; 934 _current = _futureOrPrefetch;
935 _futureOrPrefetch = null; 935 _futureOrPrefetch = null;
936 _subscription.resume(); 936 _subscription.resume();
937 return new _FutureImpl<bool>.immediate(true); 937 return new _Future<bool>.immediate(true);
938 case _STATE_EXTRA_ERROR: 938 case _STATE_EXTRA_ERROR:
939 Object prefetch = _futureOrPrefetch; 939 Object prefetch = _futureOrPrefetch;
940 _clear(); 940 _clear();
941 return new _FutureImpl<bool>.immediateError(prefetch); 941 return new _Future<bool>.immediateError(prefetch);
942 case _STATE_EXTRA_DONE: 942 case _STATE_EXTRA_DONE:
943 _clear(); 943 _clear();
944 return new _FutureImpl<bool>.immediate(false); 944 return new _Future<bool>.immediate(false);
945 } 945 }
946 } 946 }
947 } 947 }
948 948
949 /** Clears up the internal state when the iterator ends. */ 949 /** Clears up the internal state when the iterator ends. */
950 void _clear() { 950 void _clear() {
951 _subscription = null; 951 _subscription = null;
952 _futureOrPrefetch = null; 952 _futureOrPrefetch = null;
953 _current = null; 953 _current = null;
954 _state = _STATE_DONE; 954 _state = _STATE_DONE;
955 } 955 }
956 956
957 void cancel() { 957 void cancel() {
958 StreamSubscription subscription = _subscription; 958 StreamSubscription subscription = _subscription;
959 if (_state == _STATE_MOVING) { 959 if (_state == _STATE_MOVING) {
960 _FutureImpl<bool> hasNext = _futureOrPrefetch; 960 _Future<bool> hasNext = _futureOrPrefetch;
961 _clear(); 961 _clear();
962 hasNext._setValue(false); 962 hasNext._complete(false);
963 } else { 963 } else {
964 _clear(); 964 _clear();
965 } 965 }
966 subscription.cancel(); 966 subscription.cancel();
967 } 967 }
968 968
969 void _onData(T data) { 969 void _onData(T data) {
970 if (_state == _STATE_MOVING) { 970 if (_state == _STATE_MOVING) {
971 _current = data; 971 _current = data;
972 _FutureImpl<bool> hasNext = _futureOrPrefetch; 972 _Future<bool> hasNext = _futureOrPrefetch;
973 _futureOrPrefetch = null; 973 _futureOrPrefetch = null;
974 _state = _STATE_FOUND; 974 _state = _STATE_FOUND;
975 hasNext._setValue(true); 975 hasNext._complete(true);
976 return; 976 return;
977 } 977 }
978 _subscription.pause(); 978 _subscription.pause();
979 assert(_futureOrPrefetch == null); 979 assert(_futureOrPrefetch == null);
980 _futureOrPrefetch = data; 980 _futureOrPrefetch = data;
981 _state = _STATE_EXTRA_DATA; 981 _state = _STATE_EXTRA_DATA;
982 } 982 }
983 983
984 void _onError(Object error) { 984 void _onError(Object error) {
985 if (_state == _STATE_MOVING) { 985 if (_state == _STATE_MOVING) {
986 _FutureImpl<bool> hasNext = _futureOrPrefetch; 986 _Future<bool> hasNext = _futureOrPrefetch;
987 // We have cancelOnError: true, so the subscription is canceled. 987 // We have cancelOnError: true, so the subscription is canceled.
988 _clear(); 988 _clear();
989 hasNext._setError(error); 989 hasNext._completeError(error);
990 return; 990 return;
991 } 991 }
992 _subscription.pause(); 992 _subscription.pause();
993 assert(_futureOrPrefetch == null); 993 assert(_futureOrPrefetch == null);
994 _futureOrPrefetch = error; 994 _futureOrPrefetch = error;
995 _state = _STATE_EXTRA_ERROR; 995 _state = _STATE_EXTRA_ERROR;
996 } 996 }
997 997
998 void _onDone() { 998 void _onDone() {
999 if (_state == _STATE_MOVING) { 999 if (_state == _STATE_MOVING) {
1000 _FutureImpl<bool> hasNext = _futureOrPrefetch; 1000 _Future<bool> hasNext = _futureOrPrefetch;
1001 _clear(); 1001 _clear();
1002 hasNext._setValue(false); 1002 hasNext._complete(false);
1003 return; 1003 return;
1004 } 1004 }
1005 _subscription.pause(); 1005 _subscription.pause();
1006 _futureOrPrefetch = null; 1006 _futureOrPrefetch = null;
1007 _state = _STATE_EXTRA_DONE; 1007 _state = _STATE_EXTRA_DONE;
1008 } 1008 }
1009 } 1009 }
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698