Merge pull request #18 from dart-lang/transaction-helpers
Add StreamQueue transaction helpers.
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e51766e..615e33e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,9 +3,12 @@
* Add an `AsyncCache` class that caches asynchronous operations for a period of
time.
-* Add `StreamQueue.startTransaction()` and `StreamQueue.startTransactions()`.
+* Add `StreamQueue.startTransaction()` and `StreamQueue.withTransaction()`.
These allow users to conditionally consume events based on their values.
+* Add `StreamQueue.cancelable()`, which allows users to easily make a
+ `CancelableOperation` that can be canceled without affecting the queue.
+
## 1.11.3
* Fix strong-mode warning against the signature of Future.then
diff --git a/lib/src/stream_queue.dart b/lib/src/stream_queue.dart
index 051f8b2..f770e16 100644
--- a/lib/src/stream_queue.dart
+++ b/lib/src/stream_queue.dart
@@ -7,6 +7,7 @@
import 'package:collection/collection.dart';
+import "cancelable_operation.dart";
import "result.dart";
import "subscription_stream.dart";
import "stream_completer.dart";
@@ -228,9 +229,11 @@
///
/// Until the transaction finishes, this queue won't emit any events.
///
+ /// See also [withTransaction] and [cancelable].
+ ///
/// ```dart
/// /// Consumes all empty lines from the beginning of [lines].
- /// Future consumeEmptyLines(StreamQueue<String> lines) {
+ /// Future consumeEmptyLines(StreamQueue<String> lines) async {
/// while (await lines.hasNext) {
/// var transaction = lines.startTransaction();
/// var queue = transaction.newQueue();
@@ -238,7 +241,7 @@
/// transaction.reject();
/// return;
/// } else {
- /// transaction.accept(queue);
+ /// transaction.commit(queue);
/// }
/// }
/// }
@@ -251,6 +254,82 @@
return request.transaction;
}
+ /// Passes a copy of this queue to [callback], and updates this queue to match
+ /// the copy's position if [callback] returns `true`.
+ ///
+ /// This queue won't emit any events until [callback] returns. If it returns
+ /// `false`, this queue continues as though [withTransaction] hadn't been
+ /// called. If it throws an error, this updates this queue to match the copy's
+ /// position and throws the error from the returned `Future`.
+ ///
+ /// Returns the same value as [callback].
+ ///
+ /// See also [startTransaction] and [cancelable].
+ ///
+ /// ```dart
+ /// /// Consumes all empty lines from the beginning of [lines].
+ /// Future consumeEmptyLines(StreamQueue<String> lines) async {
+ /// while (await lines.hasNext) {
+ /// // Consume a line if it's empty, otherwise return.
+ /// if (!await lines.withTransaction(
+ /// (queue) async => (await queue.next).isEmpty)) {
+ /// return;
+ /// }
+ /// }
+ /// }
+ /// ```
+ Future<bool> withTransaction(Future<bool> callback(StreamQueue<T> queue)) {
+ var transaction = startTransaction();
+
+ /// Avoid async/await to ensure that [startTransaction] is called
+ /// synchronously and so ends up in the right place in the request queue.
+ var queue = transaction.newQueue();
+ return callback(queue).then((result) {
+ if (result) {
+ transaction.commit(queue);
+ } else {
+ transaction.reject();
+ }
+ }, onError: (error) {
+ transaction.commit(queue);
+ throw error;
+ });
+ }
+
+ /// Passes a copy of this queue to [callback], and updates this queue to match
+ /// the copy's position once [callback] completes.
+ ///
+ /// If the returned [CancelableOperation] is canceled, this queue instead
+ /// continues as though [cancelable] hadn't been called. Otherwise, it emits
+ /// the same value or error as [callback].
+ ///
+ /// See also [startTransaction] and [withTransaction].
+ ///
+ /// ```dart
+ /// final _stdinQueue = new StreamQueue(stdin);
+ ///
+ /// /// Returns an operation that completes when the user sends a line to
+ /// /// standard input.
+ /// ///
+ /// /// If the operation is canceled, stops waiting for user input.
+ /// CancelableOperation<String> nextStdinLine() =>
+ /// _stdinQueue.cancelable((queue) => queue.next);
+ /// ```
+ CancelableOperation/*<S>*/ cancelable/*<S>*/(
+ Future/*<S>*/ callback(StreamQueue<T> queue)) {
+ var transaction = startTransaction();
+ var completer = new CancelableCompleter/*<S>*/(onCancel: () {
+ transaction.reject();
+ });
+
+ var queue = transaction.newQueue();
+ completer.complete(callback(queue).whenComplete(() {
+ if (!completer.isCanceled) transaction.commit(queue);
+ }));
+
+ return completer.operation;
+ }
+
/// Cancels the underlying event source.
///
/// If [immediate] is `false` (the default), the cancel operation waits until
diff --git a/test/stream_queue_test.dart b/test/stream_queue_test.dart
index d5ab74b..02a7733 100644
--- a/test/stream_queue_test.dart
+++ b/test/stream_queue_test.dart
@@ -800,6 +800,106 @@
});
});
+ group("withTransaction operation", () {
+ StreamQueue<int> events;
+ setUp(() async {
+ events = new StreamQueue(createStream());
+ expect(await events.next, 1);
+ });
+
+ test("passes a copy of the parent queue", () async {
+ await events.withTransaction(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ expect(await queue.next, 3);
+ expect(await queue.next, 4);
+ expect(await queue.hasNext, isFalse);
+ return true;
+ }));
+ });
+
+ test("the parent queue continues from the child position if it returns "
+ "true", () async {
+ await events.withTransaction(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ return true;
+ }));
+
+ expect(await events.next, 3);
+ });
+
+ test("the parent queue continues from its original position if it returns "
+ "false", () async {
+ await events.withTransaction(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ return false;
+ }));
+
+ expect(await events.next, 2);
+ });
+
+ test("the parent queue continues from the child position if it throws", () {
+ expect(events.withTransaction(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ throw "oh no";
+ })), throwsA("oh no"));
+
+ expect(events.next, completion(3));
+ });
+ });
+
+ group("cancelable operation", () {
+ StreamQueue<int> events;
+ setUp(() async {
+ events = new StreamQueue(createStream());
+ expect(await events.next, 1);
+ });
+
+ test("passes a copy of the parent queue", () async {
+ await events.cancelable(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ expect(await queue.next, 3);
+ expect(await queue.next, 4);
+ expect(await queue.hasNext, isFalse);
+ })).value;
+ });
+
+ test("the parent queue continues from the child position by default",
+ () async {
+ await events.cancelable(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ })).value;
+
+ expect(await events.next, 3);
+ });
+
+ test("the parent queue continues from the child position if an error is "
+ "thrown", () async {
+ expect(events.cancelable(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ throw "oh no";
+ })).value, throwsA("oh no"));
+
+ expect(events.next, completion(3));
+ });
+
+ test("the parent queue continues from the original position if canceled",
+ () async {
+ var operation = events.cancelable(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ }));
+ operation.cancel();
+
+ expect(await events.next, 2);
+ });
+
+ test("forwards the value from the callback", () async {
+ expect(await events.cancelable(expectAsync1((queue) async {
+ expect(await queue.next, 2);
+ return "value";
+ })).value, "value");
+ });
+ });
+
test("all combinations sequential skip/next/take operations", () async {
// Takes all combinations of two of next, skip and take, then ends with
// doing rest. Each of the first rounds do 10 events of each type,
