Effection v3 And Deno.run

README.md

Notes

Run it via deno run -A main.ts. It runs and uses the child_process from Deno.run to run deno run -A server.ts.

We were having issues with the process, a daemon aka long-running process, not sticking around. The commentted sleep() was for testing as opposed to a signal listener.

deno.json

{
  "lint": {
    "rules": {
      "exclude": ["prefer-const", "require-yield"]
    }
  }
}

main.ts

import {
  run,
  resource,
  sleep,
  subscribe,
  spawn,
  expect,
  action,
  type Operation,
  type Subscription,
  suspend,
} from "https://deno.land/x/effection/mod.ts";

await run(function* () {
  console.log("boop");
  let child = yield* exec({ cmd: "deno run -A server.ts".split(" ") });

  // yield* sleep(2000)
  yield* action<void>(function* (resolve) {
    Deno.addSignalListener("SIGINT", resolve);
    try {
      yield* suspend();
    } finally {
      Deno.removeSignalListener("SIGINT", resolve);
    }
  });
  console.log("booped");
});

interface Process {
  pid: number;
  stdout: Subscription<Uint8Array, void>;
  stderr: Subscription<Uint8Array, void>;
}

function exec({ cmd }: { cmd: string[] }): Operation<Process> {
  return resource(function* (provide) {
    yield* useInterval(() => {}, 5000_000);
    let p = Deno.run({ cmd, stdout: "piped", stderr: "piped" });
    console.dir(p);
    let stdout = subscribe(p.stdout.readable[Symbol.asyncIterator]());
    let stderr = subscribe(p.stderr.readable[Symbol.asyncIterator]());
    try {
      yield* provide({ pid: p.pid, stdout, stderr });
    } finally {
      console.dir(p);
      let shutdown = yield* spawn(function* () {
        let output = yield* expect(p.output());
        let stderrOutput = yield* expect(p.stderrOutput());
        let status = yield* expect(p.status());
        let decoder = new TextDecoder();
        return {
          status,
          stdout: decoder.decode(output),
          stderr: decoder.decode(stderrOutput),
        };
      });
      p.kill("SIGINT");
      let result = yield* shutdown;
      console.dir(result);
      if (result.status.code !== 0) {
        throw new Error(`stderr: ${result.stderr}`);
      }
    }
  });
}

function useInterval(code: () => void, delayMS: number): Operation<void> {
  return resource(function* (provide) {
    let id = setInterval(code, delayMS);
    try {
      yield* provide();
    } finally {
      clearInterval(id);
    }
  });
}

function* forEach<T, R>(
  subscription: Subscription<T, R>,
  op?: (value: T) => Operation<void>
) {
  while (true) {
    let next = yield* subscription;
    // console.dir(next);
    if (next.done) {
      return next.value;
    } else if (op) {
      yield* op(next.value);
    }
  }
}