dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Remove Signal class and use Future/.whenComplete instead.

Browse Source 
Review URL: https://codereview.chromium.org//11794043

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@16792 260f80e4-7a28-3924-810f-c04153c831b5
This commit is contained in:
lrn@google.com
2013-01-08 12:59:00 +00:00
parent 925171e498
commit aa7cbb2b41
11 changed files with 52 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/lib/async/async.dart
-1
View File
@@ -8,7 +8,6 @@ part 'async_error.dart';
part 'future.dart';
part 'future_impl.dart';
part 'merge_stream.dart';
part 'signal.dart';
part 'stream.dart';
part 'stream_controller.dart';
part 'stream_impl.dart';
sdk/lib/async/async_sources.gypi
-1
View File
@@ -9,7 +9,6 @@
'future.dart',
'future_impl.dart',
'merge_stream.dart',
'signal.dart',
'stream.dart',
'stream_controller.dart',
'stream_impl.dart',
sdk/lib/async/future.dart
+14 -6
View File
@@ -141,16 +141,24 @@ abstract class Completer<T> {
factory Completer() => new _CompleterImpl<T>();
/** The future that will contain the value produced by this completer. */
/** The future that will contain the result provided to this completer. */
Future get future;
/** Supply a value for [future]. */
void complete(T value);
/**
* Completes [future] with the supplied values.
*
* All listeners on the future will be immediately informed about the value.
*/
void complete([T value]);
/**
* Indicate in [future] that an exception occured while trying to produce its
* value. The argument [exception] should not be [:null:]. A [stackTrace]
* object can be provided as well to give the user information about where
* Complete [future] with an error.
*
* Completing a future with an error indicates that an exception was thrown
* while trying to produce a value.
*
* The argument [exception] should not be [:null:]. A [stackTrace]
* object can be provided as well, to give the user information about where
* the error occurred. If omitted, it will be [:null:].
*/
void completeError(Object exception, [Object stackTrace]);
sdk/lib/async/future_impl.dart
+1 -1
View File
@@ -14,7 +14,7 @@ class _CompleterImpl<T> implements Completer<T> {
_CompleterImpl() : future = new _FutureImpl<T>();
void complete(T value) {
void complete([T value]) {
if (_isComplete) throw new StateError("Future already completed");
_isComplete = true;
_FutureImpl future = this.future;
sdk/lib/async/signal.dart
-98
View File
@@ -1,98 +0,0 @@
// Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
// part of dart.async;
/**
* A basic asynchronous notification.
*
* *DEPRECATED*
*
* This class is scheduled for removal. Please don't use.
*/
abstract class Signal {
factory Signal.delayed(int milliseconds) {
var completer = new SignalCompleter();
new Timer(milliseconds, (_) => completer.complete());
return completer.signal;
}
/**
* The [onComplete] handler is called when the signal completes.
*
* If the signal is already complete, the [onComplete] handler is called
* as soon as possible, but no sooner than the next time an event is fired.
*/
void then(void onComplete());
}
typedef _SignalCompleteHandler();
/**
* Simple [Signal] controller that creates a [Signal] and allows completing it.
*
* *DEPRECATED*
*
* This class is scheduled for removal. Please don't use.
*/
class SignalCompleter {
final Signal signal;
SignalCompleter() : signal = new _SignalImpl();
void complete() {
_SignalImpl mySignal = signal;
mySignal._complete();
}
}
/**
* Simple Signal implementation receiving its completion from a
* [SignalCompleter].
*/
class _SignalImpl implements Signal {
/** Single-linked list of "done" event handlers to notify. */
_SignalListener _listeners = null;
/** Whether the signal is already completed. */
bool _isComplete = false;
void then(void onComplete()) {
_listeners = new _SignalListener(_listeners, onComplete);
if (_isComplete) {
// Schedule the done events as soon as the event queue is ready.
new Timer(0, (Timer timer) { _sendDone(); });
}
}
/**
* Complete the signal.
*
* This immediately notifies all listeners on the signal.
*/
void _complete() {
assert(!_isComplete); // Only complete once.
_isComplete = true;
_sendDone();
}
/**
* Notify all listeners.
*/
void _sendDone() {
while (_listeners != null) {
_DoneHandler onDone = _listeners.listener;
_listeners = _listeners.next;
try {
onDone();
} catch (e, s) {
new AsyncError(e, s).throwDelayed();
}
}
}
}
/** Single-linked list element of the listeners on a [_SignalImpl]. */
class _SignalListener {
_SignalListener next;
_SignalCompleteHandler listener;
_SignalListener(this.next, this.listener);
}
sdk/lib/async/stream.dart
+6 -6
View File
@@ -153,19 +153,19 @@ abstract class Stream<T> {
}
// Deprecated method, previously called 'pipe', retained for compatibility.
Signal pipeInto(Sink<T> sink,
Future pipeInto(Sink<T> sink,
{void onError(AsyncError error),
bool unsubscribeOnError}) {
SignalCompleter completer = new SignalCompleter();
Completer completer = new Completer();
this.listen(
sink.add,
onError: onError,
onDone: () {
sink.close();
completer.complete();
completer.complete(null);
},
unsubscribeOnError: unsubscribeOnError);
return completer.signal;
return completer.future;
}
@@ -716,13 +716,13 @@ abstract class StreamSubscription<T> {
* Request that the stream pauses events until further notice.
*
* If [resumeSignal] is provided, the stream will undo the pause
* when the signal completes.
* when the future completes in any way.
* A call to [resume] will also undo a pause.
*
* If the subscription is paused more than once, an equal number
* of resumes must be performed to resume the stream.
*/
void pause([Signal resumeSignal]);
void pause([Future resumeSignal]);
/**
* Resume after a pause.
sdk/lib/async/stream_controller.dart
+1 -1
View File
@@ -80,7 +80,7 @@ class StreamController<T> extends Stream<T> implements StreamSink<T> {
/**
* Send or queue a data event.
*/
Signal add(T value) => _stream._add(value);
void add(T value) => _stream._add(value);
/**
* Send or enqueue an error event.
sdk/lib/async/stream_impl.dart
+4 -4
View File
@@ -216,7 +216,7 @@ abstract class _StreamImpl<T> extends Stream<T> {
* subscriptions, e.g., a filtering stream pausing its own source if all its
* subscribers are paused.
*/
void _pause(_StreamListener<T> listener, Signal resumeSignal) {
void _pause(_StreamListener<T> listener, Future resumeSignal) {
assert(identical(listener._source, this));
if (!listener._isSubscribed) {
throw new StateError("Subscription has been canceled.");
@@ -225,7 +225,7 @@ abstract class _StreamImpl<T> extends Stream<T> {
bool wasPaused = _isPaused;
_incrementPauseCount(listener);
if (resumeSignal != null) {
resumeSignal.then(() { this._resume(listener, true); });
resumeSignal.whenComplete(() { this._resume(listener, true); });
}
if (!wasPaused) {
_onPauseStateChange();
@@ -684,7 +684,7 @@ class _StreamSubscriptionImpl<T> extends _StreamListener<T>
_source._cancel(this);
}
void pause([Signal resumeSignal]) {
void pause([Future resumeSignal]) {
_source._pause(this, resumeSignal);
}
@@ -983,7 +983,7 @@ class _DoneSubscription<T> implements StreamSubscription<T> {
_handler = handleDone;
}
void pause([Signal signal]) {
void pause([Future signal]) {
if (_isComplete) {
throw new StateError("Subscription has been canceled.");
}
tests/lib/async/event_helper.dart
+7 -7
View File
@@ -112,7 +112,7 @@ class Events implements StreamSink {
* Should only be used when there is a subscription. That is, after a
* call to [subscribeTo].
*/
void pause([Signal resumeSignal]) {
void pause([Future resumeSignal]) {
throw new StateError("Not capturing events.");
}
@@ -134,12 +134,12 @@ class Events implements StreamSink {
class CaptureEvents extends Events {
StreamSubscription subscription;
SignalCompleter onDoneSignal;
Completer onDoneSignal;
bool unsubscribeOnError = false;
CaptureEvents(Stream stream,
{ bool unsubscribeOnError: false })
: onDoneSignal = new SignalCompleter() {
: onDoneSignal = new Completer() {
this.unsubscribeOnError = unsubscribeOnError;
subscription = stream.listen(add,
onError: signalError,
@@ -149,15 +149,15 @@ class CaptureEvents extends Events {
void signalError(AsyncError error) {
super.signalError(error);
if (unsubscribeOnError) onDoneSignal.complete();
if (unsubscribeOnError) onDoneSignal.complete(null);
}
void close() {
super.close();
if (onDoneSignal != null) onDoneSignal.complete();
if (onDoneSignal != null) onDoneSignal.complete(null);
}
void pause([Signal resumeSignal]) {
void pause([Future resumeSignal]) {
subscription.pause(resumeSignal);
}
@@ -168,6 +168,6 @@ class CaptureEvents extends Events {
bool get isPaused => subscription.isPaused;
void onDone(void action()) {
onDoneSignal.signal.then(action);
onDoneSignal.future.whenComplete(action);
}
}
tests/lib/async/stream_controller_async_test.dart
+18 -16
View File
@@ -15,7 +15,7 @@ testController() {
test("StreamController.reduce", () {
StreamController c = new StreamController();
c.reduce(0, (a,b) => a + b)
.then(expectAsync1((int v) {
.then(expectAsync1((int v) {
Expect.equals(42, v);
}));
c.add(10);
@@ -26,9 +26,7 @@ testController() {
test("StreamController.reduce throws", () {
StreamController c = new StreamController();
c.reduce(0, (a,b) { throw "Fnyf!"; })
.catchError(expectAsync1((e) {
Expect.equals("Fnyf!", e.error);
}));
.catchError(expectAsync1((e) { Expect.equals("Fnyf!", e.error); }));
c.add(42);
});
@@ -36,7 +34,9 @@ testController() {
StreamController c = new StreamController();
var list = <int>[];
c.pipeInto(new CollectionSink<int>(list))
.then(expectAsync0(() { Expect.listEquals(<int>[1,2,9,3,9], list); }));
.whenComplete(expectAsync0(() {
Expect.listEquals(<int>[1,2,9,3,9], list);
}));
c.add(1);
c.add(2);
c.add(9);
@@ -67,7 +67,9 @@ testSingleController() {
StreamController c = new StreamController.singleSubscription();
var list = <int>[];
c.pipeInto(new CollectionSink<int>(list))
.then(expectAsync0(() { Expect.listEquals(<int>[1,2,9,3,9], list); }));
.whenComplete(expectAsync0(() {
Expect.listEquals(<int>[1,2,9,3,9], list);
}));
c.add(1);
c.add(2);
c.add(9);
@@ -290,8 +292,8 @@ testPause() {
expectedEvents.add(42);
c.add(42);
Expect.listEquals(expectedEvents.events, actualEvents.events);
SignalCompleter completer = new SignalCompleter();
actualEvents.pause(completer.signal);
Completer completer = new Completer();
actualEvents.pause(completer.future);
c..add(43)..add(44)..close();
Expect.listEquals(expectedEvents.events, actualEvents.events);
completer.complete();
@@ -308,10 +310,10 @@ testPause() {
expectedEvents.add(42);
c.add(42);
Expect.listEquals(expectedEvents.events, actualEvents.events);
SignalCompleter completer = new SignalCompleter();
SignalCompleter completer2 = new SignalCompleter();
actualEvents.pause(completer.signal);
actualEvents.pause(completer2.signal);
Completer completer = new Completer();
Completer completer2 = new Completer();
actualEvents.pause(completer.future);
actualEvents.pause(completer2.future);
c..add(43)..add(44)..close();
Expect.listEquals(expectedEvents.events, actualEvents.events);
completer.complete();
@@ -352,8 +354,8 @@ testPause() {
expectedEvents.add(42);
c.add(42);
Expect.listEquals(expectedEvents.events, actualEvents.events);
SignalCompleter completer = new SignalCompleter();
actualEvents.pause(completer.signal);
Completer completer = new Completer();
actualEvents.pause(completer.future);
actualEvents.pause();
c.add(43);
c.add(44);
@@ -375,8 +377,8 @@ testPause() {
expectedEvents.add(42);
c.add(42);
Expect.listEquals(expectedEvents.events, actualEvents.events);
SignalCompleter completer = new SignalCompleter();
actualEvents.pause(completer.signal);
Completer completer = new Completer();
actualEvents.pause(completer.future);
actualEvents.pause();
c.add(43);
c.add(44);
tests/lib/async/stream_controller_test.dart
+1 -1
View File
@@ -231,7 +231,7 @@ testSingleController() {
c = new StreamController.singleSubscription();
var list = <int>[];
c.pipeInto(new CollectionSink<int>(list))
.then(() { Expect.listEquals(<int>[1,2,9,3,9], list); });
.whenComplete(() { Expect.listEquals(<int>[1,2,9,3,9], list); });
c.add(1);
c.add(2);
c.add(9);