Reject pending promises when a generator is closed

src/generators/observe.js

@@ -4,6 +4,7 @@ export function observe(initialize) {
   let stale = false;
   let value;
   let resolve;
+  let reject;
   const dispose = initialize(change);
 
   if (dispose != null && typeof dispose !== "function") {
@@ -13,21 +14,26 @@ export function observe(initialize) {
   }
 
   function change(x) {
-    if (resolve) resolve(x), resolve = null;
+    if (resolve) resolve(x), resolve = reject = null;
     else stale = true;
     return value = x;
   }
 
   function next() {
     return {done: false, value: stale
         ? (stale = false, Promise.resolve(value))
-        : new Promise(_ => (resolve = _))};
+        : new Promise((res, rej) => (resolve = res, reject = rej))};
   }
 
   return {
     [Symbol.iterator]: that,
     throw: () => ({done: true}),
-    return: () => (dispose != null && dispose(), {done: true}),
+    return: () => {
+      dispose != null && dispose();
+      // Reject pending promises
+      if (reject) reject(new Error("Generator returned")), resolve = reject = null;
+      return {done: true};
+    },
     next
   };
 }

test/generators/generators-test.js

@@ -56,3 +56,19 @@ it("queue yields all values", async () => {
   assert.strictEqual(await q.next().value, 1);
   assert.strictEqual(await q.next().value, 2);
 });
+
+it("observe rejects pending promise on return", async () => {
+  let o = observe(() => {});
+  const pending = o.next().value;
+  o.return();
+  await assert.rejects(pending, /Generator returned/);
+});
+
+it("observe does not reject if no pending promise on return", async () => {
+  let o = observe(change => {
+    change(1);
+  });
+  assert.strictEqual(await o.next().value, 1);
+  // No pending promise, so return should complete without error
+  assert.deepStrictEqual(o.return(), {done: true});
+});