Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 44 additions & 7 deletions packages/devframe/src/rpc/transports/ws-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import type { NodeAdapter } from 'crossws/adapters/node'
import type { Buffer } from 'node:buffer'
import type { Server as HttpServer, IncomingMessage } from 'node:http'
import type { Server as HttpsServer, ServerOptions as HttpsServerOptions } from 'node:https'
import type { AddressInfo } from 'node:net'
import type { Duplex } from 'node:stream'
import type { RpcFunctionDefinitionAny } from '../types'
import { createServer as createHttpServer } from 'node:http'
Expand Down Expand Up @@ -44,9 +45,12 @@ export interface WsRpcTransportOptions {
* this transport detaches the upgrade listener without closing the server.
*/
server?: HttpServer | HttpsServer
/** Port for a newly-created standalone WS server. */
/**
* Port for the standalone WebSocket server. Defaults to `0`, which lets the
* operating system assign an available port.
*/
port?: number
/** Host for a newly-created standalone WS server. Defaults to `localhost`. */
/** Host for the standalone WebSocket server. Defaults to `localhost`. */
host?: string
/**
* Restrict the WS endpoint to a single upgrade route (e.g. `/__devframe_ws`). When
Expand Down Expand Up @@ -168,6 +172,10 @@ export interface WsRpcTransport {
* `peers` and pub/sub. See https://crossws.h3.dev.
*/
ws: NodeAdapter
/** Resolves when the transport-owned server is listening. */
ready: Promise<void>
/** Returns the bound address, or `null` when the server is not listening. */
address: () => AddressInfo | string | null
/** Remove the upgrade listener from a shared `server` (a no-op otherwise). */
detach: () => void
/**
Expand All @@ -184,6 +192,27 @@ const EMPTY_DEFS: ReadonlyMap<string, Pick<RpcFunctionDefinitionAny, 'jsonSerial

function NOOP() {}

function listen(
server: HttpServer | HttpsServer,
port: number,
host: string,
): Promise<void> {
return new Promise((resolve, reject) => {
const onError = (error: Error) => reject(error)
server.once('error', onError)
try {
server.listen(port, host, () => {
server.off('error', onError)
resolve()
})
}
catch (error) {
server.off('error', onError)
reject(error)
}
})
}

/** Compare two URL paths ignoring a trailing slash. */
function pathMatches(a: string, b: string): boolean {
const strip = (p: string) => (p.length > 1 && p.endsWith('/') ? p.slice(0, -1) : p)
Expand Down Expand Up @@ -276,9 +305,9 @@ function routeUpgrades(
* `server` (sharing its port, optionally scoped to a `path`), or let this
* helper create a standalone server from `port` / `host` / `https`.
*
* Returns the crossws node adapter plus `detach` (remove the upgrade
* listener from a shared `server`) and `close` (full deterministic
* teardown).
* Returns the crossws node adapter, standalone-server readiness/address
* accessors, `detach` (remove the upgrade listener from a shared `server`),
* and `close` (full deterministic teardown).
*/
export function attachWsRpcTransport<
ClientFunctions extends object,
Expand Down Expand Up @@ -388,6 +417,7 @@ export function attachWsRpcTransport<
})

let detach = NOOP
let ready = Promise.resolve()
// A server created (and thus owned) by this transport. Nothing else
// handles its upgrades, so off-route clients are rejected promptly.
let ownedServer: HttpServer | HttpsServer | undefined
Expand All @@ -399,7 +429,7 @@ export function attachWsRpcTransport<
else if (https) {
ownedServer = createHttpsServer(https)
detach = routeUpgrades(ownedServer, ws, path, true, allowedOrigins)
ownedServer.listen(port, host)
ready = listen(ownedServer, port ?? 0, host)
}
else {
// Standalone server on its own port. Plain HTTP requests get the
Expand All @@ -409,11 +439,15 @@ export function attachWsRpcTransport<
res.end('Upgrade Required')
})
detach = routeUpgrades(ownedServer, ws, path, true, allowedOrigins)
ownedServer.listen(port, host)
ready = listen(ownedServer, port ?? 0, host)
}

const activeServer = server ?? ownedServer

return {
ws,
ready,
address: () => activeServer?.address() ?? null,
detach,
async close() {
// Detach our upgrade listener first so a shared host server stops
Expand All @@ -425,6 +459,9 @@ export function attachWsRpcTransport<
ws.closeAll(undefined, undefined, true)
if (ownedServer) {
const srv = ownedServer
await ready.catch(() => {})
if (!srv.listening)
return
await new Promise<void>(r => srv.close(() => r()))
}
},
Expand Down
24 changes: 24 additions & 0 deletions packages/devframe/src/rpc/transports/ws.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,30 @@ describe('ws client post on a non-open socket', () => {
})

describe('devframe rpc', () => {
it('atomically allocates unique ports for concurrent standalone transports', async () => {
const HOST = '127.0.0.1'
const transports = Array.from({ length: 3 }, () => {
const server = createRpcServer<Record<string, never>, Record<string, never>>({})
return attachWsRpcTransport(server, { host: HOST })
})

try {
await Promise.all(transports.map(transport => transport.ready))
const ports = transports.map((transport) => {
const address = transport.address()
if (!address || typeof address === 'string')
throw new TypeError('Expected an IP socket address')
return address.port
})

expect(ports.every(port => port > 0)).toBe(true)
expect(new Set(ports).size).toBe(transports.length)
}
finally {
await Promise.all(transports.map(transport => transport.close()))
}
})

it('should work w/ ws transport', async () => {
// Use 127.0.0.1 on both client and server so they agree on the
// address family — `localhost` resolution is ambiguous (IPv4 vs IPv6)
Expand Down
Loading