feat: first class support for websockets

README.md

@@ -71,6 +71,34 @@ serve(app, (info) => {
 })
 ```
 
+## WebSocket
+
+You can upgrade WebSocket connections with `upgradeWebSocket` from `@hono/node-server`.
+To enable this, install `ws` (and `@types/ws`) in your project, then create and provide a `WebSocketServer` as shown in the example below.
+
+```ts
+import { serve, upgradeWebSocket } from '@hono/node-server'
+import { WebsocketServer } from 'ws'
+import { Hono } from 'hono'
+
+const app = new Hono()
+
+app.get(
+  '/ws',
+  upgradeWebSocket(() => ({
+    onMessage(event, ws) {
+      ws.send(event.data)
+    },
+  }))
+)
+
+const wss = new WebSocketServer({ noServer: true }) // important to create with `noServer: true`
+serve({
+  fetch: app.fetch,
+  websocket: { server: wss },
+})
+```
+
 For example, run it using `ts-node`. Then an HTTP server will be launched. The default port is `3000`.
 
 ```sh
@@ -132,6 +160,23 @@ serve({
 })
 ```
 
+### `websocket`
+
+provide a websocket server to enable websocket support.
+
+```ts
+import { serve, upgradeWebSocket } from '@hono/node-server'
+import { WebsocketServer } from 'ws'
+
+// ...
+const wss = new WebSocketServer({ noServer: true })
+
+serve({
+  fetch: app.fetch,
+  websocket: { server: wss },
+})
+```
+
 ## Middleware
 
 Most built-in middleware also works with Node.js.

package.json

@@ -83,6 +83,7 @@
     "@hono/eslint-config": "^1.0.1",
     "@types/node": "^20.10.0",
     "@types/supertest": "^2.0.12",
+    "@types/ws": "^8.18.1",
     "@whatwg-node/fetch": "^0.9.14",
     "eslint": "^9.10.0",
     "hono": "^4.12.8",
@@ -92,7 +93,8 @@
     "supertest": "^6.3.3",
     "tsdown": "^0.20.3",
     "typescript": "^5.3.2",
-    "vitest": "^4.0.18"
+    "vitest": "^4.0.18",
+    "ws": "^8.19.0"
   },
   "peerDependencies": {
     "hono": "^4"

src/index.ts

@@ -1,4 +1,5 @@
 export { serve, createAdaptorServer } from './server'
+export { upgradeWebSocket } from './websocket'
 export { getRequestListener } from './listener'
 export { RequestError } from './request'
 export type { HttpBindings, Http2Bindings, ServerType } from './types'

src/server.ts

@@ -2,6 +2,7 @@ import { createServer as createServerHTTP } from 'node:http'
 import type { AddressInfo } from 'node:net'
 import { getRequestListener } from './listener'
 import type { Options, ServerType } from './types'
+import { setupWebSocket } from './websocket'
 
 export const createAdaptorServer = (options: Options): ServerType => {
   const fetchCallback = options.fetch
@@ -14,6 +15,11 @@ export const createAdaptorServer = (options: Options): ServerType => {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   const createServer: any = options.createServer || createServerHTTP
   const server: ServerType = createServer(options.serverOptions || {}, requestListener)
+  if (options.websocket && options.websocket.server) {
+    if (options.websocket.server.options.noServer !== true)
+      throw new Error('WebSocket server must be created with { noServer: true } option')
+    setupWebSocket({ server, fetchCallback, wss: options.websocket.server })
+  }
   return server
 }
 

src/types.ts

@@ -1,3 +1,4 @@
+import type { WebSocketServer } from 'ws'
 import type {
   createServer,
   IncomingMessage,
@@ -73,6 +74,9 @@ export type Options = {
   autoCleanupIncoming?: boolean
   port?: number
   hostname?: string
+  websocket?: {
+    server: WebSocketServer
+  }
 } & ServerOptions
 
 export type CustomErrorHandler = (err: unknown) => void | Response | Promise<void | Response>

src/url.ts

@@ -41,7 +41,6 @@ for (let c = 0x61; c <= 0x7a; c++) {
   allowedRequestUrlChar[c] = 1
 }
 ;(() => {
-  // eslint-disable-next-line quotes
   const chars = "-./:?#[]@!$&'()*+,;=~_"
   for (let i = 0; i < chars.length; i++) {
     allowedRequestUrlChar[chars.charCodeAt(i)] = 1

src/websocket.ts

@@ -0,0 +1,259 @@
+import type { UpgradeWebSocket, WSContext } from 'hono/ws'
+import { defineWebSocketHelper } from 'hono/ws'
+import type { RawData, WebSocket, WebSocketServer } from 'ws'
+import type { IncomingMessage } from 'node:http'
+import { STATUS_CODES } from 'node:http'
+import type { Duplex } from 'node:stream'
+import type { FetchCallback, ServerType } from './types'
+
+interface CloseEventInit extends EventInit {
+  code?: number
+  reason?: string
+  wasClean?: boolean
+}
+
+/**
+ * @link https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
+ */
+export const CloseEvent: typeof globalThis.CloseEvent =
+  globalThis.CloseEvent ??
+  class extends Event {
+    #eventInitDict
+
+    constructor(type: string, eventInitDict: CloseEventInit = {}) {
+      super(type, eventInitDict)
+      this.#eventInitDict = eventInitDict
+    }
+
+    get wasClean(): boolean {
+      return this.#eventInitDict.wasClean ?? false
+    }
+
+    get code(): number {
+      return this.#eventInitDict.code ?? 0
+    }
+
+    get reason(): string {
+      return this.#eventInitDict.reason ?? ''
+    }
+  }
+
+const generateConnectionSymbol = () => Symbol('connection')
+
+type WaitForWebSocket = (request: IncomingMessage, connectionSymbol: symbol) => Promise<WebSocket>
+
+const CONNECTION_SYMBOL_KEY: unique symbol = Symbol('CONNECTION_SYMBOL_KEY')
+const WAIT_FOR_WEBSOCKET_SYMBOL: unique symbol = Symbol('WAIT_FOR_WEBSOCKET_SYMBOL')
+
+export type UpgradeBindings = {
+  incoming: IncomingMessage
+  outgoing: undefined
+  wss: WebSocketServer
+  [CONNECTION_SYMBOL_KEY]?: symbol
+  [WAIT_FOR_WEBSOCKET_SYMBOL]?: WaitForWebSocket
+}
+
+type UpgradeWebSocketOptions = {
+  onError: (err: unknown) => void
+}
+
+const rejectUpgradeRequest = (socket: Duplex, status: number) => {
+  socket.end(
+    `HTTP/1.1 ${status.toString()} ${STATUS_CODES[status] ?? ''}\r\n` +
+      'Connection: close\r\n' +
+      'Content-Length: 0\r\n' +
+      '\r\n'
+  )
+}
+
+const createUpgradeRequest = (request: IncomingMessage): Request => {
+  const protocol = (request.socket as { encrypted?: boolean }).encrypted ? 'https' : 'http'
+  const url = new URL(request.url ?? '/', `${protocol}://${request.headers.host ?? 'localhost'}`)
+  const headers = new Headers()
+  for (const key in request.headers) {
+    const value = request.headers[key]
+    if (!value) {
+      continue
+    }
+    headers.append(key, Array.isArray(value) ? value[0] : value)
+  }
+  return new Request(url, {
+    headers,
+  })
+}
+
+export const setupWebSocket = (options: {
+  server: ServerType
+  fetchCallback: FetchCallback
+  wss: WebSocketServer
+}): void => {
+  const { server, fetchCallback, wss } = options
+
+  const waiterMap = new Map<
+    IncomingMessage,
+    { resolve: (ws: WebSocket) => void; connectionSymbol: symbol }
+  >()
+
+  wss.on('connection', (ws, request) => {
+    const waiter = waiterMap.get(request)
+    if (waiter) {
+      waiter.resolve(ws)
+      waiterMap.delete(request)
+    }
+  })
+
+  const waitForWebSocket: WaitForWebSocket = (request, connectionSymbol) => {
+    return new Promise<WebSocket>((resolve) => {
+      waiterMap.set(request, { resolve, connectionSymbol })
+    })
+  }
+
+  server.on('upgrade', async (request, socket: Duplex, head) => {
+    if (request.headers.upgrade?.toLowerCase() !== 'websocket') {
+      return
+    }
+
+    const env: UpgradeBindings = {
+      incoming: request,
+      outgoing: undefined,
+      wss,
+      [WAIT_FOR_WEBSOCKET_SYMBOL]: waitForWebSocket,
+    }
+
+    let status = 400
+    try {
+      const response = (await fetchCallback(
+        createUpgradeRequest(request),
+        env as unknown as Parameters<FetchCallback>[1]
+      )) as Response
+      if (response instanceof Response) {
+        status = response.status
+      }
+    } catch {
+      if (server.listenerCount('upgrade') === 1) {
+        rejectUpgradeRequest(socket, 500)
+      }
+      return
+    }
+
+    const waiter = waiterMap.get(request)
+
+    if (!waiter || waiter.connectionSymbol !== env[CONNECTION_SYMBOL_KEY]) {
+      waiterMap.delete(request)
+      if (server.listenerCount('upgrade') === 1) {
+        rejectUpgradeRequest(socket, status)
+      }
+      return
+    }
+
+    wss.handleUpgrade(request, socket, head, (ws) => {
+      wss.emit('connection', ws, request)
+    })
+  })
+
+  server.on('close', () => {
+    wss.close()
+  })
+}
+
+export const upgradeWebSocket: UpgradeWebSocket<WebSocket, UpgradeWebSocketOptions> =
+  defineWebSocketHelper(async (c, events, options) => {
+    if (c.req.header('upgrade')?.toLowerCase() !== 'websocket') {
+      return
+    }
+
+    const env = c.env as UpgradeBindings
+    const waitForWebSocket = env[WAIT_FOR_WEBSOCKET_SYMBOL]
+
+    if (!waitForWebSocket || !env.incoming) {
+      return new Response(null, { status: 500 })
+    }
+
+    const connectionSymbol = generateConnectionSymbol()
+    env[CONNECTION_SYMBOL_KEY] = connectionSymbol
+    ;(async () => {
+      const ws = await waitForWebSocket(env.incoming, connectionSymbol)
+
+      const messagesReceivedInStarting: [data: RawData, isBinary: boolean][] = []
+      const bufferMessage = (data: RawData, isBinary: boolean) => {
+        messagesReceivedInStarting.push([data, isBinary])
+      }
+      ws.on('message', bufferMessage)
+
+      const ctx: WSContext<WebSocket> = {
+        binaryType: 'arraybuffer',
+        close(code, reason) {
+          ws.close(code, reason)
+        },
+        protocol: ws.protocol,
+        raw: ws,
+        get readyState() {
+          return ws.readyState
+        },
+        send(source, opts) {
+          ws.send(source, {
+            compress: opts?.compress,
+          })
+        },
+        url: new URL(c.req.url),
+      }
+
+      try {
+        events?.onOpen?.(new Event('open'), ctx)
+      } catch (e) {
+        ;(options?.onError ?? console.error)(e)
+      }
+
+      const handleMessage = (data: RawData, isBinary: boolean) => {
+        const datas = Array.isArray(data) ? data : [data]
+        for (const data of datas) {
+          try {
+            events?.onMessage?.(
+              new MessageEvent('message', {
+                data: isBinary
+                  ? data instanceof ArrayBuffer
+                    ? data
+                    : data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength)
+                  : data.toString('utf-8'),
+              }),
+              ctx
+            )
+          } catch (e) {
+            ;(options?.onError ?? console.error)(e)
+          }
+        }
+      }
+
+      ws.off('message', bufferMessage)
+      for (const message of messagesReceivedInStarting) {
+        handleMessage(...message)
+      }
+
+      ws.on('message', (data, isBinary) => {
+        handleMessage(data, isBinary)
+      })
+
+      ws.on('close', (code, reason) => {
+        try {
+          events?.onClose?.(new CloseEvent('close', { code, reason: reason.toString() }), ctx)
+        } catch (e) {
+          ;(options?.onError ?? console.error)(e)
+        }
+      })
+
+      ws.on('error', (error) => {
+        try {
+          events?.onError?.(
+            new ErrorEvent('error', {
+              error,
+            }),
+            ctx
+          )
+        } catch (e) {
+          ;(options?.onError ?? console.error)(e)
+        }
+      })
+    })()
+
+    return new Response()
+  })