Long polling server and client

client.ts

import fetch from 'node-fetch';

const abortController = new AbortController();
const signal = abortController.signal;

process.stdin.on('readable', () => {
  const s = process.stdin.read().toString() as string;
  if (s.startsWith('abort')) {
    console.log('aborting...');
    abortController.abort();
  }
});

const fn = async () => {
  while (true) {
    try {
      console.log('pooling started...');
      const res = await fetch('http://localhost:3000/get_update/', { signal: signal as any });
      console.log('updated received...');
      console.log(await res.text());
    } catch (e) {
      if (e.name === 'AbortError') {
        console.log('aborted');
        break;
      } else {
        console.error(e);
        break;
      }
    }
  }
};

fn();

server.ts

import Koa from 'koa';
import Router from '@koa/router';
import EventEmitter from 'events';

const app = new Koa();
const router = new Router();
const m = new Map();
let listenerId = 0;

router.get('/', (ctx: any) => ctx.body = `Listeners: ${m.size}` );

router.get('/get_update', async (ctx: any) => {
  const nextId = listenerId++;
  console.log(`pooling received, listenerId ${nextId}...`);
  return new Promise<void>(
    (resolve, reject) => {
      const ee = new EventEmitter();
      m.set(nextId, ee);

      ctx.req.on('error', () => {
        console.log(`connection closed by client, listenerId ${nextId}...`);
        m.delete(nextId);
        reject();
      });

      ee.once('update', () => {
        console.log(`update sent to client, listenerId ${nextId}...`);
        ctx.body = 'Data updated';
        m.delete(nextId);
        resolve();
      });
    } );
} );

router.get('/update', (ctx: any) => {
  console.log('update posted...');
  m.forEach( (ee) => ee.emit('update') );
  ctx.body = `Update received and sent to listeners...`
 } );

app
  .use(router.routes())
  .use(router.allowedMethods());

app.listen(3000);