warpsocket
Advanced tools
Node-API addon for writing high-performance multi-threaded WebSocket servers.
How does this work?
So what WarpSocket buys you compared to the standard Node.js WebSocket library (ws) is:
Compared to PushPin, WarpSocket is:
npm install warpsocket
Requirements:
import { start, send, subscribe } from 'warpsocket';
// Start the server and point it at your worker module (defaults to one worker thread per CPU core)
start({ bind: '0.0.0.0:3000', workerPath: './my-worker.js' });
Create my-worker.js exporting any of the optional handlers:
export function handleOpen(socketId, ip, headers) {
console.log(`New connection ${socketId} from ip=${ip} origin=${headers.origin}`);
return true; // accept
}
export async function handleTextMessage(text, socketId) {
subscribe(socketId, 'general');
send('general', `User ${socketId}: ${text}`);
}
export async function handleTextMessage(data, socketId) {
console.error('Binary message received, but not handled');
}
export function handleClose(socketId) {
console.log('Closed', socketId);
}
When you start() a WarpSocket server from the main thread of your Node.js/Bun application, it will spawn a given number of JavaScript worker threads. Each of these worker threads will load a JavaScript file you specified (workerPath) and register itself with the native addon (written in Rust using NEON for the Node.js bindings). As these are actual threads, not processes, they share the same memory space. Though JavaScript objects are not shared between threads, the native addon does share its internal state (connections, channels, subscriptions, etc.) between all threads.
Besides firing up worker threads, when starting a WarpSocket server, the native addon will also bind to the specified address and start accepting WebSocket connections. It handles incoming connections and messages using asynchronous multi-threaded Rust code (using Tokio and Tungstenite). This should be fast!
Each incoming WebSocket connection is assigned a unique socket ID and coupled to a worker thread using a round-robin strategy. All events for that connection (open, incoming message, close) are then routed to that same worker thread. This means that your JavaScript code in the worker thread can maintain per-connection state in memory (e.g. in a Map keyed by socket ID) without needing to worry about synchronization between threads.
Events are scheduled to be run in the order that they came in (at least for messages coming from a single WebSocket) in the coupled worker's main thread. Worker handler functions may be synchronous or asynchronous. They may do whatever it is a backend server usually does: access databases, call other services, etc. Besides that, they may also call WarpSocket functions to send messages (to specific socket ids or channels) and subscribe to channels. Because WarpSocket data structures are shared between threads, a worker can subscribe to any channel and send messages to any channel/connection, even if that connection is coupled to a different worker thread.
Be aware that if you block a worker thread for too long, it will delay processing of all events for all connections assigned to that worker (as is usual in Node.js). In order to prevent an infinite loop (caused by a logic error) from bringing down the entire server, WarpSocket includes automatic worker thread monitoring and recovery. When a worker thread is unable to respond to a ping message within 3 seconds, it is considered unresponsive and will be terminated. All connections assigned to that worker are then closed (when they next send a message) with an appropriate error code, while other workers continue to operate normally. Client that get disconnected should be made to reconnect (they'll be assigned a new worker), reinitialize their state, and continue business as usual.
The examples/performance/ directory contains a simple benchmarking test. I'm planning to do a more thorough performance analysis on AWS soon (I've already had AI generate a Terraform config for it, but haven't had the heart to run it yet), but for now here's a workload I was able to sustain on my laptop:
I suspect that it will be possible to squeeze out more performance, using some profiling and optimization. But given the above numbers, I haven't felt the need yet.
WarpSocket provides a comprehensive TypeScript API for building real-time applications. The core functions allow you to start the server, send messages, handle channels, and manage authentication. All functions are fully typed and include detailed JSDoc documentation.
The following is auto-generated from src/index.cts:
Starts a WebSocket server bound to the given address and spawns worker threads that handle WebSocket events.
Signature: (options: { bind: string | string[]; workerPath?: string; threads?: number; workerArg?: any; }) => Promise<void>
Parameters:
options: { bind: string | string[], workerPath?: string, threads?: number, workerArg?: any } - - Configuration object:WorkerInterface handlers.Worker threads
are created and set up to handle WebSocket events. When omitted,
defaults to the number of CPU cores or 4, whichever is higher.Returns: A Promise that resolves after worker threads (if any) have been started and the native addon has been instructed to bind to the address. The Promise rejects if worker initialization fails.
Throws:
options is or the bind and workerPath properties are
missing or invalid, or if already started.[object Object],[object Object],[object Object]
Signature: (socketIdOrChannelName: string | number | number[] | ArrayBuffer | Uint8Array<ArrayBufferLike> | (string | number | ArrayBuffer | Uint8Array<ArrayBufferLike>)[], channelName: string | ... 1 more ... | Uint8Array<...>, delta?: number) => number[]
Parameters:
socketIdOrChannelName: number | number[] | Uint8Array | ArrayBuffer | string | (number | Uint8Array | ArrayBuffer | string)[]channelName: Uint8Array | ArrayBuffer | stringdelta: number (optional)Signature: (fromChannelName: string | ArrayBuffer | Uint8Array<ArrayBufferLike>, toChannelName: string | ArrayBuffer | Uint8Array<ArrayBufferLike>) => number[]
Parameters:
fromChannelName: Uint8Array | ArrayBuffer | string - - The source channel name (Buffer, ArrayBuffer, or string).toChannelName: Uint8Array | ArrayBuffer | string - - The destination channel name (Buffer, ArrayBuffer, or string).Returns: An array of socket IDs that were newly added to the destination channel. Sockets that were already subscribed (and had their reference count incremented) are not included.
Interface that worker threads must implement to handle WebSocket events. All handler methods are optional - if not provided, the respective functionality will be unavailable.
Called when the worker is starting up, before registering with the native addon. This allows for initialization logic that needs to run before handling WebSocket events.
Type: (workerArg?: any) => void
Handles new WebSocket connections and can reject them. If not provided, all connections are accepted.
Type: (socketId: number, ip: string, headers: Record<string, string>) => boolean
Handles incoming WebSocket text messages from clients.
Type: (data: string, socketId: number) => void
Handles incoming WebSocket binary messages from clients.
Type: (data: Uint8Array<ArrayBufferLike>, socketId: number) => void
Handles WebSocket connection closures.
Type: (socketId: number) => void
Sends data to a specific WebSocket connection, multiple connections, or broadcasts to all subscribers of a channel.
Signature: (target: string | number | number[] | ArrayBuffer | Uint8Array<ArrayBufferLike> | (string | number | ArrayBuffer | Uint8Array<ArrayBufferLike>)[], data: string | ... 1 more ... | Uint8Array<...>) => number
Parameters:
target - - The target for the message:data - - The data to send (Buffer, ArrayBuffer, or string).Returns: the number of recipients that got sent the message.
When target is a virtual socket with user prefix (or a channel that has such a subscriber), that prefix is prepended to the message. In case of a text message, the prefix bytes are assumed to be valid UTF-8.
When target is an array, the message is sent to each target in the array.
Subscribes one or more WebSocket connections to a channel, or copies subscriptions from one channel to another. Multiple subscriptions to the same channel by the same connection are reference-counted.
Signature: (socketIdOrChannelName: string | number | number[] | ArrayBuffer | Uint8Array<ArrayBufferLike> | (string | number | ArrayBuffer | Uint8Array<ArrayBufferLike>)[], channelName: string | ... 1 more ... | Uint8Array<...>, delta?: number) => number[]
Parameters:
socketIdOrChannelName - - Can be:channelName - - The target channel name (Buffer, ArrayBuffer, or string).delta - - Optional. The amount to change the subscription count by (default: 1).
Positive values add subscriptions, negative values remove them. When the count reaches zero, the subscription is removed.Returns: An array of socket IDs that were affected by the operation:
Checks if a channel has any subscribers.
Signature: (channelName: string | ArrayBuffer | Uint8Array<ArrayBufferLike>) => boolean
Parameters:
channelName - - The name of the channel to check (Buffer, ArrayBuffer, or string).Returns: True if the channel has subscribers, false otherwise.
Creates a virtual socket that points to an actual WebSocket connection. Virtual sockets can be subscribed to channels, and messages will be relayed to the underlying actual socket. This allows for convenient bulk unsubscription by deleting the virtual socket. Virtual sockets can also point to other virtual sockets, creating a chain that resolves to an actual socket.
Signature: (socketId: number, userPrefix?: string | ArrayBuffer | Uint8Array<ArrayBufferLike>) => number
Parameters:
socketId - - The identifier of the actual WebSocket connection or another virtual socket to point to.userPrefix - - Optional user prefix (up to 15 bytes) that will be prepended to all messages sent to this virtual socket (possibly through a channel). For text messages, this prefix is assumed to be valid UTF-8.Returns: The unique identifier of the newly created virtual socket, which can be used just like another socket.
Deletes a virtual socket and unsubscribes it from all channels. This is a convenient way to bulk-unsubscribe a virtual socket from all its channels at once.
Signature: (virtualSocketId: number, expectedTargetSocketId?: number) => boolean
Parameters:
virtualSocketId - - The unique identifier of the virtual socket to delete.expectedTargetSocketId - - Optional. If provided, the virtual socket will only be deleted
if it points to this specific target socket ID. This can help prevent unauthorized unsubscribes.Returns: true if the virtual socket was deleted, false if it was not found or target didn't match.
Reads the raw bytes stored for a key in the shared in-memory store.
Signature: (key: string | ArrayBuffer | Uint8Array<ArrayBufferLike>) => Uint8Array<ArrayBufferLike>
Parameters:
key - - Key to read (Buffer, ArrayBuffer, or string).Returns: A Uint8Array when the key exists, or undefined otherwise.
Stores or deletes a value in the shared key/value store.
Signature: (key: string | ArrayBuffer | Uint8Array<ArrayBufferLike>, value?: string | ArrayBuffer | Uint8Array<ArrayBufferLike>) => Uint8Array<...>
Parameters:
key - - Key to upsert (Buffer, ArrayBuffer, or string).value - - Optional value to store. Pass undefined to delete the key instead.Returns: The previous value as a Uint8Array when the key existed, or undefined if it did not.
Atomically updates a key only when its current value matches the expected check value.
Signature: (key: string | ArrayBuffer | Uint8Array<ArrayBufferLike>, newValue?: string | ArrayBuffer | Uint8Array<ArrayBufferLike>, checkValue?: string | ... 1 more ... | Uint8Array<...>) => boolean
Parameters:
key - - Key to update (Buffer, ArrayBuffer, or string).newValue - - Optional replacement value. Pass undefined to delete the key on success.checkValue - - Optional expected value. Pass undefined to require that the key is absent.Returns: true when the compare-and-set succeeds, false otherwise.
import { start, send, subscribe, unsubscribe } from 'warpsocket';
start({ bind: '0.0.0.0:3000', workerPath: './chat-worker.js' });
chat-worker.js:
export function handleTextMessage(data, socketId) {
const message = JSON.parse(data);
switch (message.type) {
case 'join':
subscribe(socketId, message.room);
send(message.room, JSON.stringify({ type: 'user-joined', userId: socketId, room: message.room }));
break;
case 'leave':
unsubscribe(socketId, message.room);
send(message.room, JSON.stringify({ type: 'user-left', userId: socketId, room: message.room }));
break;
case 'message':
send(message.room, JSON.stringify({ type: 'chat-message', userId: socketId, text: message.text, timestamp: Date.now() }));
break;
}
};
npm run build: Builds TypeScript files to JavaScript in dist/ and builds the native addon (see below).npm run build:native: Builds only the native addon. This creates build/<platform>-<arch>.node using your local Rust toolchain. To build a debug binary, run: npm run build:native -- --debug.npm run docs: Updates the reference documentation section of README.md based on src/index.cts.The test suite consists of end-to-end tests that start a real server instance and connect to it using Node.js WebSocket clients. The tests are located in the test/e2e/ directory and can be run with:
npm test
crates/warpsocket/src/lib.rs or TypeScript code in src/npm run debug for development or npm run build for productiontest/e2e/ and running them with npm testnpm test to ensure everything works correctlynpm run docs if you've changed TypeScript interfaces or JSDoc commentsThe project includes example code to help you get started:
# Build and run the example server
npm run build && node dist/examples/chat/example.ts
# Or without compilation step (if using bun or a very recent Node.js)
bun examples/chat/example.ts
# Point your browser at http://localhost:3000
Start the server using:
node dist/examples/performance/server/server.js --bind 0.0.0.0:3000 --threads 16
Start multiple servers, preferably on different machines:
node dist/examples/performance/client/client.js --host 127.0.0.1 --port 3000 --conns 10000
The directory structure of this project is:
warpsocket/
├── Cargo.toml
├── README.md
├── dist/ # Generated TypeScript output
├── src/ # TypeScript source files
| ├── index.cts # CommonJS entry point (includes Worker spawning logic)
| ├── index.mts # ESM entry point (just loads the CJS entry point)
| └── addon-loader.cts # Loader for platform-specific binaries
├── crates/ # Rust source code
| └── warpsocket/
| └── src/
| └── lib.rs # Main Rust implementation
├── examples/ # Example applications
| ├── chat/ # Chat example
| │ ├── example.ts # Server example (sets up WarpSocket and static HTTP)
| │ ├── worker.ts # Event-handling logic for the example, ran in worker threads
| │ └── client/ # Client-side code for the example
├── build/ # Path for native addon binaries
├── build-addon.js # Build script for the native addon
├── package.json
└── target/ # Intermediate Rust build artifacts
ISC - see LICENSE.txt file for details.
FAQs
Node-API addon for writing high-performance multi-threaded WebSocket servers.
We found that warpsocket demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Multiple high-impact npm maintainers confirm they have been targeted in the same social engineering campaign that compromised Axios.
Security News
Axios compromise traced to social engineering, showing how attacks on maintainers can bypass controls and expose the broader software supply chain.
Security News
Node.js has paused its bug bounty program after funding ended, removing payouts for vulnerability reports but keeping its security process unchanged.