handlerSourcePoC.ts

/// <reference path="../node_modules/@types/node/index.d.ts" />

'use strict'

import { newStream, tap, runEffects } from '@most/core';
import { newDefaultScheduler } from '@most/scheduler';
import { Sink, Scheduler } from 'most';
import { IncomingMessage, ServerResponse, createServer } from 'http';

interface Ctx {
  req: IncomingMessage;
  res: ServerResponse;
}

/**
 * Current State: Broken
 * 
 * PoC for a middleware handler factory for nodejs webservers such as
 * expressjs, connectjs, and vanilla http/https server. Create a
 * handler using handlerSourceFactory, wrap it in Stream, then
 * handler can be used as middleware and events will propogate through
 * the stream.
 */

const handlerSourceFactory = () => {
  console.log('Handler created'); // Remove
  let handler: any = () => undefined;
  const run = (sink: Sink<Ctx>, scheduler: Scheduler) => {
    console.log('Something subscribed'); // Remove
    handler = (req: IncomingMessage, res: ServerResponse) => {
      console.log('Handling event'); // Remove
      tryEvent(scheduler.now(), {req, res}, sink);
    }
    const dispose = () => {
      handler = () => undefined;
      return Promise.resolve();
    }
    return { dispose };
  }
  handler.run = run;
  return handler
}

function tryEvent (t: number, x: Ctx, sink: Sink<Ctx>) {
  console.log('Trying event'); // Remove
  try {
    sink.event(t, x)
  } catch (e) {
    sink.error(t, e)
  }
}

const handler = handlerSourceFactory();
const stream = newStream(handler.run);

const resSideEffect = ({req, res}: Ctx) => {
  console.log('Writing some headers'); // Remove
  res.writeHead(200);
  res.end();
}
const tapped = tap(resSideEffect, stream);
const sub = runEffects(stream, newDefaultScheduler());

// To wire up the events using micro export a handler.
const server = createServer(handler).listen(3000, '0.0.0.0', () => console.log('Server started'));