uws_http_express.js

/*
"express": "4.14.0",
"uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.6.0"

# uws backend
curl -v http://localhost:3030
curl -v -d 'hello from curl' -X POST http://localhost:3030
websocat3 ws://localhost:3030/echo


# http backend
curl -v http://localhost:3031
curl -v -d 'hello from curl' -X POST http://localhost:3031
*/

const app = require("express")();
const server = require('uWebSockets.js').App();
const Readable = require('stream').Readable;

var log = ()=>{};
var log = console.log;

app.get('/', (req, res) => {
  log('get called', req.connection.remoteAddress);
  res.status(200).send('hello world get');
})

app.post('/', (req, res) => {
  log('post called');
  req.on('data', e => log('on post =>', e.toString()));
  req.on('end', e => res.status(200).send('hello world post'));
})  

function aborted(){
  log('uws onAborted');
  this.res.aborted = this.req.aborted = true;
}

function get_header(k) {
  return this.headers[k];
}

function set_header(k, v){
  if (k === "Set-Cookie") v = typeof v == "object" ? v[0] : v;
  if (this.headers[k] == v) return;
  this.headers[k] = v;
}

function _read(a){
  this.__stream_started = true;
  var chunk = this.req_body_arr.shift();
  if(chunk) this.push(chunk);
  else if(this.__stream_ended) this.push(null);
}

function on_data(chunk, isLast){
  if(chunk.byteLength){
    chunk = Buffer.from(chunk.slice(0));
    if(this.__stream_started && this.req_body_arr.length == 0) this.push(chunk);
    else this.req_body_arr.push(chunk);
  }
  if(isLast){
    this.__stream_ended = true;
    if(this.__stream_started && this.req_body_arr.length == 0) this.push(null);
  }
}

function end(chunk) {
  if(this.statusCode) this.writeStatus(this.statusCode.toString());
  delete this.headers['Content-Length'];
  Object.keys(this.headers).forEach(k => this.writeHeader(k, this.headers[k]));
  this.chunkArr.forEach(this._write.bind(this));
  this.aborted = true;
  return this._end(chunk);
}

function removeHeader(k) {
  delete this.headers[k];
}

function write(chunk) {
  this.chunkArr.push(chunk);
}

const parseParams = (querystring) => {
  const params = new URLSearchParams(querystring);
  const obj = {};
  for (const key of params.keys()) {
    if (params.getAll(key).length > 1) obj[key] = params.getAll(key);
    else obj[key] = params.get(key);
  }
  return obj;
};

function wrap_and_serve(res, req){
  var req_stream = new Readable();
  req_stream.headers = {};
  req_stream._read = _read;
  req_stream.req_body_arr = [];
  req_stream.url = req.getUrl();
  req_stream.method = req.getMethod().toUpperCase();
  req_stream.query = parseParams(req.getQuery());
  req_stream.connection = {remoteAddress: Buffer.from(res.getRemoteAddressAsText()).toString()};

  res.headers = {};
  res.chunkArr = [];
  res._end = res.end;
  res._write = res.write;

  res.end = end;
  res.write = write;
  res.setHeader = set_header;
  res.getHeader = get_header;
  res.writeHeader = res.writeHeader;
  res.writeStatus = res.writeStatus;
  res.removeHeader = removeHeader;

  res.onAborted(aborted.bind({req, res}));
  res.onData(on_data.bind(req_stream));
  req.forEach((k, v) => req_stream.headers[k] = v);

  app(req_stream, res); return;
  // setTimeout(e => app(req_stream, res), 100);
}

server.any('/*', wrap_and_serve);

const HTTP_PORT = 3030;
const HTTP_HOST = '0.0.0.0';

app.listen(HTTP_PORT+1);
server.listen(HTTP_HOST, HTTP_PORT, function(success){
  log(success ? 'Listening to port' : 'Failed to listen on port', HTTP_PORT);
});

/*ws begin*/
function uws_on_upgrade(res, req, context){
  var handle = {url: req.getUrl(), headers: {}};
  req.forEach((k, v) => k.indexOf('sec-websocket') == -1 ? handle.headers[k] = v : undefined);
  res.upgrade({hdl: handle}, req.getHeader('sec-websocket-key'), req.getHeader('sec-websocket-protocol'),
    req.getHeader('sec-websocket-extensions'), context
  );  
}

function uws_on_open(ws){
  ws.hdl._ws = ws;
  this.onopen(ws.hdl);
}

function uws_on_message(ws, msg, is_binary){
  this.onmessage(ws.hdl, is_binary ? msg : Buffer.from(msg).toString());
}

function uws_on_close(ws, code, msg){
  this.onclose(ws.hdl, code, Buffer.from(msg).toString())
}

function uws_send(hdl, msg){
  return hdl._ws.send(msg, typeof msg != 'string');
}

function uws_close(hdl, code, msg){
  hdl._ws.end(code, msg);
}

function reg_wss(opts){
  server.ws(opts.path, {
    upgrade: uws_on_upgrade,
    open: uws_on_open.bind(opts),
    message: uws_on_message.bind(opts),
    close: uws_on_close.bind(opts),
  });
  return {send: uws_send, close: uws_close};
}

const ws_funcs = reg_wss({
  path: '/echo',
  onopen: (ws) => log('ws on open', ws),
  onmessage: (ws, msg) => ws_funcs.send(ws, msg) && log('ws on message', msg),
  onclose: (ws, code, msg) => log('ws on close', code, msg),
});
/*ws_end*/