Run an operation on a heartbeat, only one operation at a time

heartbeat.ts

import {
  call,
  createSignal,
  each,
  type Operation,
  spawn,
  type Stream,
  suspend,
} from "effection";

/**
 * Create an infinite stream of pulses that happen on the given period.
 */
export function pulse(period: number): Stream<void, never> {
  return {
    *subscribe() {
      let signal = createSignal<void, never>();
      yield* spawn(function* () {
        let intervalId = setInterval(signal.send, period);
        try {
          yield* suspend();
        } finally {
          clearInterval(intervalId);
        }
      });
      return yield* signal.subscribe();
    },
  };
}
/**
 * Run an operation on a heartbeat loop.
 * For each tick of the loop, spawn the operation and await the next tick.
 * If the next pulse happens while the spawned operation is still running, it will be shut 
 * down because it passes out of scope when the `call()` returns.
 */
export function* heartbeat(period: number, operation: () => Operation<unknown>): Operation<void> {
  for (let _ of yield* each(pulse(period))) {
    yield* call(function* () {
      yield* spawn(operation);
      yield* each.next();
    });
  }
}