Add support for Promise-based iterators

[bminer]

So, what about adding some API functions:

• s.nextEvent(eventName) - basically returns new Promise(resolve => s.once(eventName, (...args) => resolve(args))). Not a huge fan of the name nextEvent, but it's an idea (also thought of waitFor, onceAsync, etc.). Also considered having s.once(eventName) return a Promise that resolves when we receive eventName... but then I think we'd need a corresponding s.on(eventName) that returns an AsyncIterator, but I don't like that idea (see s.iterate below)
• s.iterate(eventName) - sets up an event handler for eventName on the channel and returns an AsyncIterator. When the user calls next(), the next value is retrieved. I personally don't like this API very much because events can be received and buffered without being immediately consumed (i.e. next() can be called much later after receiving). It defers event processing until event consumption, which feels bad to me... and we resort to buffering when normally buffering is not needed. One idea I have is that we can buffer exactly one value. If it is not immediately consumed by the next run of the event loop, then we remove the event handler and the AsyncIterator returns a final error.
• Event handlers could be async generator functions or return AsyncIterators. In this case, the .request() method would return a Promise that resolves to an AsyncIterator. Calling .next(value) on this would pass the value to the remote end, much like a request and the return value of .next(...) is the value received by the AsyncIterator. If this works, even Go event handlers could return a channel as its data, and values are read from the channel and forwarded to the remote end until the channel closes.

[bminer]

The more I think about it, I really like the idea of event handlers returning iterable iterators, but before adding this feature, I really need to think through it. On the request side, the request method would return a Promise resolving to an AsyncIterator. But, now we have a interesting dilemma...

There will likely be two types of iterators:

• One-way iterators: the event handler emits values as quickly as possible with neither buffering nor back-pressure. (probably the most common). This feels a lot like "event handling."
• Two-way iterators: technically the AsyncIterator's next method can accept a value to support two-way communication (like coroutines) with the event handler. This feel a lot like "request / response".

This library (and underlying protocol) pretty much already supports both types of iterators. This is just syntactic sugar on top.

// Remote side (event handler)
// We simply default to being a one-way iterator
socket.on("fibonacci", function*(n) {
  let a = 0;
  let b = 1;

  for (let i = 0; i < n; i++) {
    // Notice we don't use the yield expression, but ws-wrapper doesn't know this
    yield a;
    [a, b] = [b, a + b];
  }
})
// Requestor side
// Note that we don't call `iter.next(value)` with a value, but ws-wrapper doesn't know this.
// We simply default to being a one-way iterator and ignore any value passed to `iter.next`
const iter = await socket.request("fibonacci", 10)
// for await loops don't pass a value to `iter.next`
for await (const num of iter) {
  console.log(num);
  break; // Closes iterator, triggers return
}

Under the hood, this is really just an anonymous event handler registration valid while the iterator is valid. After the iterator's next has returned {done: true}, we can remove the anonymous event handler.

For two-way iterators, this is really just a request / response with an anonymous event name.

// Remote side (event handler) with two-way communication needed
socket.on("add", function*() {
  let sum = 0;
  while (true) {
    const input = yield sum; 
    sum += input;
  }
})
// Requestor side (use `negotiate` instead of `request` to indicate two-way communication is needed)
const iter = await socket.negotiate("add")
await iter.next() // {value: 0, done: false}
await iter.next(20) // {value: 20, done: false}
await iter.next(35) // {value: 55, done: false}
await iter.return() // {value: undefined, done: true}

In the above, if socket.request was mistakenly used, the remote end will call next.value() over and over in an infinite loop causing the CPU to spin. 🤖💥

I am now quite convinced this is the "wrong" approach. Instead of indicating the iterator style on the requestor side, it should be on the event handler side. Why not have the event handler request values from the client over a prenegotiated anonymous channel / event name?

So what about something like this?

// Remote side
function* fibo(n) {
  let a = 0;
  let b = 1;

  for (let i = 0; i < n; i++) {
    yield a;
    [a, b] = [b, a + b];
  }
}
socket.on("fibonacci", (n) => {
  const g = this.generator()
  for(const val of fibo(n)) {
    g.emit(val) // communicates to ws-wrapper that only one-way communication is needed
  }
  return "the end" // return value is ignored by requestor's `for...of` loop but still gets sent back
})
// Requestor side is the same; `socket.request` returns a Promise resolving to an async iterator
for await (const num of socket.request("fibonacci", 10)) {
  console.log(num);
}

And for two-way communication:

// Remote side
function* adder() {
  let sum = 0;
  while (true) {
    const input = yield sum; 
    sum += input;
  }
}
socket.on("add", () => {
  const g = this.generator()
  const addGen = adder()
  // Get the first value from the generator
  let {value, done} = addGen.next()
  while(!done) {
    // Get the addend from the remote end's async iterator
    // Note: we use `g.request` instead of `g.emit` to indicate two-way communication
    const {value: addend, done: remoteDone} = await g.request(value)
    if(remoteDone) {
      // Remote end called `iter.return`, so we will do the same
      // addend will be `undefined` in this case (matching the value passed to `return`)
      return addGen.return(addend).value
    }
    // Remote end called `iter.next`, so we will do the same
    const result = addGen.next(addend)
    // Update value / done
    value = result.value
    done = result.done
  }
})
// Requestor side
const iter = await socket.request("add")
await iter.next() // {value: 0, done: false}
await iter.next(20) // {value: 20, done: false}
await iter.next(35) // {value: 55, done: false}
await iter.return() // {value: undefined, done: true}

[bminer]

Generalizing this a bit more... what if we added the concept of "anonymous channels" that were scoped to a given request?

We first add a close() method to channels that remove them from the socket wrapper:

const newsCh = socket.of("news")
newsCh.on("headline", (headline) => console.log("headline", headline))
newsCh.close() // stops listening for events on the news channel and cleans up

Now we can do this:

// Remote end
socket.on("fibonacci", (n) => {
  // Returning an anonymous channel sends a new response type `{i, h: 1}` with `h` set to a truthy value
  const chan = this.anonymousChannel(async (chan) => {
    // this optional callback is executed after we send the request response
    for(const val of fibo(n)) {
      chan.emit("value", val)
    }
    chan.emit("done")
    // anonymous channel automatically closes at the end of the callback or if timeout / cancellation occurs
  })
  // We can set up any event handlers before returning chan
  chan.on("echo", (str) => str)
  // We are not allowed to call `emit` or `request` yet. This must be done in the callback.
  // chan.emit("value", 23) // Throws an error!
  return chan
})
// Requestor end
const chan = await socket.request("fibonacci", 10)
chan.on("value", (val) => {
  console.log("fibo", val)
})
chan.on("done", () => chan.close())
// anonymous channel is also auto-closed if request timeout / cancellation occurs

Or perhaps the API is more like this:

// Remote end
socket.on("fibonacci", (n) => {
  // Returning an anonymous channel sends a new response type `{i, h: 1}` with `h` set to a truthy value
  const chan = this.anonymousChannel()
  // We can set up any event handlers before returning chan
  chan.on("start", () => {
    for(const val of fibo(n)) {
      chan.emit("value", val)
    }
    chan.emit("done")
    chan.close()
  })
  // We are not allowed to call `emit` or `request` yet. This must be done in a callback.
  // chan.emit("value", 23) // Throws an error!
  return chan
  // anonymous channel is also auto-closed if request timeout / cancellation occurs
})

// Requestor end
const chan = await socket.request("fibonacci", 10)
chan.on("value", (val) => {
  console.log("fibo", val)
})
chan.on("done", () => chan.close())
chan.emit("start")
// anonymous channel is also auto-closed if request timeout / cancellation occurs

Now we can add wrappers to support one-way or two-way async iterators.

We may also want channels themselves to have a timeout or cancellation signal... where if no message is received within the timeout, they auto-close themselves.

At the protocol level, we would add a field h that is sent with the request response, indicating that an anonymous channel was created for the request. For example {i: 7, h: 1} means that request 7 called an event handler that returned an anonymous channel. Even though the request is "complete", we can still send a cancellation message for request 7, and it will close any anonymous channels on the remote end.

We might also need to update the chan.signal method to allow chaining of abort signals via AbortSignal.any(...).