Generators.observe needs a better way to throw? (And end?)

[mbostock]

It’s possible if you notify with a rejected Promise, but this seems a little awkward. For example:

Generators.observe(notify => {
  const socket = new WebSocket("wss://ws.blockchain.info/inv");
  socket.onerror = () => notify(Promise.reject(new Error("socket error")));
  socket.onopen = () => socket.send(JSON.stringify({op: "unconfirmed_sub"}));
  socket.onmessage = event => notify(JSON.parse(event.data));
  return () => socket.close();
})

We could pass two callbacks, resolve and reject, instead (or next and error for symmetry with the Observable proposal; maybe it would be confusing to use the name resolve when it can be called repeatedly)?

Generators.observe((resolve, reject) => {
  const socket = new WebSocket("wss://ws.blockchain.info/inv");
  socket.onerror = () => reject(new Error("socket error"));
  socket.onopen = () => socket.send(JSON.stringify({op: "unconfirmed_sub"}));
  socket.onmessage = event => resolve(JSON.parse(event.data));
  return () => socket.close();
})

But even then we don’t have a way to indicate that the observable stream has ended (without an error). That would correspond to the observer.complete method. If we passed an observer in instead, that would look like:

Generators.observe(observer => {
  const socket = new WebSocket("wss://ws.blockchain.info/inv");
  socket.onerror = () => observer.error(new Error("socket error"));
  socket.onopen = () => socket.send(JSON.stringify({op: "unconfirmed_sub"}));
  socket.onmessage = event => observer.next(JSON.parse(event.data));
  return () => socket.close();
})

[randName]

Currently the return method does not actually cause Generator.observe to finish.

I propose modifying the return method to fit the description in the MDN docs

diff a/src/generators/observe.js b/src/generators/observe.js
@@ near the start @@
+  let done = false;

@@ near the end @@
   function next() {
-    return {done: false, value: stale
+    return {done, value: stale
         ? (stale = false, Promise.resolve(value))
         : new Promise(_ => (resolve = _))};
   }

   return {
     [Symbol.iterator]: that,
     throw: () => ({ done }),
-    return: () => (dispose != null && dispose(), {done: true}),
+    return: (v) => (
+      change(v), done = true, stale = true,
+      dispose != null && dispose(), {done, value}
+    ),
     next
   }

Similar modifications can be made to throw too. The only major issue is that you need a reference to the generator object to make use of return and throw, unless this is bound.

Currently my use case is that I have some object that I am updating in seperate async functions that can run in parallel, but I want to yield on every change (not sure if this is an anti-pattern).

const stuff= { /* some shared state */ };
const g = observe((notify) => {
  const work = [ /* multiple async functions that call notify */ ];
  Promise.all(work).then(() => g.return(stuff));
});
yield * g;
// do other stuff. current Generators.observe will never reach here

[kcarnold]

Possible example strategy (but quite complex) from node.js: https://github.com/nodejs/node/blob/main/lib%2Fevents.js#L1061

h/t: https://www.reddit.com/r/javascript/comments/1c3gkr1/comment/kzgxjv8/

[kcarnold]

Maybe better: https://github.com/azu/events-to-async