add proxy example

Author: ronag

examples/proxy/index.js

@@ -0,0 +1,47 @@
+const { Pool, Client } = require('../../')
+const http = require('http')
+const proxy = require('./proxy')
+
+const client = new Pool('http://localhost:4001', {
+  connections: 256,
+  pipelining: 1
+})
+
+Promise.all([
+  new Promise(resolve => {
+    // Proxy
+    http.createServer(async (req, res) => {
+      try {
+        await proxy({ req, res, proxyName: 'example' }, client)
+      } catch (err) {
+        if (res.headersSent) {
+          res.destroy(err)
+        } else {
+          for (const name of res.getHeaderNames()) {
+            res.removeHeader(name)
+          }
+          res.writeHead(err.statusCode || 500)
+          res.end()
+        }
+      }
+    }).listen(4000, resolve)
+  }),
+  new Promise(resolve => {
+    // Upstream
+    http.createServer((req, res) => {
+      res.end('hello world')
+    }).listen(4001, resolve)
+  })
+]).then(async () => {
+  const client = new Client('http://localhost:4000')
+  const { body } = await client.request({
+    method: 'GET',
+    path: '/'
+  })
+
+  for await (const chunk of body) {
+    console.log(String(chunk))
+  }
+})
+
+// TODO: Add websocket example.

examples/proxy/proxy.js

@@ -0,0 +1,257 @@
+const net = require('net')
+const { pipeline } = require('stream')
+const createError = require('http-errors')
+
+module.exports = async function proxy (ctx, client) {
+  const { req, socket } = ctx
+
+  if (socket) {
+    const handler = new WSHandler(ctx)
+    client.dispatch({
+      method: req.method,
+      path: req.url,
+      headers: getRequestHeaders(ctx),
+      upgrade: 'Websocket'
+    }, handler)
+    return handler.promise
+  } else {
+    const handler = new HTTPHandler(ctx)
+    client.dispatch({
+      method: req.method,
+      path: req.url,
+      headers: getRequestHeaders(ctx),
+      body: req
+    }, handler)
+    return handler.promise
+  }
+}
+
+class HTTPHandler {
+  constructor (ctx) {
+    const { req, res } = ctx
+
+    this.req = req
+    this.res = res
+    this.resume = null
+    this.abort = null
+    this.promise = new Promise((resolve, reject) => {
+      this.callback = err => err ? reject(err) : resolve()
+    })
+  }
+
+  onConnect (abort) {
+    if (this.req.aborted) {
+      abort()
+    } else {
+      this.abort = abort
+      this.res.on('close', abort)
+    }
+  }
+
+  onHeaders (statusCode, headers, resume) {
+    if (statusCode < 200) {
+      return
+    }
+
+    this.resume = resume
+    this.res.on('drain', resume)
+    writeHead(this.res, statusCode, getResponseHeaders(headers))
+  }
+
+  onData (chunk) {
+    return this.res.write(chunk)
+  }
+
+  onComplete () {
+    this.res.end()
+    this.callback()
+  }
+
+  onError (err) {
+    this.res
+      .off('drain', this.resume)
+      .off('close', this.abort)
+    this.callback(err)
+  }
+}
+
+class WSHandler {
+  constructor (ctx) {
+    const { socket, head } = ctx
+
+    setupSocket(socket)
+
+    this.socket = socket
+    this.head = head
+    this.promise = new Promise((resolve, reject) => {
+      this.callback = err => err ? reject(err) : resolve()
+    })
+  }
+
+  onConnect (abort) {
+    if (this.socket.destroyed) {
+      abort()
+    } else {
+      this.socket.on('close', abort)
+    }
+  }
+
+  onUpgrade (statusCode, headers, socket) {
+    if (this.head && this.head.length) {
+      socket.unshift(this.head)
+    }
+
+    headers = getResponseHeaders(headers)
+
+    setupSocket(socket)
+
+    let head = 'HTTP/1.1 101 Switching Protocols\r\nconnection: upgrade\r\nupgrade: websocket'
+
+    for (let n = 0; n < headers.length; n += 2) {
+      const key = headers[n + 0]
+      const val = headers[n + 1]
+
+      if (!Array.isArray(val)) {
+        head += `\r\n${key}: ${val}`
+      } else {
+        for (let i = 0; i < val.length; i++) {
+          head += `\r\n${key}: ${val[i]}`
+        }
+      }
+    }
+    head += '\r\n\r\n'
+
+    this.socket.write(head)
+
+    pipeline(socket, this.socket, socket, this.callback)
+  }
+
+  onError (err) {
+    this.callback(err)
+  }
+}
+
+const HOP_EXPR = /^(te|via|host|upgrade|trailers|forwarded|connection|keep-alive|http2-settings|transfer-encoding|proxy-connection|proxy-authenticate|proxy-authorization)$/i
+
+function getRequestHeaders (ctx) {
+  const { req, proxyName } = ctx
+  const { rawHeaders: headers, socket } = req
+
+  const result = []
+
+  let via = ''
+  let forwarded = ''
+  let host = ''
+  let authority = ''
+
+  for (let n = 0; n < headers.length; n += 2) {
+    const key = headers[n + 0]
+    const val = headers[n + 1]
+
+    if (!via && /^via$/i.test(key)) {
+      via = val
+    } else if (!forwarded && /^forwarded$/.test(key)) {
+      forwarded = val
+    } else if (!authority && /^:authority$/i.test(key)) {
+      authority = val
+    } else if (!host && /^host$/i.test(key)) {
+      host = val
+    } else if (key.charAt(0) !== ':' && !HOP_EXPR.test(key)) {
+      result.push(key, val)
+    }
+  }
+
+  // TODO(fix): <host> [ ":" <port> ] vs <pseudonym>
+  // See, https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Via.
+  if (proxyName) {
+    if (via) {
+      if (via.split(',').some(name => name.endsWith(proxyName))) {
+        throw new createError.LoopDetected()
+      }
+      via += ', '
+    }
+    via += `${req.httpVersion} ${proxyName}`
+  }
+
+  if (via) {
+    result.push('via', via)
+  }
+
+  result.push('forwarded', (forwarded ? forwarded + ', ' : '') + [
+    `by=${printIp(socket.localAddress, socket.localPort)}`,
+    `for=${printIp(socket.remoteAddress, socket.remotePort)}`,
+    `proto=${socket.encrypted ? 'https' : 'http'}`,
+    `host=${printIp(authority || host || '')}`
+  ].join(';'))
+
+  return result
+}
+
+function getResponseHeaders (headers) {
+  let remove
+  for (let n = 0; n < headers.length; n += 2) {
+    const key = headers[n + 0]
+    if (key.length === 10 && /^connection$/i.test(key)) {
+      const val = headers[n + 1]
+      if (!HOP_EXPR.test(val)) {
+        remove = val.split(',')
+      }
+      break
+    }
+  }
+
+  const result = []
+  for (let n = 0; n < headers.length; n += 2) {
+    const key = headers[n + 0]
+    const val = headers[n + 1]
+
+    if (!HOP_EXPR.test(key) && (!remove || !remove.includes(key))) {
+      result.push(key, val)
+    }
+  }
+
+  return result
+}
+
+function setupSocket (socket) {
+  socket.setTimeout(0)
+  socket.setNoDelay(true)
+  socket.setKeepAlive(true, 0)
+}
+
+function printIp (address, port) {
+  const isIPv6 = net.isIPv6(address)
+  let str = `${address}`
+  if (isIPv6) {
+    str = `[${str}]`
+  }
+  if (port) {
+    str = `${str}:${port}`
+  }
+  if (isIPv6 || port) {
+    str = `"${str}"`
+  }
+  return str
+}
+
+function writeHead (res, statusCode, headers) {
+  // TODO (perf): res.writeHead should support Array and/or string.
+  const obj = {}
+  for (var i = 0; i < headers.length; i += 2) {
+    var key = headers[i]
+    var val = obj[key]
+    if (!val) {
+      obj[key] = headers[i + 1]
+    } else {
+      if (!Array.isArray(val)) {
+        val = [val]
+        obj[key] = val
+      }
+      val.push(headers[i + 1])
+    }
+  }
+
+  res.writeHead(statusCode, obj)
+
+  return res
+}