implement a stream zip using Effection

zip.ts

import {
  all,
  createChannel,
  type Operation,
  resource,
  spawn,
  type Stream,
} from "effection";

export function zip<T>(streams: Stream<T, never>[]): Stream<T[], never> {
  return createStream(function* (publish) {
    let subscriptions = [];
    for (let stream of streams) {
      subscriptions.push(yield* stream);
    }
    while (true) {
      let result = yield* all(subscriptions.map((sub) => sub.next()));

      yield* publish(result.map((result) => result.value));
    }
  });
}
/**
 * takes a "publisher" which is an operation that that takes a "publish" operation
 * as its argument.
 *
 * ```javascript
 * export function toSlowStream(array) {
 *   return createStream(function*(publish) {
 *     let start = new Date();
 *     for (let item of arry) {
 *       yield* sleep(100);
 *       yield* publish(item);
 *     }
 *     let end = new Date();
 *     let totalTime = end - start;
 *     return totalTime;
 *   });
 */
function createStream<T, TClose>(
  publisher: (publish: (value: T) => Operation<void>) => Operation<TClose>,
): Stream<T, TClose> {
  return resource(function* (provide) {
    let { input, output } = createChannel<T, TClose>();
    yield* spawn(function* () {
      let result = yield* publisher(input.send);
      yield* input.close(result);
    });

    yield* provide(yield* output);
  });
}