add proxy example

Author: ronag

examples/proxy/index.js

@@ -0,0 +1,37 @@
+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((req, res) => {
+      proxy({ req, res }, client)
+    }).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))
+  }
+})
+
+http.createServer((req, res) => {
+  res.end('hello world')
+}).listen(3001)

examples/proxy/proxy.js

@@ -0,0 +1,237 @@
+const net = require('net')
+const { pipeline } = require('stream')
+
+module.exports = async function proxy (ctx, client) {
+  const { req, socket } = ctx
+
+  // TODO: requestTimeout?
+
+  if (socket) {
+    const handler = new WSHandler(ctx)
+    client.dispatch({
+      method: req.method,
+      path: req.originalUrl || req.url,
+      headers: getRequestHeaders(ctx),
+      upgrade: 'Websocket'
+    }, handler)
+    return handler.promise
+  } else {
+    const handler = new HTTPHandler(ctx)
+    client.dispatch({
+      method: req.method,
+      path: req.originalUrl || req.url,
+      headers: getRequestHeaders(ctx),
+      body: req
+    }, handler)
+    return handler.promise
+  }
+}
+
+class WSHandler {
+  constructor ({ socket, head }) {
+    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)
+  }
+}
+
+class HTTPHandler {
+  constructor ({ req, res }) {
+    this.req = req
+    this.res = res
+    this.promise = new Promise((resolve, reject) => {
+      this.callback = err => err ? reject(err) : resolve()
+    })
+  }
+
+  onConnect (abort) {
+    if (this.req.aborted) {
+      abort()
+    } else {
+      this.res.on('close', abort)
+    }
+  }
+
+  onHeaders (statusCode, headers, resume) {
+    if (statusCode < 200) {
+      return
+    }
+
+    this.res.on('drain', resume)
+
+    headers = getResponseHeaders(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])
+      }
+    }
+
+    this.res.writeHead(statusCode, obj)
+  }
+
+  onData (chunk) {
+    return this.res.write(chunk)
+  }
+
+  onComplete () {
+    this.res.end()
+    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 ({ req, proxyName }) {
+  const result = []
+
+  for (const [key, value] of Object.entries(req.headers)) {
+    if (key.charAt(0) !== ':' && !HOP_EXPR.test(key)) {
+      result.push(key, value)
+    }
+  }
+
+  // TODO(fix): <host> [ ":" <port> ] vs <pseudonym>
+  // See, https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Via.
+  let via = req.headers.via || ''
+  if (proxyName) {
+    if (via) {
+      for (const name of via.split(',')) {
+        if (name.endsWith(proxyName)) {
+          const err = new Error('loop detected')
+          err.statusCode = 508
+          throw err
+        }
+      }
+      via += ','
+    }
+
+    result.push('via', `${via}${req.httpVersion} ${proxyName}`)
+  } else if (via) {
+    result.push('via', via)
+  }
+
+  const forwarded = [
+    `by=${printIp(req.socket.localAddress, req.socket.localPort)}`,
+    `for=${printIp(req.socket.remoteAddress, req.socket.remotePort)}`,
+    `proto=${req.socket.encrypted ? 'https' : 'http'}`,
+    `host=${printIp(req.headers[':authority'] || req.headers.host || '')}`
+  ].join(';')
+
+  if (req.headers.forwarded) {
+    result.push('forwarded', `, ${forwarded}`)
+  } else {
+    result.push('forwarded', `${forwarded}`)
+  }
+
+  return result
+}
+
+function getResponseHeaders (headers) {
+  let result = []
+  let remove
+  for (let n = 0; n < headers.length; n += 2) {
+    const key = headers[n + 0]
+    if (/^connection$/i.test(key)) {
+      const val = headers[n + 1]
+      if (/^(connection|keep-alive|upgrade)$/i.test(val)) {
+        remove = val.split(',')
+      }
+    }
+  }
+
+  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))) {
+      if (!result) {
+        result = []
+      }
+      result.push(key, val)
+    }
+  }
+
+  return result || headers
+}
+
+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
+}