kriskowal / context Goto Github PK

import context from '@kriskowal/context';

// this runs "forever" (until the iterable encounters an upstream socket error, say)
async function consumeChangeStream(context, iterable) {
  for await (const entry of iterable) {
    // Make a request for each entry in the change stream. Per the example in
    // the README, this would use context.cancelled.then to register a handler
    // that invokes abort on the abortController. Since this context is never
    // directly halted until the process receives a SIGINT, the context's
    // cancelled promise will just accumulate handlers.
    await fetchWithContext(context, entry.path, { method: 'POST' });

    // Rate-limit the consumption of the change-stream (and maybe even apply
    // extra backpressure!)
    await context.delay(1000);
  }
}

const { cancel, context: procContext } = context.withCancel();
process.on('SIGINT', () => cancel(new Error('process exiting')));

consumeChangeStream(procContext, getChangeStream()).catch((err) => {
  process.nextTick(() => {
    throw err;
  });
});