Revamp Subscription, Stream APIs to make combinators easy

[cowboyd]

Right now a stream is a context-free operation that when run, yields a subscription that is coupled to the current context. And the subscription will yield the next item, and importantly, when that context passes away, the subscription will be automatically be released and do whatever cleanup is necessary.

It looks like this:

export interface Operation<T> {
  [Symbol.iterator](): Iterator<Instruction, T, any>;
}

export type Stream<T, TReturn> = Operation<Subscription<T, TReturn>>;

export type Subscription<T, TReturn> = Operation<IteratorResult<T, TReturn>>;

But this is annoying because streams and subscriptions are very similar in the way they behave. The only difference is that one is "concrete" and the other is abstract. However, we would expect the same kind of result for both yield* first(stream) and yield* first(subscription) and so we'd like for first not to care if it is operating on a Stream or a Subscription. The same applies to filter, map, forEach, etc....

I think the answer might lie in more closely aligning Stream and Subscription with Iterable and Iterator and AsyncIterable, and AsyncIterator which is what they really are: Stream === OperationIterable, Subscription === Operation Iterator.

1. AsyncIterators are AsyncIterable by returning themselves for the [Symbol.asyncIterator]() method.
2. Generators are Iterable by returning themselves for the [Symbol.iterator]() method.

I propose that we make Subscription implement Stream, by returning itself for the [Symbol.iterator] method. That way, all combinators can be implemented to work on Stream, and a live subscription is just a special type of stream to which you are already subscribed. The new types would look like:

export interface Operation<T> {
  [Symbol.iterator](): Iterator<Instruction, T, any>;
}

export type Stream<T, TReturn> = Operation<Subscription<T, TReturn>>;

export interface Subscription<T, TReturn> extends Stream<T,TReturn> {
  next(): Operation<IteratorResult<T, TReturn>>
}

Now, if we define forEach to work on a Stream:

declare function* forEach<T,R>(stream: Stream<T,R>, each?: (t: T) => Operation<void>): Operation<R>;

Then, it will work on both a stream, and a live subscription. The same applies for all our combinators you would care to write.

[taras]

This sounds like a nice simplification.

[cowboyd]

The one difficulty is that it is annoying to have to save the back reference to Subscription when creating it by hand:

let subscription = {
  *next() {
     // implement next
  },
  *[Symbol.iterator]() { return subscription; },
};
return subscription;

Perhaps we can make a helper "createSubscription()" that helps you do it my just accepting the content of next:

createSubscription(function* next() {
  // implementation of next operation.
});

So for example, the implementation of Subscription in Channel would be something like:

subscription: createSubscription(function* next() {
  let message = items.pop();
  if (message) {
    return message;
  } else {
    return yield* action<Item>(function* (resolve) {
       consumers.unshift(resolve);
    });
  }
}),

Thus far, it has been extremely rare to create subscriptions by hand, since most of the time, I just end up using Channel for lower level ops.

[cowboyd]

I'm going to advocate that we first disambiguate streams from subscriptions (you should never need to deal with a subscription directly) and then we can add make subscriptions streamable individually which will be possible if they have separate API.

[cowboyd]

Closed by #696