Handling Async Flows with JS Generator Functions

// A coroutine runner takes a generator, drives it by resolving yielded promises,
// and feeds results back in via .next()
 
function run(generatorFn, ...args) {
  const generator = generatorFn(...args);
 
  return new Promise((resolve, reject) => {
    function step(method, value) {
      let result;
      try {
        result = generator[method](value);
      } catch (err) {
        return reject(err);
      }
 
      if (result.done) {
        return resolve(result.value);
      }
 
      // Handle yielded promises
      Promise.resolve(result.value)
        .then(
          resolved => step("next", resolved),
          rejected => step("throw", rejected)
        );
    }
 
    step("next", undefined);
  });
}
 
// Usage: looks like async/await but uses yield
function* fetchUserData(userId) {
  const user = yield fetch(`/api/users/${userId}`).then(r => r.json());
  const posts = yield fetch(`/api/users/${userId}/posts`).then(r => r.json());
  const comments = yield fetch(`/api/posts/${posts[0].id}/comments`).then(r => r.json());
 
  return { user, posts, comments };
}
 
// run(fetchUserData, 42).then(data => console.log(data));
 
// ENHANCED RUNNER WITH EFFECT TYPES
const EFFECTS = {
  CALL: "CALL",
  ALL: "ALL",
  RACE: "RACE",
  DELAY: "DELAY",
  FORK: "FORK"
};
 
function call(fn, ...args) {
  return { type: EFFECTS.CALL, fn, args };
}
 
function all(effects) {
  return { type: EFFECTS.ALL, effects };
}
 
function race(effects) {
  return { type: EFFECTS.RACE, effects };
}
 
function delay(ms) {
  return { type: EFFECTS.DELAY, ms };
}
 
function runEnhanced(generatorFn, ...args) {
  const gen = generatorFn(...args);
 
  return new Promise((resolve, reject) => {
    function step(method, value) {
      let result;
      try {
        result = gen[method](value);
      } catch (err) {
        return reject(err);
      }
 
      if (result.done) return resolve(result.value);
 
      const effect = result.value;
 
      if (effect && effect.type) {
        handleEffect(effect).then(
          val => step("next", val),
          err => step("throw", err)
        );
      } else {
        Promise.resolve(effect).then(
          val => step("next", val),
          err => step("throw", err)
        );
      }
    }
 
    function handleEffect(effect) {
      switch (effect.type) {
        case EFFECTS.CALL:
          return Promise.resolve(effect.fn(...effect.args));
        case EFFECTS.ALL:
          return Promise.all(effect.effects.map(e =>
            e.type ? handleEffect(e) : Promise.resolve(e)
          ));
        case EFFECTS.RACE:
          return Promise.race(Object.entries(effect.effects).map(
            ([key, e]) => (e.type ? handleEffect(e) : Promise.resolve(e))
              .then(val => ({ [key]: val }))
          ));
        case EFFECTS.DELAY:
          return new Promise(r => setTimeout(r, effect.ms));
        default:
          return Promise.resolve(effect);
      }
    }
 
    step("next", undefined);
  });
}

// SEQUENTIAL: yield one promise at a time
function* sequential() {
  const results = [];
 
  for (const url of ["/api/a", "/api/b", "/api/c"]) {
    const data = yield call(fakeFetch, url);
    results.push(data);
  }
 
  return results; // Each request waits for the previous
}
 
// PARALLEL: yield an ALL effect
function* parallel() {
  const [a, b, c] = yield all([
    call(fakeFetch, "/api/a"),
    call(fakeFetch, "/api/b"),
    call(fakeFetch, "/api/c")
  ]);
 
  return { a, b, c }; // All three run concurrently
}
 
// RACE: first to resolve wins
function* withTimeout() {
  const result = yield race({
    data: call(fakeFetch, "/api/slow-endpoint"),
    timeout: call(() => new Promise((_, reject) =>
      setTimeout(() => reject(new Error("Request timed out")), 5000)
    ))
  });
 
  if (result.timeout) {
    throw new Error("Operation timed out");
  }
 
  return result.data;
}
 
// MIXED: sequential steps with parallel sub-tasks
function* mixedFlow(userId) {
  // Step 1: fetch user (sequential)
  const user = yield call(fakeFetch, `/api/users/${userId}`);
 
  // Step 2: fetch posts and profile in parallel
  const [posts, profile] = yield all([
    call(fakeFetch, `/api/users/${userId}/posts`),
    call(fakeFetch, `/api/users/${userId}/profile`)
  ]);
 
  // Step 3: fetch comments for each post (parallel within sequential)
  const postsWithComments = yield all(
    posts.map(post => call(fakeFetch, `/api/posts/${post.id}/comments`))
  );
 
  return { user, profile, postsWithComments };
}
 
function fakeFetch(url) {
  return new Promise(resolve =>
    setTimeout(() => resolve({ url, data: "mock" }), 100)
  );
}

// Cancellation token pattern for generator-based workflows
 
class CancellationToken {
  #cancelled = false;
  #reason = null;
  #listeners = [];
 
  cancel(reason = "Cancelled") {
    if (this.#cancelled) return;
    this.#cancelled = true;
    this.#reason = reason;
    for (const listener of this.#listeners) {
      listener(reason);
    }
    this.#listeners = [];
  }
 
  get isCancelled() {
    return this.#cancelled;
  }
 
  get reason() {
    return this.#reason;
  }
 
  onCancel(listener) {
    if (this.#cancelled) {
      listener(this.#reason);
    } else {
      this.#listeners.push(listener);
    }
  }
 
  throwIfCancelled() {
    if (this.#cancelled) {
      const err = new Error(this.#reason);
      err.name = "CancellationError";
      throw err;
    }
  }
}
 
function runCancellable(generatorFn, token, ...args) {
  const gen = generatorFn(token, ...args);
 
  return new Promise((resolve, reject) => {
    // Listen for cancellation
    token.onCancel(reason => {
      try {
        gen.return(undefined); // Trigger finally blocks
      } catch (e) { /* ignore */ }
      reject(new Error(reason));
    });
 
    function step(method, value) {
      if (token.isCancelled) return;
 
      let result;
      try {
        result = gen[method](value);
      } catch (err) {
        return reject(err);
      }
 
      if (result.done) return resolve(result.value);
 
      Promise.resolve(result.value).then(
        val => step("next", val),
        err => step("throw", err)
      );
    }
 
    step("next", undefined);
  });
}
 
// Usage
function* downloadFiles(token, fileList) {
  const downloaded = [];
 
  try {
    for (const file of fileList) {
      token.throwIfCancelled();
 
      console.log(`Downloading: ${file}`);
      const data = yield fakeFetch(`/files/${file}`);
      downloaded.push({ file, data });
 
      console.log(`Completed: ${file} (${downloaded.length}/${fileList.length})`);
    }
 
    return downloaded;
  } finally {
    if (token.isCancelled) {
      console.log(`Download cancelled. ${downloaded.length} files completed.`);
      // Cleanup partial downloads
    }
  }
}
 
const token = new CancellationToken();
const promise = runCancellable(downloadFiles, token, ["a.txt", "b.txt", "c.txt"]);
 
// Cancel after 200ms
setTimeout(() => token.cancel("User pressed cancel"), 200);

// Generator-based retry with exponential backoff
 
function* retryable(operation, maxRetries = 3, baseDelay = 1000) {
  let lastError;
 
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      const result = yield call(operation);
      return result;
    } catch (err) {
      lastError = err;
      console.log(`Attempt ${attempt}/${maxRetries} failed: ${err.message}`);
 
      if (attempt < maxRetries) {
        const backoff = baseDelay * Math.pow(2, attempt - 1);
        const jitter = Math.random() * backoff * 0.1;
        yield delay(backoff + jitter);
      }
    }
  }
 
  throw lastError;
}
 
// CIRCUIT BREAKER WITH GENERATORS
function* circuitBreaker(operation, options = {}) {
  const { failureThreshold = 5, resetTimeout = 30000 } = options;
  let failures = 0;
  let circuitOpen = false;
  let lastFailure = 0;
 
  while (true) {
    // Check if circuit should reset
    if (circuitOpen && Date.now() - lastFailure > resetTimeout) {
      console.log("Circuit half-open: attempting reset");
      circuitOpen = false;
      failures = 0;
    }
 
    if (circuitOpen) {
      yield { success: false, error: "Circuit open", retryAfter: resetTimeout };
      continue;
    }
 
    try {
      const result = yield call(operation);
      failures = 0;
      yield { success: true, data: result };
    } catch (err) {
      failures++;
      lastFailure = Date.now();
 
      if (failures >= failureThreshold) {
        circuitOpen = true;
        console.log(`Circuit opened after ${failures} failures`);
      }
 
      yield { success: false, error: err.message, failures };
    }
  }
}
 
// SAGA-STYLE ERROR COMPENSATION
function* bookingTransaction(details) {
  const compensations = [];
 
  try {
    // Step 1: Reserve flight
    const flight = yield call(reserveFlight, details.flight);
    compensations.push(() => cancelFlight(flight.id));
 
    // Step 2: Reserve hotel
    const hotel = yield call(reserveHotel, details.hotel);
    compensations.push(() => cancelHotel(hotel.id));
 
    // Step 3: Reserve car
    const car = yield call(reserveCar, details.car);
    compensations.push(() => cancelCar(car.id));
 
    // Step 4: Charge payment
    const payment = yield call(chargePayment, details.payment);
    compensations.push(() => refundPayment(payment.id));
 
    return { flight, hotel, car, payment, status: "confirmed" };
 
  } catch (err) {
    // Compensate in reverse order (undo completed steps)
    console.log(`Booking failed: ${err.message}. Running compensations...`);
 
    for (let i = compensations.length - 1; i >= 0; i--) {
      try {
        yield call(compensations[i]);
      } catch (compErr) {
        console.error(`Compensation ${i} failed: ${compErr.message}`);
      }
    }
 
    throw new Error(`Booking failed and rolled back: ${err.message}`);
  }
}
 
function reserveFlight(d) { return Promise.resolve({ id: "FL-001" }); }
function reserveHotel(d) { return Promise.resolve({ id: "HT-001" }); }
function reserveCar(d) { return Promise.resolve({ id: "CR-001" }); }
function chargePayment(d) { return Promise.resolve({ id: "PY-001" }); }
function cancelFlight(id) { return Promise.resolve(); }
function cancelHotel(id) { return Promise.resolve(); }
function cancelCar(id) { return Promise.resolve(); }
function refundPayment(id) { return Promise.resolve(); }

// Generators guarantee finally blocks run on .return() or completion
 
function* managedConnection(config) {
  const connection = yield call(createConnection, config);
  console.log("Connection established");
 
  try {
    // Yield the connection for the caller to use
    while (true) {
      const query = yield { connection, status: "ready" };
      if (!query) break;
 
      const result = yield call(executeQuery, connection, query);
      yield { result, status: "completed" };
    }
  } finally {
    // Always runs: normal completion, .return(), or .throw()
    console.log("Closing connection");
    yield call(closeConnection, connection);
    console.log("Connection closed");
  }
}
 
function createConnection(config) {
  return Promise.resolve({ id: Date.now(), config });
}
function executeQuery(conn, query) {
  return Promise.resolve({ rows: [], query });
}
function closeConnection(conn) {
  return Promise.resolve();
}
 
// MULTIPLE RESOURCE MANAGEMENT
function* withResources(resourceFactories) {
  const resources = [];
 
  try {
    // Acquire all resources
    for (const factory of resourceFactories) {
      const resource = yield call(factory.acquire);
      resources.push({ resource, release: factory.release });
    }
 
    // Yield resources for use
    return yield resources.map(r => r.resource);
 
  } finally {
    // Release in reverse order
    for (let i = resources.length - 1; i >= 0; i--) {
      try {
        yield call(resources[i].release, resources[i].resource);
      } catch (err) {
        console.error(`Failed to release resource ${i}: ${err.message}`);
      }
    }
  }
}
 
// DISPOSABLE PATTERN
function* using(resourceGen, bodyFn) {
  const resource = yield* resourceGen;
 
  try {
    const result = yield* bodyFn(resource);
    return result;
  } finally {
    if (resource && typeof resource.dispose === "function") {
      yield call(() => resource.dispose());
    }
  }
}

// Redux saga-inspired pattern using generators for side effect management
 
class SagaRuntime {
  #store;
  #sagas = new Map();
  #running = new Set();
 
  constructor(store) {
    this.#store = store;
  }
 
  run(saga, ...args) {
    const gen = saga(this.#store.getState, ...args);
    const id = Symbol("saga");
    this.#running.add(id);
 
    return this.#drive(gen, id);
  }
 
  async #drive(gen, id) {
    let input = undefined;
 
    while (this.#running.has(id)) {
      const { value: effect, done } = gen.next(input);
 
      if (done) {
        this.#running.delete(id);
        return effect;
      }
 
      try {
        input = await this.#resolve(effect);
      } catch (err) {
        const { value, done: errDone } = gen.throw(err);
        if (errDone) {
          this.#running.delete(id);
          return value;
        }
        input = value;
      }
    }
  }
 
  async #resolve(effect) {
    if (!effect || !effect.type) {
      return effect;
    }
 
    switch (effect.type) {
      case "CALL":
        return effect.fn(...effect.args);
      case "SELECT":
        return effect.selector(this.#store.getState());
      case "PUT":
        return this.#store.dispatch(effect.action);
      case "TAKE":
        return this.#waitForAction(effect.actionType);
      case "ALL":
        return Promise.all(effect.effects.map(e => this.#resolve(e)));
      case "DELAY":
        return new Promise(r => setTimeout(r, effect.ms));
      default:
        return effect;
    }
  }
 
  #waitForAction(actionType) {
    return new Promise(resolve => {
      const unsub = this.#store.subscribe(() => {
        const action = this.#store.getLastAction();
        if (action && action.type === actionType) {
          unsub();
          resolve(action);
        }
      });
    });
  }
 
  cancel(id) {
    this.#running.delete(id);
  }
}
 
// Saga effect creators
function select(selector) {
  return { type: "SELECT", selector };
}
 
function put(action) {
  return { type: "PUT", action };
}
 
function take(actionType) {
  return { type: "TAKE", actionType };
}
 
// Example saga
function* loginSaga(getState) {
  while (true) {
    const action = yield take("LOGIN_REQUEST");
 
    try {
      yield put({ type: "LOGIN_LOADING" });
 
      const response = yield call(
        fakeFetch,
        `/api/login?user=${action.payload.username}`
      );
 
      yield put({ type: "LOGIN_SUCCESS", payload: response });
    } catch (err) {
      yield put({ type: "LOGIN_FAILURE", payload: err.message });
    }
  }
}