pubsub / live data pattern

[jim-alexander]

Hey,

Curious what it looks like using this tool to setup a 'live data' type connection. Its mentioned that "Supports bidirectional calling. The client can call the server, and the server can also call the client." I figured this would be a simple implementation given the websocket implementation but I wasn't able to get it hooked up / not sure its possible without polling?

Massive likelyhood I've not wired it up correctly.

Added the full reproduction example for reference but here is the gist of it:

Server

class LiveCounter extends RpcTarget implements CounterServer {
    private count = 0;
    private clients: CounterClient[] = [];

    async get() {
        return this.count;
    }

    async increment() {
        this.count++;
        console.log('[SERVER] Increment', this.count);
        this.notify();
    }

    async notify() {
        this.clients.forEach((client) => {
            try {
                console.log('[SERVER] Notifying client', this.count);
                client.onUpdate(this.count);
            } catch (err) {
                console.error('[SERVER] Error notifying client', err);
            }
        });
    }

    async subscribe(client: CounterClient) {
        this.clients.push(client);
    }
}

Client

import { newWebSocketRpcSession, RpcTarget } from 'capnweb';
import { CounterClient, CounterServer } from './types.js';

class CounterClientImpl extends RpcTarget implements CounterClient {
    private currentValue = 0;

    async onUpdate(newVal: number): Promise<void> {
        this.currentValue = newVal;
        console.log(`[CLIENT] updated: ${newVal}`);
    }

    async getCurrentValue(): Promise<number> {
        return this.currentValue;
    }
}

const client = new CounterClientImpl();

async function main() {
    console.log("[CLIENT] Connecting to Cap'n Web counter server...");

    try {
        // Connect to the WebSocket server
        const api = newWebSocketRpcSession<CounterServer>('ws://localhost:8080');

        await api.subscribe(client);

        await api.increment();
        await api.increment();
        await api.increment();

        const valueFromServer = await api.get();
        console.log(`[CLIENT] re-fetch: ${valueFromServer}`);

        const currentValue = await client.getCurrentValue();
        console.log(`[CLIENT] cached: ${currentValue}`);
    } catch (error) {
        console.error('Error:', error);
        process.exit(1);
    }
}

main().catch(console.error);

Types

export interface CounterClient {
    onUpdate(newVal: number): void;
}

export interface CounterServer {
    get(): Promise<number>;
    increment(): Promise<void>;
    subscribe(client: CounterClient): void;
    notify(): Promise<void>;
}

pubsub.zip

I understand if this is outside the scope of this dependancy, thanks really liking it so far 🙂

[kentonv]

I think there's a bug here:

    subscribe(client: CounterClient) {
        client.onUpdate(this.count);
    }

Don't you need to do this.clients.push(client)?

BTW you should also probably use client.onRpcBroken() to detect when the client has disconnected and remove it from the list.

[jim-alexander]

Ahh good catch silly mistake, that being said I updated the code but still the same issue persists. I've also tried passing a callback function which the server should be able to trigger but thats not working or throwing errors either 🤔

Server

import http from 'node:http';
import { WebSocketServer } from 'ws';
import { RpcTarget, newWebSocketRpcSession, nodeHttpBatchRpcResponse } from 'capnweb';

import { CounterServer } from './types';

class LiveCounter extends RpcTarget implements CounterServer {
    private count = 0;
    private callbacks: ((count: number) => void)[] = [];

    async get() {
        return this.count;
    }

    async increment() {
        this.count++;
        console.log('[SERVER] Increment', this.count);
        this.notify();
    }

    async notify() {
        this.callbacks.forEach((callback) => {
            try {
                console.log('[SERVER] Notifying client', this.count);
                callback(this.count);
            } catch (err) {
                console.error('[SERVER] Error notifying client', err);
            }
        });
    }

    async subscribe(callback: (count: number) => void) {
        this.callbacks.push(callback);
    }
}

Client

import { newWebSocketRpcSession, RpcTarget } from 'capnweb';
import { CounterServer } from './types.js';

async function main() {
    console.log("[CLIENT] Connecting to Cap'n Web counter server...");

    let currentValue = 0;

    try {
        // Connect to the WebSocket server
        const api = newWebSocketRpcSession<CounterServer>('ws://localhost:8080');

        await api.subscribe(() => ({
            onUpdate: (newVal: number) => {
                currentValue = newVal;
                console.log(`[CLIENT] updated: ${newVal}`);
            },
        }));

        await api.increment();
        await api.increment();
        await api.increment();

        const valueFromServer = await api.get();
        console.log(`[CLIENT] re-fetch: ${valueFromServer}`);

        console.log(`[CLIENT] cached: ${currentValue}`);
    } catch (error) {
        console.error('Error:', error);
        process.exit(1);
    }
}

main().catch(console.error);

Logs

# Server
Simple Cap'n Web server running on http://localhost:8080
[SERVER] Increment 1
[SERVER] Notifying client 1
[SERVER] Increment 2
[SERVER] Notifying client 2
[SERVER] Increment 3
[SERVER] Notifying client 3

# Client
[CLIENT] Connecting to Cap'n Web counter server...
[CLIENT] re-fetch: 3
[CLIENT] cached: 0

[kentonv]

        await api.subscribe(() => ({
            onUpdate: (newVal: number) => {
                currentValue = newVal;
                console.log(`[CLIENT] updated: ${newVal}`);
            },
        }));

The callback you are passing to subscribe here seems to be a function that takes no arguments and returns an object containing an onUpdate function.

But on the server you are expecting the parameter to subscribe() is the callback that takes a number.

Your onUpdate is never being called because the server never calls it; it only calls the outer function.

[jim-alexander]

haha fuck.

Either way in the fixed CounterClient method or even using callbacks correctly 😅 its not passing the value back

        await api.subscribe((newVal) => {
            console.log(`[CLIENT] updated: ${newVal}`);
            currentValue = newVal;
        });

Curious, have you got a pubsub pattern working yourself?

Edit:

Interestingly if i call the callback directly from the subscribe method it does work but just storing the callback to use later doesn't

    async subscribe(callback) {
        callback(this.count)
        console.log('[SERVER] Subscribing client', callback)
        this.callbacks.push(callback)
    }

Will play around more..

[kentonv]

Ah, another problem is that any stubs passed as arguments to your RPC method will be disposed as soon as it returns. To keep copies of them long term, you need to call .dup() on them.

this.callbacks.push(callback.dup())

This is covered in the readme.

Also, if you were properly catching exceptions from the callback and reporting them, it would have told you the problem is that it is disposed. However in order to catch exceptions from the callback, you have to await it. Currently it looks like you have it in a try/catch but you did not await it, so the try/catch ends before the callback actually throws.

(This would also normally show up as an "unhandled rejection", which you could be monitoring for and reporting globally.)

[jim-alexander]

Amazing thanks for the continued guidance @kentonv.

I've created a repo incase others like me want to see a minimal example of this flow: https://github.com/jim-alexander/capnweb-pubsub-stub-example