SSE stream

index.ts

import { PassThrough } from 'stream';
import { ChildProcessWithoutNullStreams, spawn } from 'child_process';

// process spawned executing crawler to fetch subalterns
let subalterns_task: ChildProcessWithoutNullStreams | undefined;

router.all('/api/subalterns/stop', (ctx: Context) => {
  if (!subalterns_task) {
    ctx.body = 'Task is not running.';
  } else {
    subalterns_task?.kill('SIGINT');
    ctx.body = 'Done.';
  }
});

router.all('/api/subalterns/:province_name?/:municipality_name?', (ctx: Context) => {
  const { province_name = 'ROMA', municipality_name = 'ROMA' } = <Record<string, string>>ctx.params;

  if (!subalterns_task) {
    const command_full = `node --no-warnings --experimental-specifier-resolution=node --loader=ts-node/esm -r envk src/crawler-sister/subalterns.ts --p="${province_name}" --m="${municipality_name}" --forever=1`;
    const [command, ...args] = command_full.split(' ');
    subalterns_task = spawn(command, args, { env: process.env });
    subalterns_task.on('exit', () => (subalterns_task = undefined));
  }

  ctx.request.socket.setTimeout(0);
  ctx.request.socket.setNoDelay(true);
  ctx.request.socket.setKeepAlive(true);

  ctx.set({
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  });

  const stream = new PassThrough();
  subalterns_task.stdout.pipe(stream);
  subalterns_task.stderr.pipe(stream);

  ctx.body = stream;
});