Add StreamQueue.startTransactions().

[nex3]

This is important for advanced pull-based stream manipulation. It allows
users to express logic of the form "consume the next events if they
match this predicate".

[lrhn]

I think it's OK, the comments are just suggestions.

[lrhn]

A TransactionStreamQueue is itself a StreamQueue, so you can start another transaction on it. What happens if you call accept while a transaction is open?
That is:

var sq = ... some stream queue ...;
var t1 = sq.startTransaction();
await t1.next();
var t2 = t1.startTransaction();
await t2.next();
t1.accept();
// I guess this moves sq past the first event, then makes it stop at the 
// start transaction request, and t2 is still active. 
await t2.next();
t2.accept();
// Moves sq past two more events.

(Or is this prevented because the startTransaction request would make the request-queue non-empty?)

[nex3]

(Or is this prevented because the startTransaction request would make the request-queue non-empty?)

Yep. An open transaction counts as a pending request.

[lrhn]

I'm not particularly happy about the TransactionStreamQueue. I was never happy about my own designs for this feature either :)

There is no object representing the transaction itself, only a number of streams that each represent part of it.

Would it make sense to have a StreamQueueTransaction object returned by startTransaction():

class StreamQueueTransaction {
  /// Create a new queue running under the transaction.
  ///
  /// The queue gets the events of the original stream, and must either be 
  /// [reject]ed or [commit]ted.
  /// Cannot be called after [cancel] or [commit].
  StreamQueue newQueue();
  /// Forwards the original queue to where [queue] is.
  ///
  /// The [queue] must have been returned by [newQueue] and not 
  /// have been rejected.
  /// If [queue] has an open transaction, that transaction effectively becomes
  /// a transaction of the original stream, and can continue being used.
  /// Rejects all other open queues of the transaction.
  /// Cannot be called after [cancel] has been called.
  void commit(StreamQueue queue);
  /// Cancels the transaction.
  ///
  /// Continues the original stream as if the transaction had consumed no events.
  /// Cannot be called after [commit] has been called.
  /// Rejects all open queues of the transaction.
  void cancel();
  /// The queue, which must have been created by [newQueue], is rejected.
  ///
  /// The queue will not receive any further events.
  /// If [cancelIfLast] is true and there are no further open streams in the 
  /// transaction, the transaction is [cancel]'ed.
  void reject(StreamQueue queue, {cancelIfLast: false});
}

That way you don't need to extend StreamQueue, and all the transaction logic is kept in a single object.
Also, you don't need to know the number of streams when creating the transaction (that might be the only real advantage).

Since there is a strong connection between the stream state and whether you can call accept, keeping them together might make sense.

[nex3]

I like that. It'll be somewhat more verbose, but transactions are a pretty low-level construct and hopefully people will mostly use them through utility APIs.

With a dedicated transaction object, I don't think you even need the ability to reject individual queues. The user can just ignore them once they're no longer relevant—or call queue.cancel().

[lrhn]

The plural name suggests that each stream is an individual transaction, but they are all part of the same transaction (since you can only commit one of them). So maybe startParallelTransaction(int streamCount) (because startTransaction is taken below), or combine the two and make [int count = 1] optional.

[lrhn]

Would it make sense to make commit a request as well? So, you can do:

var sq = ...;
var s1 = sq.startTransaction();
int count = s1.next();
if (count < 0) {
  s1.reject();
} else {
  s1.skip(count);
  s1.commit();  // Handled after the skips.
}
print(await sq.next());

It's not a big difference - I could just add an await before the skip call, but as a request it avoids having this test and error message.

[lrhn]

Instead of using a StreamSplitter, could you just have made _controller.stream a broadcast stream?
(Maybe not, if StreamSplitter pauses when all the splits are paused - we probably want that).

[lrhn]

Just: result.addTo(_controller);

[nex3]

Done.

[lrhn]

Alternatively (because who doesn't like alternatives), would it make sense to commit with requests still pending, even if commit is not a request?
Something like the example above, where you ask to skip some events, but you don't really need to wait for them to arrive, you just want to move the requests into the outer queue's requests, so it'll wait for them there.

That would make sense, but obviously the requests must be built to withstand being moved (they shouldn't see a difference in events, but if they have a reference to the queue itself, then they are not movable, and I think that's the case for at least some of them. Maybe as a later update, if it turns out to be needed).

[nex3]

I went with the "just throw" solution because I didn't think it was really clear which of the several possible behaviors was the most obvious or most useful, and I couldn't come up with any really compelling use-cases that weren't supportable in other ways. If we throw for now it leaves us open to choose a different behavior in the future if it becomes clear that one is correct.

[lrhn]

LGTM!