Bun Support Missing

[BurkeDesigns]

How do I get capnweb to work in bun? I recieve the client message, but no server side function is executed.

[BurkeDesigns]

After much experimentation (trial and error) I discovered the following:

The issue is that Bun's WebSocket object doesn't support addEventListener in the same way as a standard WebSocket, so I wrapped it to store the event listeners and manually trigger them in the message handler. As this is very messy and hacky, could you guy's fix that in the capnweb library itself to handle that automatically?

The screenshot below is how I finally got it to work in Bun.

[kentonv]

Oof, it seems like Bun has chosen a deeply non-standard and weirdly procedural-style design for their WebSocket server-side API. I guess supporting this would require implementing a custom RpcTransport just for Bun.

Or I suppose you could use Bun's node:http compat, but I imagine that's not a very palatable solution?

[BurkeDesigns]

@kentonv Yeah, that would be less than ideal to use "node:http".

[balazskosi]

Here is the skeleton how I use capnweb with bun:

const webServer = Bun.serve<BunWebSocketTransport, any>({
    port,
    routes: {
        "/connect": req => {                    
            if (webServer.upgrade(req, {data: new BunWebSocketTransport()})) {
                return;
            }
            return new Response("Upgrade failed", {status: 500});
        },                
    },
    websocket: {
        open: ws => ws.data.onOpen(ws),
        message: (ws, message: string | Buffer) => ws.data.onMessage(message),
        close: (ws, code, reason) => ws.data.onClose(code, reason)
    }
});

import type {RpcTransport} from "capnweb";
import type {ServerWebSocket} from "bun";

export default class BunWebSocketTransport implements RpcTransport {
    #websocket: ServerWebSocket<any> | undefined;
    #sendQueue : string[] | undefined = [];
    #receiveResolver?: (message: string) => void;
    #receiveRejecter?: (err: any) => void;
    #receiveQueue: string[] = [];
    #error?: any;

    onOpen(websocket: ServerWebSocket<any>): void {
        this.#websocket = websocket;
        this.#sendQueue?.forEach(message => {
            websocket.send(message);
        });
        this.#sendQueue = undefined;
    }

    onMessage(message: string | Buffer): void {
        if (this.#error) {
            // Ignore further messages.
        } else if (typeof message === "string") {
            if (this.#receiveResolver) {
                this.#receiveResolver(message);
                this.#receiveResolver = undefined;
                this.#receiveRejecter = undefined;
            } else {
                this.#receiveQueue.push(message);
            }
        } else {
            this.#receivedError(new TypeError("Received non-string message from WebSocket."));
        }
    }

    onClose(code: number, reason: string): void {
        this.#receivedError(new Error(`Peer closed WebSocket: ${code} ${reason}`));
    }


    async receive(): Promise<string> {
        if (this.#receiveQueue.length > 0) {
            return this.#receiveQueue.shift()!;
        } else if(this.#error) {
            throw this.#error;
        } else {
            return new Promise<string>(
                (resolver, rejector) => {
                    this.#receiveResolver = resolver;
                    this.#receiveRejecter = rejector
                });
        }
    }

    async send(message: string): Promise<void> {
        if (this.#sendQueue) {
            // Not open yet, queue for later.
            this.#sendQueue.push(message);
        } else {
            this.#websocket!.send(message);
        }
    }

    abort?(reason: any): void {
        let message: string;
        if (reason instanceof Error) {
            message = reason.message;
        } else {
            message = `${reason}`;
        }
        this.#websocket!.close(3000, message);

        if (!this.#error) {
            this.#error = reason;
        }
    }

    #receivedError(err: any) {
        if (!this.#error) {
            this.#error = err;
            this.#receiveRejecter?.(err);
            this.#receiveRejecter = undefined;
            this.#receiveResolver = undefined;
        }
    }
}

HTH

[grampelberg]

Here's something with a little less surface area - instead of trying to extend RpcTransport, just make something that quacks like a WebSocket:

interface RpcData {
  rpc: RpcBridge
}

export type RpcSocket = ServerWebSocket<RpcData>

export class RpcBridge extends EventTarget {
  #socket: RpcSocket | undefined

  onOpen(ws: RpcSocket): void {
    this.#socket = ws
  }

  send(...args: [string | ArrayBuffer | Uint8Array, boolean]): number {
    if (!this.#socket) throw new Error('call onOpen first')
    return this.#socket.send(...args)
  }

  close(...args: [number, string]): void {
    if (!this.#socket) throw new Error('call onOpen first')
    this.#socket.close(...args)
  }
}

export const upgrade = (req: Request, server: Server): boolean =>
  server.upgrade(req, { data: { rpc: new RpcBridge() } })

export const server = {
  message(ws: RpcSocket, msg: Buffer<ArrayBufferLike>): void {
    ws.data.rpc.dispatchEvent(new MessageEvent('message', { data: msg }))
  },

  open(ws: RpcSocket): void {
    ws.data.rpc.onOpen(ws)

    newWebSocketRpcSession(ws.data.rpc as unknown as WebSocket, new Relay())
  },

  close(ws: RpcSocket, code: number, reason: string): void {
    ws.data.rpc.dispatchEvent(new CloseEvent('close', { code, reason }))
  },
}