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

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

Issue 12545056: Make stream subscription more lax wrt. calling methods on it after it's completed. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 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 | Annotate | Revision Log
« no previous file with comments | « no previous file | sdk/lib/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 // 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 // States shared by single/multi stream implementations. 7 // States shared by single/multi stream implementations.
8 8
9 // Completion state of the stream. 9 // Completion state of the stream.
10 /// Initial and default state where the stream can receive and send events. 10 /// Initial and default state where the stream can receive and send events.
(...skipping 924 matching lines...) Expand 10 before | Expand all | Expand 10 after
935 void _sendError(AsyncError error) { 935 void _sendError(AsyncError error) {
936 _onError(error); 936 _onError(error);
937 if (_unsubscribeOnError) _source._cancel(this); 937 if (_unsubscribeOnError) _source._cancel(this);
938 } 938 }
939 939
940 void _sendDone() { 940 void _sendDone() {
941 _onDone(); 941 _onDone();
942 } 942 }
943 943
944 void cancel() { 944 void cancel() {
945 if (!_isSubscribed) return;
945 _source._cancel(this); 946 _source._cancel(this);
946 } 947 }
947 948
948 void pause([Future resumeSignal]) { 949 void pause([Future resumeSignal]) {
950 if (!_isSubscribed) return;
949 _source._pause(this, resumeSignal); 951 _source._pause(this, resumeSignal);
950 } 952 }
951 953
952 void resume() { 954 void resume() {
953 if (!isPaused) { 955 if (!_isSubscribed || !isPaused) return;
954 throw new StateError("Resuming unpaused subscription");
955 }
956 _source._resume(this, false); 956 _source._resume(this, false);
957 } 957 }
958 } 958 }
959 959
960 // Internal helpers. 960 // Internal helpers.
961 961
962 // Types of the different handlers on a stream. Types used to type fields. 962 // Types of the different handlers on a stream. Types used to type fields.
963 typedef void _DataHandler<T>(T value); 963 typedef void _DataHandler<T>(T value);
964 typedef void _ErrorHandler(AsyncError error); 964 typedef void _ErrorHandler(AsyncError error);
965 typedef void _DoneHandler(); 965 typedef void _DoneHandler();
(...skipping 297 matching lines...) Expand 10 before | Expand all | Expand 10 after
1263 1263
1264 void onData(void handleAction(T value)) {} 1264 void onData(void handleAction(T value)) {}
1265 1265
1266 void onError(void handleError(AsyncError error)) {} 1266 void onError(void handleError(AsyncError error)) {}
1267 1267
1268 void onDone(void handleDone()) { 1268 void onDone(void handleDone()) {
1269 _handler = handleDone; 1269 _handler = handleDone;
1270 } 1270 }
1271 1271
1272 void pause([Future signal]) { 1272 void pause([Future signal]) {
1273 if (_isComplete) { 1273 if (_isComplete) return;
1274 throw new StateError("Subscription has been canceled.");
1275 }
1276 if (_timer != null) { 1274 if (_timer != null) {
1277 _timer.cancel(); 1275 _timer.cancel();
1278 _timer = null; 1276 _timer = null;
1279 } 1277 }
1280 _pauseCount++; 1278 _pauseCount++;
1281 if (signal != null) signal.whenComplete(resume); 1279 if (signal != null) signal.whenComplete(resume);
1282 } 1280 }
1283 1281
1284 void resume() { 1282 void resume() {
1285 if (_isComplete) { 1283 if (_isComplete) return;
1286 throw new StateError("Subscription has been canceled.");
1287 }
1288 if (_pauseCount == 0) return; 1284 if (_pauseCount == 0) return;
1289 _pauseCount--; 1285 _pauseCount--;
1290 if (_pauseCount == 0) { 1286 if (_pauseCount == 0) {
1291 _delayDone(); 1287 _delayDone();
1292 } 1288 }
1293 } 1289 }
1294 1290
1295 bool get isPaused => _pauseCount > 0; 1291 bool get isPaused => _pauseCount > 0;
1296 1292
1297 void cancel() { 1293 void cancel() {
1298 if (_isComplete) { 1294 if (_isComplete) return;
1299 throw new StateError("Subscription has been canceled.");
1300 }
1301 if (_timer != null) { 1295 if (_timer != null) {
1302 _timer.cancel(); 1296 _timer.cancel();
1303 _timer = null; 1297 _timer = null;
1304 } 1298 }
1305 _pauseCount = 0; 1299 _pauseCount = 0;
1306 } 1300 }
1307 } 1301 }
1308 1302
1309 class _SingleStreamMultiplexer<T> extends _MultiStreamImpl<T> { 1303 class _SingleStreamMultiplexer<T> extends _MultiStreamImpl<T> {
1310 final Stream<T> _source; 1304 final Stream<T> _source;
(...skipping 24 matching lines...) Expand all
1335 onError: this._addError, 1329 onError: this._addError,
1336 onDone: this._close); 1330 onDone: this._close);
1337 } else { 1331 } else {
1338 // TODO(lrn): Check why this can happen. 1332 // TODO(lrn): Check why this can happen.
1339 if (_subscription == null) return; 1333 if (_subscription == null) return;
1340 _subscription.cancel(); 1334 _subscription.cancel();
1341 _subscription = null; 1335 _subscription = null;
1342 } 1336 }
1343 } 1337 }
1344 } 1338 }
OLDNEW
« no previous file with comments | « no previous file | sdk/lib/async/stream_pipe.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698