Streaming Bitwise Adder in Effection

adder.ts

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

type bit = 0 | 1;
type Bit = Stream<bit, never>;

function create1BitAdder(a$: Bit, b$: Bit, c$: Bit): Stream<[bit, bit], never> {
  // the easiest way to represent a stream is as a resource because
  // a resource that "provides" a subscription is by definition a stream
  // because a stream is an operation that yields a subscription
  return resource(function*(provide) {
    //this is the channel that will transmit the output
    let { input, output } = createChannel<[bit, bit], never>();
    let as = yield* a$;
    let bs = yield* b$;
    let cs = yield* c$;

    // when the resource is evaluated, spawn a task to read the
    // inputs and pump results onto the output channel.
    yield* spawn(function*() {
      let [a, b, c] = yield* all([as.next(), bs.next(), cs.next()]);

      while (true) {
        let sout = (c.value ^ (a.value ^ b.value)) as bit;
        let cout = (a.value & b.value | b.value & c.value | a.value & c.value) as bit;

        yield* input.send([sout, cout]);

        [a, b, c ] = yield* all([as.next(), bs.next(), cs.next()]);
      }
    });

    // provide a subscription to the output stream of the channel
    yield* provide(yield* output);
  });
}

await main(function*() {
  let a = createChannel<bit, never>();
  let b = createChannel<bit, never>();
  let c = createChannel<bit, never>();

  let adder = create1BitAdder(a.output, b.output, c.output);

  yield* spawn(function*() {
    let subscription = yield* adder;
    let next = yield* subscription.next();
    while (!next.done) {
      let [sOut, cOut] = next.value;
      console.dir({ sOut, cOut });
      next = yield* subscription.next();
    }
  });

  yield* a.input.send(0);
  yield* a.input.send(1);
  yield* a.input.send(0);

  yield* b.input.send(0);
  yield* b.input.send(0);
  yield* b.input.send(1);

  yield* c.input.send(1);
  yield* c.input.send(0);
  yield* c.input.send(1);
});