CAMP

CAMP is a small binary protocol for sending arbitrary messages, large data and real-time information over WebSocket. It uses fixed frame types for acknowledgements, heartbeats, UTF-8 text, binary payloads and streamed transactions.

This is a server implementation of the CAMP protocol specified at CAMP-Protocol

CAMP-Server / Overview

The CAMP-Server takes care of the following:

• Client session authentication
• Client session lifecycle management
• Correct framing and structuring of received and sent data

The CAMP-Server provides a public API for creating and destroying an instance. It provides access to sessions for communication purposes via events

The CAMP-Server is extensible via its Extension interface. As such it is possible to, for example, implement an RPC server.

Setup

To set up a CAMP Server, simply import the CAMP function from the CAMP-server package.

The camp-function takes two arguments:

• a required ITokenValidator object
• an optional ICAMPWebsocketServerOptions object

Configuration

ITokenValidator

interface ITokenValidator {
    validate(token: string): Promise<boolean>;
}

This interface describes how the server will authenticate incoming connections.

CAMP-Clients send a token, which you may validate in the validate-function.

ICAMPWebsocketServerOptions

interface ICAMPWebsocketServerOptions {
    keepAliveIntervalMs?: number;
    port?: number;
    ssl?: {
        key: Buffer;
        cert: Buffer;
    };
    backpressure?: {
        highWaterMark?: number;
        lowWaterMark?: number;
        maxQueuedBytes?: number;
        maxQueueCount?: number;
        dropPolicy?: "drop-oldest" | "drop-newest" | "dedupe-latest";
    };
}

This interface describes various configuration options for the CAMPServer:

• keepAliveIntervalMs
  • The frequency in which the server will send application-level pings to the clients, by default 15000 is chosen
• port
  • The local port to listen on, by default 8080 is chosen

In the ssl object, you may set a key and a cert to enable SSL connections.

Backpressure

The backpressure object allows to you control how the CAMP server handles high load and prevents memory bloating

| Option | Type | Default | Description | |------------------|-----------------------------------------------------|--------------------|------------------------------------------------------------------------------| | highWaterMark | number | 16 * 1024 * 1024 | Maximum number of bytes allowed in the send buffer before throttling starts. | | lowWaterMark | number | 1024 * 1024 | Once buffer usage drops below this, sending resumes. | | maxQueuedBytes | number | 8 * 1024 * 1024 | Hard cap on total queued bytes. Frames beyond this are dropped. | | maxQueueCount | number | 1024 | Hard cap on number of queued frames. | | dropPolicy | "drop-oldest" \| "drop-newest" \| "dedupe-latest" | "drop-oldest" | Strategy used when the queue is full. |

Drop policies

• drop-oldest
  • Keeps new data, discards the oldest frames first
• drop-newest
  • Keeps old data, discards the newest frame attempting to be sent
• dedupe-latest
  • If a frame of the same type or payload is already queued, replace it instead adding a duplicate

Public methods

| Name | Parameter | Description | Returns | |---------------------|-----------------------------------|------------------------------------------------------------------------------|---------------------------------------| | RegisterExtension | ext: ICAMPExtension | Registers a server-side CAMP extension | | | UnregisterExtension | ext: ICAMPExtension | Unregisters a server-side CAMP extension | | | GetExtension | id: string | Gets a server-side CAMP extension by its id | ICAMPExtension | null | | ConnectPeer | host: string host, bearer: string | Connects to another CAMP-server | Promise<CAMPServerWebsocketSession> | | Destroy | | Closes the server and destroys all sessions | | | (get) http_server | | Returns the underlying http.Server-object for interop with other libraries | |

CAMP-Server / Example

import {CAMP} from "CAMP-server";

const PORT = 8080;
const server = await CAMP({
    async validate(token: string): Promise<boolean> {
        return token === "MySuperSecretToken";
    }
}, {port: PORT});

/*
The CAMP server emits events as described by the CAMPWebsocketServerEvents interface

interface CAMPWebsocketServerEvents {
  "session": (session: CAMPServerWebsocketSession) => void;

  "listening": () => void;
}

as such, we can react to when the server starts listening and react to new sessions
*/
server.on("listening", () => {
    console.info("Server is listening...");

    server.on("session", async session => {
        console.info(`Session ID: ${session.id}`);
    });
});

CAMPServerWebsocketSession / Overview

Data Events

These events are emitted when the server-side session receives data from a client-side session

| Name | Parameter | Description | |----------------|--------------|----------------------------------------------------------------| | message-utf8 | data: string | Emitted, when the session receives a utf8 text message | | message-binary | data: Buffer | Emitted, when the session receives an arbitrary binary message | | message-error | data: string | Emitted, when the session receives an error message |

Statistic Events

These events are emitted periodically and serve as a way to implement metrics, monitoring or other such things

| Name | Parameter | Description | |------------------|--------------|---------------------------------------------------------------| | stat-rtt | data: number | EWMA Average in milliseconds between messages-ACK round trips | | stat-bytes-rx | data: number | Total bytes received from the client session | | stat-bytes-tx | data: number | Total bytes sent to the client session | | stat-ack-timeout | data: number | Count of messages that timed out waiting for an ACK |

Meta events

This sad category of events is emitted when something about the session changes

| Name | Parameter | Description | |--------|-----------|------------------------------------------------------------| | closed | | Emitted, when the session closes or was closed by a client |

Public methods

| Name | Parameter | Description | Returns | |---------|-------------------------|-------------------------------------------------------------------|---------| | Set | key: string, value: any | Sets key to value in the session's custom data store | | | Get | key: string | Retrieves the value of key from the session's custom data store | any | | Close | reason: string | Closes the session gracefully | | | Destroy | | Closes and destroys the session object | |

CAMP-Functionality is divided among namespaces as defined in CAMP.

Base

Accessible via session.base.<method>(...);

| Name | Parameter | Description | Returns | |------------|-----------------|------------------------------------------|---------| | Ping | | Pings the client | | | SendUTF8 | data: string | Sends the passed utf8 text to the client | | | SendBinary | data: Buffer | Sends the passed buffer to the client | | | SendError | message: string | Send an error to the client | |

Transaction

Accessible via session.transaction.<method>(...);

| Name | Parameter | Description | Returns | |--------------------|-----------------------------------------------------------------------------------|-------------------------------------------------------------|-------------------------| | Stream | source: Readable, options: { streamName: string, behaviour: CAMP_FLOW_BEHAVIOUR } | Streams a readable to the client | | | WaitForStream | name: string, timeout: number | Waits for a named stream from client | Promise<CAMPReadable> | | StreamRequestRange | stream: CAMPReadable, start: bigint, end: bigint | When in TX_PULL-mode, requests a byte range from the client | |

CAMP-Server / Extension interface

The extension interface allows you to modify the behaviour of the CAMP server's message lifecycle, before sending and after receiving per session.

Extensions can inspect and modify messages, enabling you to do custom behaviour, such as logging, filtering, building metrics or transforming the payloads.

Extensions are registered using the CAMP-server's RegisterExtension-method.

Interface definition

type Box<T> = { value: T };

export declare interface ICAMPExtension {
    /**
     * Executed upon registration of the extension on the server
     * @param server - Reference to the running CAMP websocket server
     * */
    on_register(server: CAMPWebsocketServer): void;

    /**
     * Executed upon unregistration of the extension on the server
     * @param server - Reference to the running CAMP websocket server
     * */
    on_unregister(server: CAMPWebsocketServer): void;

    /**
     * Executed before a binary message is sent to the client session
     * @param session - The CAMP websocket session
     * @param outgoing_message - The message buffer to be sent to the client
     * */
    before_send_binary?(session: CAMPServerWebsocketSession, outgoing_message: Box<Buffer>): Promise<boolean>;

    /**
     * Executed before a text message is sent to the client session
     * @param session - The CAMP websocket session
     * @param outgoing_message - The message text to be sent to the client
     * */
    before_send_utf8?(session: CAMPServerWebsocketSession, outgoing_message: Box<string>): Promise<boolean>;

    /**
     * Executed before an error message is sent to the client session
     * @param session - The CAMP websocket session
     * @param outgoing_message - The error message to be sent to the client
     * */
    before_send_error?(session: CAMPServerWebsocketSession, outgoing_message: Box<string>): Promise<boolean>;

    /**
     * Executed after a binary message is received from the client, but before the session can emit the `message-binary` event
     * @param session - The CAMP websocket session
     * @param incoming_message - The incoming binary message from the client
     * */
    on_receive_binary?(session: CAMPServerWebsocketSession, incoming_message: Box<Buffer>): Promise<boolean>;

    /**
     * Executed after a text message is received from the client, but before the session can emit the `message-utf8` event
     * @param session - The CAMP websocket session
     * @param incoming_message - The incoming text message from the client
     * */
    on_receive_utf8?(session: CAMPServerWebsocketSession, incoming_message: Box<string>): Promise<boolean>;

    /**
     * Executed after an error message is received from the client, but before the session can emit the `message-error` event
     * @param session - The CAMP websocket session
     * @param incoming_message - The incoming error message from the client
     * */
    on_receive_error?(session: CAMPServerWebsocketSession, incoming_message: Box<string>): Promise<boolean>;

    /**
     * The unique name of this extension
     * */
    name: string;
}

Example extension

This is an exemplary logging extension. It simply logs each incoming and outgoing message

import {Box, ICAMPExtension} from "./CAMPServerExtension";
import {CAMPServerWebsocketSession} from "./CAMPServerWebsocketSession";

class LoggerExtension implements ICAMPExtension {
    public name = "unique_name_for_logger";

    public async before_send_utf8(session: CAMPServerWebsocketSession, outgoing_message: Box<string>) {
        console.log(`Outgoing UTF-8 message: "${outgoing_message.value}"`);
        return true;
    }

    public async before_send_binary(session: CAMPServerWebsocketSession, outgoing_message: Box<Buffer>) {
        console.log(`Outgoing binary message: "${outgoing_message.value.toString(16)}"`);
        return true;
    }

    public async on_receive_utf8(session: CAMPServerWebsocketSession, incoming_message: Box<string>) {
        console.log(`Incoming UTF-8 message: "${outgoing_message.value}"`);
        return true;
    }

    public async before_send_utf8(session: CAMPServerWebsocketSession, incoming_message: Box<Buffer>) {
        console.log(`Incoming binary message: "${outgoing_message.value.toString(16)}"`);
        return true;
    }
}