binaryrpc.js

class BinaryRPC {
    constructor(url, options = {}) {
        this.url = url;
        this.options = {
            reconnectInterval: 1000,
            maxReconnectAttempts: 5,
            ...options
        };
        this.ws = null;
        this.reconnectAttempts = 0;
        this.handlers = new Map(); // For server-initiated RPCs and client call responses
        this.pendingCalls = new Map(); // Kept for potential future use, but primary mechanism is reply_to
        this.connected = false;
        this.sessionToken = null;
        this.clientId = null;
        this.deviceId = null;
        this.nextId = 1; // Used for generating unique reply_to handlers
    }

    static FRAME_TYPES = {
        DATA: 0x00,
        ACK: 0x01,
        PREPARE: 0x02,
        PREPARE_ACK: 0x03,
        COMMIT: 0x04,
        COMPLETE: 0x05
    };

    _createFrame(type, id, payload = new Uint8Array()) {
        const frame = new Uint8Array(1 + 8 + payload.length);

        // Frame type
        frame[0] = type;

        // ID (big endian)
        const idBuffer = new ArrayBuffer(8);
        const idView = new DataView(idBuffer);
        idView.setBigUint64(0, BigInt(id), false);
        frame.set(new Uint8Array(idBuffer), 1);

        // Payload
        if (payload.length > 0) {
            frame.set(payload, 9);
        }

        return frame;
    }

    connect(clientId, deviceId, sessionToken = null) {
        this.clientId = clientId;
        this.deviceId = deviceId;
        this.sessionToken = sessionToken;

        const queryParams = new URLSearchParams({
            clientId: clientId,
            deviceId: deviceId
        });

        if (sessionToken) {
            queryParams.append('sessionToken', sessionToken);
        }

        const wsUrl = `${this.url}?${queryParams.toString()}`;
        this.ws = new WebSocket(wsUrl);

        this.ws.binaryType = 'arraybuffer';

        this.ws.onopen = () => {
            console.log('Connected to BinaryRPC server');
            this.connected = true;
            this.reconnectAttempts = 0;
            if (this.options.onConnect) {
                this.options.onConnect();
            }
        };

        this.ws.onclose = () => {
            console.log('Disconnected from BinaryRPC server');
            this.connected = false;
            this._handleReconnect();
        };

        this.ws.onerror = (error) => {
            console.error('WebSocket error:', error);
            if (this.options.onError) {
                this.options.onError(error);
            }
        };

        this.ws.onmessage = (event) => {
            this._handleMessage(event.data);
        };
    }

    _handleMessage(data) {
        try {
            const frame = new Uint8Array(data);

            if (frame.length < 9) {
                console.error('Invalid frame size');
                return;
            }

            const type = frame[0];
            const id = new DataView(frame.buffer, 1, 8).getBigUint64(0, false);
            const payload = frame.slice(9);

            switch (type) {
                case BinaryRPC.FRAME_TYPES.DATA:
                    // This is a message from the server. It could be a QoS1 response or a server-initiated RPC.

                    // Step 1: Handle QoS requirement. If the message has an ID, it's QoS >= 1 and requires an ACK.
                    if (id > 0) {
                        console.log(`Received QoS1 DATA frame (ID: ${id}), sending ACK.`);
                        this._sendFrame(BinaryRPC.FRAME_TYPES.ACK, id);
                    }

                    // Step 2: Process the payload and dispatch to the correct handler.
                    if (payload.length > 0) {
                        const message = MessagePack.decode(payload);
                        // The `method` field directs the payload to the correct handler.
                        // This works for server-initiated RPCs and for responses to our `call`s,
                        // which use a temporary `reply_to` method name.
                        if (message.method) {
                            const handler = this.handlers.get(message.method);
                            if (handler) {
                                handler(this._decodePayload(message.payload));
                            } else {
                                console.warn(`No handler found for method: ${message.method}`);
                            }
                        }
                    }
                    break;

                case BinaryRPC.FRAME_TYPES.ACK:
                    // Server acknowledges a QoS 1 message we sent. We can just log this.
                    console.log(`Received ACK for message ${id}`);
                    break;

                case BinaryRPC.FRAME_TYPES.PREPARE:
                    // Server wants to start a QoS 2 message delivery. We must acknowledge.
                    this._sendFrame(BinaryRPC.FRAME_TYPES.PREPARE_ACK, id);
                    break;

                case BinaryRPC.FRAME_TYPES.COMMIT:
                    // Server confirms the QoS 2 message. We must send COMPLETE.
                    this._sendFrame(BinaryRPC.FRAME_TYPES.COMPLETE, id);
                    break;

                case BinaryRPC.FRAME_TYPES.COMPLETE:
                    // Server acknowledges our response in a QoS 2 flow.
                    console.log(`Received COMPLETE for message ${id}`);
                    break;
            }
        } catch (error) {
            console.error('Error handling message:', error);
        }
    }

    _decodePayload(payload) {
        if (!payload) return null;
        try {
            const utf8decoder = new TextDecoder("utf-8");
            const payloadStr = utf8decoder.decode(new Uint8Array(payload));
            return JSON.parse(payloadStr);
        } catch (e) {
            console.error('Error parsing JSON payload:', e);
            return null;
        }
    }

    _sendFrame(type, id, payload = new Uint8Array()) {
        if (!this.connected) {
            throw new Error('Not connected to server');
        }

        const frame = this._createFrame(type, id, payload);
        this.ws.send(frame);
    }

    registerHandler(method, handler) {
        this.handlers.set(method, handler);
    }

    call(method, payload = {}) {
        if (!this.connected)
            throw new Error('Not connected to server');

        const callId = this.nextId++;
        const encoded = MessagePack.encode({ method, payload });

        this._sendFrame(BinaryRPC.FRAME_TYPES.DATA, callId, encoded);
    }

    disconnect() {
        if (this.ws) {
            this.ws.close();
        }
    }

    _handleReconnect() {
        if (this.options.onError) {
            this.options.onError();
        }
        if (this.reconnectAttempts < this.options.maxReconnectAttempts) {
            this.reconnectAttempts++;
            console.log(`Attempting to reconnect (${this.reconnectAttempts}/${this.options.maxReconnectAttempts})...`);
            setTimeout(() => {
                this.connect(this.clientId, this.deviceId, this.sessionToken);
            }, this.options.reconnectInterval);
        } else {
            console.error('Max reconnection attempts reached');
            if (this.options.onMaxReconnectAttempts) {
                this.options.onMaxReconnectAttempts();
            }
        }
    }
}

// Export the class
if (typeof module !== 'undefined' && module.exports) {
    module.exports = BinaryRPC;
} else {
    window.BinaryRPC = BinaryRPC;
}