feat(hono-websocket): add Hono WebSocket adapter support

Add comprehensive Hono WebSocket integration to enable kkrpc usage with Hono framework. Includes new adapter implementation with manual message handling, support for multiple runtime environments (Bun, Deno, Cloudflare Workers), and complete test coverage covering RPC calls, concurrent connections, property access, and error handling.

Author: HuakunShen

package.json

@@ -21,7 +21,7 @@
 		"turbo": "^2.4.4",
 		"typescript": "5.8.2"
 	},
-	"packageManager": "pnpm@10.19.0",
+	"packageManager": "pnpm@10.20.0",
 	"engines": {
 		"node": ">=18"
 	},

packages/kkrpc/__tests__/hono-websocket.test.ts

@@ -0,0 +1,144 @@
+import { afterAll, beforeAll, expect, test } from "bun:test"
+import { Hono } from "hono"
+import { upgradeWebSocket, websocket } from "hono/bun"
+import { RPCChannel, WebSocketClientIO, createHonoWebSocketHandler } from "../mod.ts"
+import type { DestroyableIoInterface } from "../src/interface.ts"
+import { apiMethods, type API } from "./scripts/api.ts"
+
+const PORT = 3002
+let server: ReturnType<typeof Bun.serve> | null = null
+
+beforeAll(() => {
+	// Create Hono app with WebSocket support
+	const app = new Hono()
+
+	app.get("/ws", upgradeWebSocket(() => {
+		return createHonoWebSocketHandler<API>({
+			expose: apiMethods
+		})
+	}))
+
+	// Start server
+	server = Bun.serve({
+		port: PORT,
+		fetch: app.fetch,
+		websocket
+	})
+})
+
+afterAll(() => {
+	if (server) {
+		server.stop()
+	}
+})
+
+test("Hono WebSocket RPC", async () => {
+	const clientIO = new WebSocketClientIO({
+		url: `ws://localhost:${PORT}/ws`
+	})
+
+	const clientRPC = new RPCChannel<API, API, DestroyableIoInterface>(clientIO, {
+		expose: apiMethods
+	})
+	const api = clientRPC.getAPI()
+
+	// Test individual calls
+	const sum = await api.add(5, 3)
+	expect(sum).toBe(8)
+
+	const product = await api.math.grade2.multiply(4, 6)
+	expect(product).toBe(24)
+
+	// Test concurrent calls
+	const results = await Promise.all([
+		api.add(10, 20),
+		api.math.grade2.multiply(10, 20),
+		api.add(30, 40),
+		api.math.grade2.multiply(30, 40)
+	])
+
+	expect(results).toEqual([30, 200, 70, 1200])
+
+	// Test multiple random calls
+	for (let i = 0; i < 50; i++) {
+		const a = Math.floor(Math.random() * 100)
+		const b = Math.floor(Math.random() * 100)
+
+		const sum = await api.add(a, b)
+		expect(sum).toBe(a + b)
+
+		const product = await api.math.grade2.multiply(a, b)
+		expect(product).toBe(a * b)
+	}
+
+	clientIO.destroy()
+})
+
+test("Hono WebSocket concurrent connections", async () => {
+	const numClients = 5
+	const clients = Array.from({ length: numClients }, () => {
+		const clientIO = new WebSocketClientIO({
+			url: `ws://localhost:${PORT}/ws`
+		})
+		return {
+			io: clientIO,
+			rpc: new RPCChannel<{}, API, DestroyableIoInterface>(clientIO)
+		}
+	})
+
+	try {
+		// Test concurrent calls from multiple clients
+		const results = await Promise.all(
+			clients.flatMap(({ rpc }) => {
+				const api = rpc.getAPI()
+				return [api.add(10, 20), api.math.grade2.multiply(10, 20)]
+			})
+		)
+
+		// Verify results
+		for (let i = 0; i < results.length; i += 2) {
+			expect(results[i]).toBe(30) // add result
+			expect(results[i + 1]).toBe(200) // multiply result
+		}
+	} finally {
+		// Cleanup
+		clients.forEach(({ io }) => io.destroy())
+	}
+})
+
+test("Hono WebSocket property access", async () => {
+	const clientIO = new WebSocketClientIO({
+		url: `ws://localhost:${PORT}/ws`
+	})
+
+	const clientRPC = new RPCChannel<{}, API, DestroyableIoInterface>(clientIO)
+	const api = clientRPC.getAPI()
+
+	// Test property access
+	const counter = await api.counter
+	expect(counter).toBe(42)
+
+	const nestedValue = await api.nested.value
+	expect(nestedValue).toBe("hello world")
+
+	const deepProp = await api.nested.deepObj.prop
+	expect(deepProp).toBe(true)
+
+	clientIO.destroy()
+})
+
+test("Hono WebSocket error handling", async () => {
+	const clientIO = new WebSocketClientIO({
+		url: `ws://localhost:${PORT}/ws`
+	})
+
+	const clientRPC = new RPCChannel<{}, API, DestroyableIoInterface>(clientIO)
+	const api = clientRPC.getAPI()
+
+	// Test error throwing
+	await expect(api.throwSimpleError()).rejects.toThrow("This is a simple error")
+
+	await expect(api.throwCustomError()).rejects.toThrow("This is a custom error")
+
+	clientIO.destroy()
+})

packages/kkrpc/mod.ts

@@ -10,6 +10,7 @@
  * - deno
  * - websocket
  * - http
+ * - hono-websocket
  * - RPC Channel
  * - serialization
  *
@@ -21,6 +22,7 @@ export * from "./src/adapters/node.ts"
 export * from "./src/adapters/websocket.ts"
 export * from "./src/adapters/http.ts"
 export * from "./src/adapters/tauri.ts"
+export * from "./src/adapters/hono-websocket.ts"
 export * from "./src/interface.ts"
 export * from "./src/channel.ts"
 export * from "./src/utils.ts"

packages/kkrpc/package.json

@@ -114,14 +114,18 @@
 	"peerDependencies": {
 		"typescript": "^5.0.0",
 		"socket.io": "^4.0.0",
-		"socket.io-client": "^4.0.0"
+		"socket.io-client": "^4.0.0",
+		"hono": "^4.0.0"
 	},
 	"peerDependenciesMeta": {
 		"socket.io": {
 			"optional": true
 		},
 		"socket.io-client": {
 			"optional": true
+		},
+		"hono": {
+			"optional": true
 		}
 	},
 	"dependencies": {

packages/kkrpc/src/adapters/hono-websocket.ts

@@ -0,0 +1,176 @@
+import type { DestroyableIoInterface, IoCapabilities, IoMessage } from "../interface.ts"
+import { RPCChannel } from "../channel.ts"
+
+/**
+ * Options for creating a Hono WebSocket handler
+ */
+export interface HonoWebSocketOptions<API extends Record<string, any>> {
+	/** The API implementation to expose on the server */
+	expose: API
+	/** Optional serialization options */
+	serialization?: {
+		version: "json" | "superjson"
+	}
+}
+
+/**
+ * WebSocket IO adapter specifically for Hono that processes messages manually
+ * This is needed because Hono handles messages through callbacks rather than native onmessage
+ */
+class HonoWebSocketIO implements DestroyableIoInterface {
+	name = "hono-websocket-io"
+	private messageQueue: string[] = []
+	private resolveRead: ((value: string | null) => void) | null = null
+	capabilities: IoCapabilities = {
+		structuredClone: false,
+		transfer: false
+	}
+
+	constructor(private ws: WebSocket) {}
+
+	/**
+	 * Manually feed a message from Hono's onMessage callback
+	 */
+	feedMessage(message: string): void {
+		const DESTROY_SIGNAL = "__DESTROY__"
+		
+		if (message === DESTROY_SIGNAL) {
+			this.destroy()
+			return
+		}
+
+		if (this.resolveRead) {
+			this.resolveRead(message)
+			this.resolveRead = null
+		} else {
+			this.messageQueue.push(message)
+		}
+	}
+
+	async read(): Promise<string | null> {
+		if (this.messageQueue.length > 0) {
+			return this.messageQueue.shift() ?? null
+		}
+
+		return new Promise((resolve) => {
+			this.resolveRead = resolve
+		})
+	}
+
+	async write(message: string | IoMessage): Promise<void> {
+		if (typeof message !== "string") {
+			throw new Error("HonoWebSocketIO only supports string messages")
+		}
+		this.ws.send(message)
+	}
+
+	destroy(): void {
+		this.ws.close()
+	}
+
+	signalDestroy(): void {
+		this.write("__DESTROY__")
+	}
+}
+
+/**
+ * Creates a Hono WebSocket handler that integrates kkrpc with Hono's upgradeWebSocket
+ * 
+ * This function works with Hono's upgradeWebSocket from:
+ * - hono/bun
+ * - hono/deno  
+ * - hono/cloudflare-workers
+ * 
+ * @example
+ * ```ts
+ * // Bun example
+ * import { Hono } from 'hono'
+ * import { upgradeWebSocket, websocket } from 'hono/bun'
+ * import { createHonoWebSocketHandler } from 'kkrpc'
+ * 
+ * const app = new Hono()
+ * 
+ * app.get('/ws', upgradeWebSocket(() => {
+ *   return createHonoWebSocketHandler({
+ *     expose: myAPI
+ *   })
+ * }))
+ * 
+ * Bun.serve({
+ *   fetch: app.fetch,
+ *   websocket
+ * })
+ * ```
+ * 
+ * @example
+ * ```ts
+ * // Deno example
+ * import { Hono } from 'hono'
+ * import { upgradeWebSocket } from 'hono/deno'
+ * import { createHonoWebSocketHandler } from 'kkrpc'
+ * 
+ * const app = new Hono()
+ * 
+ * app.get('/ws', upgradeWebSocket(() => {
+ *   return createHonoWebSocketHandler({
+ *     expose: myAPI
+ *   })
+ * }))
+ * 
+ * Deno.serve({ fetch: app.fetch, port: 8000 })
+ * ```
+ */
+export function createHonoWebSocketHandler<API extends Record<string, any>>(
+	options: HonoWebSocketOptions<API>
+): {
+	onMessage(event: MessageEvent, ws: any): void
+	onClose(): void
+	onError?(event: Event, ws: any): void
+	onOpen?(event: Event, ws: any): void
+} {
+	let serverIO: HonoWebSocketIO | null = null
+	let rpc: RPCChannel<API, API> | null = null
+
+	return {
+		onOpen(_event: Event, ws: any) {
+			// Create the IO adapter and RPC channel when connection opens
+			// Hono passes different WebSocket types depending on runtime (WebSocket or WSContext)
+			// Extract the actual WebSocket if it's wrapped
+			const actualWs = (ws as any).raw || ws
+			serverIO = new HonoWebSocketIO(actualWs)
+			rpc = new RPCChannel<API, API>(serverIO, {
+				expose: options.expose,
+				serialization: options.serialization
+			})
+		},
+		onMessage(event: MessageEvent, _ws: any) {
+			// Convert message to string if needed
+			let message = event.data
+			if (typeof message === "object" && message !== null && "toString" in message) {
+				message = message.toString("utf-8")
+			} else if (typeof message !== "string") {
+				message = String(message)
+			}
+
+			// Feed the message to the IO adapter for processing
+			if (serverIO) {
+				serverIO.feedMessage(message)
+			}
+		},
+		onClose() {
+			if (serverIO) {
+				serverIO.destroy()
+				serverIO = null
+				rpc = null
+			}
+		},
+		onError(event: Event, _ws: any) {
+			console.error("Hono WebSocket error:", event)
+			if (serverIO) {
+				serverIO.destroy()
+				serverIO = null
+				rpc = null
+			}
+		}
+	}
+}