Persistent connections even if client disconnects
This commit is contained in:
+142
-15
@@ -4,8 +4,11 @@ import * as tls from 'tls';
|
||||
import http from 'http';
|
||||
import { parse } from 'url';
|
||||
|
||||
// Configuration for connection persistence
|
||||
const CONNECTION_PERSISTENCE_TIMEOUT = 5 * 60 * 1000; // 5 minutes in milliseconds
|
||||
// Default configuration for connection persistence (fallback values)
|
||||
const DEFAULT_PERSISTENCE_TIMEOUT = 5 * 60 * 1000; // 5 minutes in milliseconds
|
||||
const DEFAULT_MAX_BUFFER_MESSAGES = 100; // Maximum number of messages to buffer
|
||||
const DEFAULT_MAX_BUFFER_SIZE_KB = 10; // Maximum buffer size in KB
|
||||
|
||||
const HEARTBEAT_INTERVAL = 30 * 1000; // 30 seconds
|
||||
|
||||
// Create HTTP server
|
||||
@@ -15,17 +18,113 @@ const server = http.createServer();
|
||||
const wss = new WebSocketServer({ noServer: true });
|
||||
|
||||
// Active connections and their proxies
|
||||
// Key: connectionId, Value: { ws, socket, sessionId, settings }
|
||||
const connections = new Map();
|
||||
|
||||
// Persistent connections waiting for reconnection
|
||||
// Key: sessionId, Value: { socket, mudHost, mudPort, useSSL, timeoutId, lastActivity }
|
||||
// Key: sessionId, Value: { socket, mudHost, mudPort, useSSL, timeoutId, lastActivity, messageBuffer, settings }
|
||||
const persistentConnections = new Map();
|
||||
|
||||
// Parse connection settings from URL parameters with defaults
|
||||
function parseConnectionSettings(urlParts) {
|
||||
const persistenceTimeoutParam = urlParts.searchParams.get('persistenceTimeout');
|
||||
const maxBufferMessagesParam = urlParts.searchParams.get('maxBufferMessages');
|
||||
const maxBufferSizeKBParam = urlParts.searchParams.get('maxBufferSizeKB');
|
||||
|
||||
return {
|
||||
persistenceTimeoutMs: persistenceTimeoutParam ?
|
||||
parseInt(persistenceTimeoutParam) * 60 * 1000 : // Convert minutes to milliseconds
|
||||
DEFAULT_PERSISTENCE_TIMEOUT,
|
||||
maxBufferMessages: maxBufferMessagesParam ?
|
||||
parseInt(maxBufferMessagesParam) :
|
||||
DEFAULT_MAX_BUFFER_MESSAGES,
|
||||
maxBufferSizeKB: maxBufferSizeKBParam ?
|
||||
parseInt(maxBufferSizeKBParam) :
|
||||
DEFAULT_MAX_BUFFER_SIZE_KB
|
||||
};
|
||||
}
|
||||
|
||||
// Generate a unique session ID for persistent connections
|
||||
function generateSessionId() {
|
||||
return `session-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
|
||||
}
|
||||
|
||||
// Buffer a message for a persistent connection
|
||||
function bufferMessage(sessionId, data) {
|
||||
const persistentConn = persistentConnections.get(sessionId);
|
||||
if (!persistentConn) {
|
||||
return; // No persistent connection to buffer for
|
||||
}
|
||||
|
||||
if (!persistentConn.messageBuffer) {
|
||||
persistentConn.messageBuffer = [];
|
||||
persistentConn.bufferSize = 0;
|
||||
}
|
||||
|
||||
// Add timestamp to the message
|
||||
const bufferedMessage = {
|
||||
data: data,
|
||||
timestamp: Date.now()
|
||||
};
|
||||
|
||||
persistentConn.messageBuffer.push(bufferedMessage);
|
||||
persistentConn.bufferSize += data.length;
|
||||
|
||||
// Use this connection's specific settings for buffer limits
|
||||
const settings = persistentConn.settings || {
|
||||
maxBufferMessages: DEFAULT_MAX_BUFFER_MESSAGES,
|
||||
maxBufferSizeKB: DEFAULT_MAX_BUFFER_SIZE_KB
|
||||
};
|
||||
|
||||
// Trim buffer if it gets too large
|
||||
while (persistentConn.messageBuffer.length > settings.maxBufferMessages ||
|
||||
persistentConn.bufferSize > settings.maxBufferSizeKB * 1000) {
|
||||
const removed = persistentConn.messageBuffer.shift();
|
||||
if (removed) {
|
||||
persistentConn.bufferSize -= removed.data.length;
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`Buffered ${data.length} bytes for session ${sessionId} (${persistentConn.messageBuffer.length} messages, ${persistentConn.bufferSize} bytes total, limits: ${settings.maxBufferMessages} msgs/${settings.maxBufferSizeKB}KB)`);
|
||||
}
|
||||
|
||||
// Replay buffered messages to a reconnected client
|
||||
function replayBufferedMessages(ws, sessionId) {
|
||||
const persistentConn = persistentConnections.get(sessionId);
|
||||
if (!persistentConn || !persistentConn.messageBuffer) {
|
||||
return 0; // No messages to replay
|
||||
}
|
||||
|
||||
const messages = persistentConn.messageBuffer;
|
||||
console.log(`Replaying ${messages.length} buffered messages for session ${sessionId}`);
|
||||
|
||||
// Send a notification about message replay
|
||||
const replayNotification = `[SYSTEM]${JSON.stringify({
|
||||
type: 'message_replay_start',
|
||||
messageCount: messages.length,
|
||||
timespan: messages.length > 0 ? Date.now() - messages[0].timestamp : 0
|
||||
})}`;
|
||||
ws.send(replayNotification);
|
||||
|
||||
// Send all buffered messages
|
||||
for (const message of messages) {
|
||||
if (ws.readyState === 1) { // WebSocket.OPEN
|
||||
ws.send(message.data);
|
||||
}
|
||||
}
|
||||
|
||||
// Send replay complete notification
|
||||
const replayComplete = `[SYSTEM]${JSON.stringify({ type: 'message_replay_complete' })}`;
|
||||
ws.send(replayComplete);
|
||||
|
||||
// Clear the buffer since messages have been replayed
|
||||
const messageCount = messages.length;
|
||||
persistentConn.messageBuffer = [];
|
||||
persistentConn.bufferSize = 0;
|
||||
|
||||
return messageCount;
|
||||
}
|
||||
|
||||
// Clean up a persistent connection
|
||||
function cleanupPersistentConnection(sessionId) {
|
||||
const persistentConn = persistentConnections.get(sessionId);
|
||||
@@ -54,11 +153,16 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
// Create a unique ID for this connection
|
||||
const connectionId = `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
|
||||
|
||||
// Check for session ID in query parameters for reconnection
|
||||
// Check for session ID and settings in query parameters
|
||||
const url = req.url || '';
|
||||
const urlParts = new URL(`http://localhost${url}`);
|
||||
const sessionId = urlParts.searchParams.get('sessionId');
|
||||
|
||||
// Parse connection settings for this specific connection
|
||||
const connectionSettings = parseConnectionSettings(urlParts);
|
||||
|
||||
console.log(`Connection settings for ${connectionId}: timeout=${connectionSettings.persistenceTimeoutMs/1000/60}min, maxMessages=${connectionSettings.maxBufferMessages}, maxSizeKB=${connectionSettings.maxBufferSizeKB}`);
|
||||
|
||||
// Special handling for test connections
|
||||
if (mudHost === 'example.com' && mudPort === '23') {
|
||||
console.log('Test connection detected - using echo server mode');
|
||||
@@ -98,11 +202,18 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
clearTimeout(persistentConn.timeoutId);
|
||||
}
|
||||
|
||||
// Remove from persistent connections (now active again)
|
||||
// Replay any buffered messages first
|
||||
const replayedCount = replayBufferedMessages(ws, sessionId);
|
||||
|
||||
// Remove from persistent connections (now active again) - do this after replay
|
||||
persistentConnections.delete(sessionId);
|
||||
|
||||
// Send reconnection notification with session ID in proper JSON format
|
||||
ws.send(`[SYSTEM]${JSON.stringify({ type: 'session_resumed', sessionId: sessionId })}`);
|
||||
ws.send(`[SYSTEM]${JSON.stringify({
|
||||
type: 'session_resumed',
|
||||
sessionId: sessionId,
|
||||
messagesReplayed: replayedCount
|
||||
})}`);
|
||||
} else {
|
||||
// Create new connection
|
||||
currentSessionId = generateSessionId();
|
||||
@@ -142,8 +253,13 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
}
|
||||
}
|
||||
|
||||
// Store the connection
|
||||
connections.set(connectionId, { ws, socket, sessionId: currentSessionId });
|
||||
// Store the connection with its settings
|
||||
connections.set(connectionId, {
|
||||
ws,
|
||||
socket,
|
||||
sessionId: currentSessionId,
|
||||
settings: connectionSettings
|
||||
});
|
||||
|
||||
// Handle data from the MUD server - only in regular mode, not test mode
|
||||
if (socket) {
|
||||
@@ -163,6 +279,11 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
if (ws.readyState === 1) { // WebSocket.OPEN
|
||||
ws.send(data);
|
||||
console.log(`WebSocket server: Sent ${data.length} bytes to client${isGmcp ? ' (contains GMCP data)' : ''}`);
|
||||
} else {
|
||||
// WebSocket is not open, buffer the message if we have a session
|
||||
if (currentSessionId) {
|
||||
bufferMessage(currentSessionId, data);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -276,13 +397,14 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
|
||||
const conn = connections.get(connectionId);
|
||||
if (conn && !conn.testMode && conn.socket && !conn.socket.destroyed) {
|
||||
console.log(`Moving connection to persistent state for ${CONNECTION_PERSISTENCE_TIMEOUT / 1000} seconds`);
|
||||
console.log(`Moving connection to persistent state for ${conn.settings.persistenceTimeoutMs / 1000} seconds`);
|
||||
|
||||
// Move the connection to persistent storage instead of closing it
|
||||
// Use this connection's specific timeout setting
|
||||
const timeoutId = setTimeout(() => {
|
||||
console.log(`Session ${currentSessionId} timed out, closing MUD connection`);
|
||||
cleanupPersistentConnection(currentSessionId);
|
||||
}, CONNECTION_PERSISTENCE_TIMEOUT);
|
||||
}, conn.settings.persistenceTimeoutMs);
|
||||
|
||||
persistentConnections.set(currentSessionId, {
|
||||
socket: conn.socket,
|
||||
@@ -290,10 +412,13 @@ wss.on('connection', (ws, req, mudHost, mudPort, useSSL) => {
|
||||
mudPort,
|
||||
useSSL,
|
||||
timeoutId,
|
||||
lastActivity: Date.now()
|
||||
lastActivity: Date.now(),
|
||||
messageBuffer: [],
|
||||
bufferSize: 0,
|
||||
settings: conn.settings // Store the connection's settings
|
||||
});
|
||||
|
||||
console.log(`Session ${currentSessionId} will persist for ${CONNECTION_PERSISTENCE_TIMEOUT / 1000} seconds`);
|
||||
console.log(`Session ${currentSessionId} will persist for ${conn.settings.persistenceTimeoutMs / 1000} seconds with settings: ${conn.settings.maxBufferMessages} msgs/${conn.settings.maxBufferSizeKB}KB`);
|
||||
} else if (conn && conn.socket) {
|
||||
// Fallback to immediate cleanup if needed
|
||||
conn.socket.end();
|
||||
@@ -347,18 +472,20 @@ setInterval(() => {
|
||||
const now = Date.now();
|
||||
for (const [sessionId, persistentConn] of persistentConnections.entries()) {
|
||||
// Clean up connections that have been inactive for too long
|
||||
if (now - persistentConn.lastActivity > CONNECTION_PERSISTENCE_TIMEOUT * 2) {
|
||||
// Use double the connection's specific timeout or default
|
||||
const timeoutThreshold = (persistentConn.settings?.persistenceTimeoutMs || DEFAULT_PERSISTENCE_TIMEOUT) * 2;
|
||||
if (now - persistentConn.lastActivity > timeoutThreshold) {
|
||||
console.log(`Cleaning up abandoned session: ${sessionId}`);
|
||||
cleanupPersistentConnection(sessionId);
|
||||
}
|
||||
}
|
||||
}, CONNECTION_PERSISTENCE_TIMEOUT);
|
||||
}, DEFAULT_PERSISTENCE_TIMEOUT); // Run cleanup every default timeout period
|
||||
|
||||
// Start the WebSocket server
|
||||
const PORT = process.env.WS_PORT || 3001;
|
||||
server.listen(PORT, () => {
|
||||
console.log(`WebSocket server is running on port ${PORT}`);
|
||||
console.log(`Connection persistence timeout: ${CONNECTION_PERSISTENCE_TIMEOUT / 1000} seconds`);
|
||||
console.log(`Default connection persistence timeout: ${DEFAULT_PERSISTENCE_TIMEOUT / 1000} seconds (configurable per connection)`);
|
||||
});
|
||||
|
||||
export default server;
|
||||
Reference in New Issue
Block a user