Skip to main content
websocket-engineer Expert WebSocket and real-time communication engineering — WebSocket protocol (RFC 6455), ws/Socket.IO server implementation, authentication, heartbeat/reconnection, horizontal scaling with Redis pub/sub, Server-Sent Events (SSE), WebRTC signaling, and observability. Use for chat, live dashboards, multiplayer, collaborative editing, and event-driven real-time systems.
跳到安装 Skills Marketplace 发现并探索由社区构建的 Agent Skills
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/oimiragieo/agent-studio --skill websocket-engineer命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
下载 Zip 下载中... name websocket-engineer description Expert WebSocket and real-time communication engineering — WebSocket protocol (RFC 6455), ws/Socket.IO server implementation, authentication, heartbeat/reconnection, horizontal scaling with Redis pub/sub, Server-Sent Events (SSE), WebRTC signaling, and observability. Use for chat, live dashboards, multiplayer, collaborative editing, and event-driven real-time systems. version 1.0.0 model sonnet invoked_by both user_invocable true tools ["Bash","Read","Write","Edit"] best_practices ["Always implement heartbeat (ping/pong) to detect dead connections","Authenticate at upgrade time, not per-message","Use Redis pub/sub for horizontal scaling","Rate-limit message ingestion to prevent amplification attacks","Handle all error and close events explicitly"] error_handling graceful streaming supported verified false lastVerifiedAt "2026-03-14T00:00:00.000Z" source builtin trust_score 100 provenance_sha b764f6418ee0192b
WebSocket Engineer Skill
Overview
Real-time, bidirectional communication systems using WebSockets and related protocols. Covers protocol fundamentals, server implementation, authentication, scaling, fault tolerance, and browser client patterns.
Protocol Fundamentals
HTTP Upgrade → WebSocket Handshake
GET /ws HTTP/1.1
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: <base64 random>
Sec-WebSocket-Version: 13
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: <SHA-1 of key + GUID>
After handshake: full-duplex binary framing
Frame types: text (0x1), binary (0x2), close (0x8), ping (0x9), pong (0xA)
Server Implementation — ws (Node.js)
const { WebSocketServer , WebSocket } = require ('ws' );
const http = require ('http' );
const server = http.createServer (app);
const wss = new WebSocketServer ({ server, path : '/ws' });
const clients = new Map ();
wss.on ('connection' , (ws, req ) => {
const token = extractToken (req);
const user = verifyToken (token);
if (!user) {
ws.close (4001 , 'Unauthorized' );
return ;
}
if (!clients.has
id
set
id
new
Set
get
id
add
userId
id
isAlive
true
on
'pong'
() =>
isAlive
true
on
'message'
(data, isBinary ) =>
try
const
JSON
parse
toString
handleMessage
catch
send
JSON
stringify
type
'error'
message
'Invalid message'
on
'close'
(code, reason ) =>
get
userId
delete
if
get
userId
size
0
delete
userId
console
log
`Client ${ws.userId} disconnected: ${code} ${reason} `
on
'error'
err =>
console
error
`WebSocket error for ${ws.userId} :`
message
send
JSON
stringify
type
'connected'
userId
id
const
setInterval
() =>
clients
forEach
ws =>
if
isAlive
terminate
return
isAlive
false
ping
30_000
on
'close'
() =>
clearInterval
function
sendToUser
userId, payload
const
get
if
return
const
JSON
stringify
forEach
ws =>
if
readyState
WebSocket
OPEN
send
function
broadcast
payload, excludeUserId = null
const
JSON
stringify
clients
forEach
ws =>
if
readyState
WebSocket
OPEN
userId
send
Authentication Patterns
wss.on ('connection' , (ws, req ) => {
const url = new URL (req.url , 'http://localhost' );
const token = url.searchParams .get ('token' );
});
wss.on ('connection' , (ws, req ) => {
const cookies = parseCookies (req.headers .cookie );
const sessionId = cookies['session' ];
});
ws.once ('message' , data =>const { type, token } = JSON .parse (data);
if (type !== 'auth' ) {
ws.close (4001 , 'Auth required' );
return ;
}
const user = verifyToken (token);
if (!user) {
ws.close (4001 , 'Unauthorized' );
return ;
}
ws.userId = user.id ;
setupMessageHandler (ws);
});
Horizontal Scaling with Redis const { createClient } = require ('redis' );
const { WebSocketServer } = require ('ws' );
const pub = createClient ({ url : process.env .REDIS_URL });
const sub = createClient ({ url : process.env .REDIS_URL });
await pub.connect ();
await sub.connect ();
const CHANNEL = 'ws:broadcast' ;
await sub.subscribe (CHANNEL , message =>const payload = JSON .parse (message);
wss.clients .forEach (ws =>if (ws.readyState === WebSocket .OPEN ) {
if (!payload.targetUserId || ws.userId === payload.targetUserId ) {
ws.send (JSON .stringify (payload.data ));
}
}
});
});
async function broadcastViaRedis (data, targetUserId = null ) {
await pub.publish (
CHANNEL ,
JSON .stringify ({ data, targetUserId, serverId : process.env .SERVER_ID })
);
}
async function joinRoom (userId, room ) {
await pub.sAdd (`room:${room} ` , userId);
}
async function broadcastToRoom (room, data ) {
const members = await pub.sMembers (`room:${room} ` );
for (const userId of members) {
await pub.publish (`user:${userId} ` , JSON .stringify (data));
}
}
Socket.IO (Higher-Level Abstraction) const { Server } = require ('socket.io' );
const { createAdapter } = require ('@socket.io/redis-adapter' );
const io = new Server (httpServer, {
cors : { origin : process.env .FRONTEND_URL , credentials : true },
adapter : createAdapter (pub, sub),
transports : ['websocket' , 'polling' ],
pingTimeout : 20_000 ,
pingInterval : 10_000 ,
});
io.use (async (socket, next) => {
const token = socket.handshake .auth .token ;
const user = await verifyToken (token);
if (!user) return next (new Error ('Unauthorized' ));
socket.userId = user.id ;
next ();
});
io.on ('connection' , socket =>join (`user:${socket.userId} ` );
socket.on ('join-room' , roomId =>join (roomId);
socket.to (roomId).emit ('user-joined' , { userId : socket.userId });
});
socket.on ('message' , ({ roomId, text } ) => {
socket.to (roomId).emit ('message' , { userId : socket.userId , text, ts : Date .now () });
});
socket.on ('disconnect' , reason =>console .log (`${socket.userId} disconnected: ${reason} ` );
});
});
io.to (`user:${userId} ` ).emit ('notification' , { message : 'You have a new message' });
io.to (roomId).emit ('event' , data);
Browser Client class RealtimeClient {
#ws = null ;
#reconnectDelay = 1_000 ;
#maxDelay = 30_000 ;
#handlers = new Map ();
constructor (url, token ) {
this .url = url;
this .token = token;
this .connect ();
}
connect (this .#ws = new WebSocket (`${this .url} ?token=${this .token} ` );
this .#ws.onopen = () => {
console .log ('Connected' );
this .#reconnectDelay = 1_000 ;
this .emit ('connected' );
};
this .#ws.onmessage = ({ data } ) => {
try {
const msg = JSON .parse (data);
this .#handlers.get (msg.type )?.forEach (cb =>cb (msg));
this .#handlers.get ('*' )?.forEach (cb =>cb (msg));
} catch (e) {
console .error ('Parse error' , e);
}
};
this .#ws.onclose = event =>if (event.code === 4001 ) {
this .emit ('unauthorized' );
return ;
}
console .log (`Disconnected (${event.code} ), retrying in ${this .#reconnectDelay} ms` );
setTimeout (() => this .connect (), this .#reconnectDelay);
this .#reconnectDelay = Math .min (this .#reconnectDelay * 2 , this .#maxDelay);
};
this .#ws.onerror = err =>console .error ('WebSocket error' , err);
};
}
on (type, handler ) {
if (!this .#handlers.has (type)) this .#handlers.set (type, new Set ());
this .#handlers.get (type).add (handler);
return () => this .#handlers.get (type).delete (handler);
}
send (type, data = {} ) {
if (this .#ws?.readyState !== WebSocket .OPEN ) {
console .warn ('Not connected, dropping message:' , type);
return ;
}
this .#ws.send (JSON .stringify ({ type, ...data }));
}
emit (type, data ) {
this .#handlers.get (type)?.forEach (cb =>cb (data));
}
close (this .#ws?.close (1000 , 'Client closing' );
}
}
const client = new RealtimeClient ('wss://api.example.com/ws' , authToken);
const unsubscribe = client.on ('message' , msg =>displayMessage (msg));
client.send ('join-room' , { roomId : 'general' });
unsubscribe ();
Server-Sent Events (SSE) — Unidirectional Alternative
app.get ('/events' , (req, res ) => {
const user = authenticate (req);
if (!user) {
res.sendStatus (401 );
return ;
}
res.setHeader ('Content-Type' , 'text/event-stream' );
res.setHeader ('Cache-Control' , 'no-cache' );
res.setHeader ('Connection' , 'keep-alive' );
res.flushHeaders ();
const send = (event, data ) => {
res.write (`event: ${event} \ndata: ${JSON .stringify(data)} \n\n` );
};
send ('connected' , { userId : user.id });
const unsubscribe = eventBus.on (`user:${user.id} ` , data =>send ('update' , data));
const heartbeat = setInterval (() => res.write (':heartbeat\n\n' ), 15_000 );
req.on ('close' , () => {
clearInterval (heartbeat);
unsubscribe ();
});
});
const source = new EventSource ('/events' );
source.addEventListener ('update' , e =>const data = JSON .parse (e.data );
updateUI (data);
});
source.onerror = () => {
console .log ('SSE connection lost, browser will reconnect automatically' );
};
Message Protocol Design
const MessageTypes = {
JOIN_ROOM : 'join_room' ,
LEAVE_ROOM : 'leave_room' ,
SEND_MESSAGE : 'send_message' ,
TYPING_START : 'typing_start' ,
TYPING_STOP : 'typing_stop' ,
ROOM_JOINED : 'room_joined' ,
MESSAGE_RECEIVED : 'message_received' ,
USER_JOINED : 'user_joined' ,
USER_LEFT : 'user_left' ,
ERROR : 'error' ,
};
const sendMessageSchema = {
type : 'send_message' ,
roomId : 'string (required)' ,
text : 'string (max 2000 chars)' ,
replyTo : 'string | null (message ID)' ,
};
function validateMessage (msg ) {
if (!msg.type || typeof msg.type !== 'string' ) throw new Error ('Missing type' );
if (!MessageTypes [msg.type .toUpperCase ()]) throw new Error (`Unknown type: ${msg.type} ` );
switch (msg.type ) {
case MessageTypes .SEND_MESSAGE :
if (!msg.roomId || !msg.text ) throw new Error ('Missing roomId or text' );
if (msg.text .length > 2000 ) throw new Error ('Text too long' );
break ;
}
}
Rate Limiting const rateLimiters = new Map ();
function checkRateLimit (userId, limit = 60 , windowMs = 60_000 ) {
const now = Date .now ();
const limiter = rateLimiters.get (userId) ?? { count : 0 , resetAt : now + windowMs };
if (now > limiter.resetAt ) {
limiter.count = 0 ;
limiter.resetAt = now + windowMs;
}
limiter.count ++;
rateLimiters.set (userId, limiter);
if (limiter.count > limit) {
throw new Error (`Rate limit exceeded: ${limit} messages per ${windowMs / 1000 } s` );
}
}
Observability
const metrics = {
connections : 0 ,
messagesIn : 0 ,
messagesOut : 0 ,
errors : 0 ,
avgLatency : 0 ,
};
function measureLatency (ws ) {
const start = Date .now ();
const payload = Buffer .alloc (8 );
payload.writeBigInt64BE (BigInt (start));
ws.ping (payload);
ws.once ('pong' , data =>const sent = Number (data.readBigInt64BE ());
metrics.avgLatency = Date .now () - sent;
});
}
Security
Always use wss:// (TLS) in production — plain ws:// is MITM-vulnerable
Authenticate at the HTTP upgrade stage, not per-message
Rate-limit both connection attempts and message frequency
Validate and sanitize ALL incoming message content (treat as untrusted)
Set maxPayload on ws.WebSocketServer to prevent memory exhaustion attacks
Use ws.close(code, reason) for intentional closes; ws.terminate() only for dead connections
Never echo user-controlled data without sanitization (XSS via message relay)
Enable CORS properly for Socket.IO — do not use origin: '*' in production
const wss = new WebSocketServer ({ server, maxPayload : 64 * 1024 });
Related 