9.8 KiB
9.8 KiB
sequenceDiagram
participant Client
participant HTTPServer
participant WebSocketServer
participant ClientMQ
participant ClientMQWorker
participant MQWorker
participant PingInterval as Ping Interval<br/>(20s)
Note over Client,ClientMQWorker: === Connection Establishment ===
Client->>HTTPServer: HTTP Upgrade Request
HTTPServer->>WebSocketServer: upgrade event
WebSocketServer->>WebSocketServer: handleUpgrade()
WebSocketServer->>WebSocketServer: Generate connection.id (32 chars)
WebSocketServer->>WebSocketServer: Get remoteAddress (from X-Forwarded-For)
WebSocketServer->>WebSocketServer: Setup event handlers<br/>(message, close, pong, error)
WebSocketServer->>ClientMQAdapter: connectDelegate(socket, headers)
ClientMQAdapter->>ClientMQAdapter: Set __connectionStart<br/>Set __connectionId<br/>Set __meta {headers}
ClientMQAdapter->>ClientMQAdapter: Register in _socketsById
ClientMQAdapter->>ClientMQAdapter: Set __connected = true
ClientMQAdapter->>ClientMQ: emit('connect', connectionId)
ClientMQ->>ClientMQ: Register adapter in _adaptersById
ClientMQ->>ClientMQWorker: emit('connect', connectionId)
ClientMQWorker->>ClientMQWorker: Increment _connectionsTotal<br/>Increment _connectionsOpen
alt Connection limit exceeded (>2000)
ClientMQWorker->>ClientMQ: close(connectionId, 1013, 'concurrency-limit-exceeded')
ClientMQ->>ClientMQAdapter: close(connectionId, code, reason)
Note over ClientMQAdapter: See "Server-Initiated Close" below
end
Note over Client,ClientMQWorker: === Normal Message Flow (Client Request) ===
Client->>WebSocketServer: Message (binary or base64)
WebSocketServer->>ClientMQAdapter: messageDelegate(socket, data)
ClientMQAdapter->>ClientMQAdapter: Decode message (MQ protocol)
alt Unsupported message format
ClientMQAdapter->>ClientMQAdapter: Drop message, log warning
else Socket not connected
ClientMQAdapter->>ClientMQAdapter: Drop message, log warning
else Message type = 'Request'
ClientMQAdapter->>MQWorker: processRequest(connectionId, type, payload, meta)
alt Request timeout (>15s)
MQWorker-->>ClientMQAdapter: TimeoutError
ClientMQAdapter->>ClientMQAdapter: Encode mq.Error {reason: 'timeout'}
ClientMQAdapter->>Client: Response (mq.Error)
else Request failed
MQWorker-->>ClientMQAdapter: Error
ClientMQAdapter->>ClientMQAdapter: Encode mq.Error {reason: 'failed'}
ClientMQAdapter->>Client: Response (mq.Error)
else Request succeeded
MQWorker-->>ClientMQAdapter: {type, payload, wallTime}
ClientMQAdapter->>ClientMQAdapter: Build Response message<br/>(messageType: 'Response', requestId, type, payload)
alt Socket not open
ClientMQAdapter->>ClientMQAdapter: Drop response, log debug
else Socket open
ClientMQAdapter->>Client: Response (binary or base64)
ClientMQAdapter->>ClientMQ: emit('request', connectionId, data)
ClientMQ->>ClientMQWorker: emit('request', connectionId, data)
ClientMQWorker->>ClientMQWorker: Handle subscription updates<br/>(JoinRoom, LeaveRoom, etc.)
end
end
else Message type = 'Response'
alt Request handler not found
ClientMQAdapter->>ClientMQAdapter: Log warning, ignore
else Request handler exists
ClientMQAdapter->>ClientMQAdapter: Resolve pending request promise
end
else Message type = 'Event'
ClientMQAdapter->>ClientMQAdapter: Log warning (unsupported)
end
Note over Client,ClientMQWorker: === Server-Initiated Request ===
ClientMQWorker->>ClientMQ: sendRequest(connectionId, type, message)
ClientMQ->>ClientMQAdapter: sendRequest(connectionId, type, message)
ClientMQAdapter->>ClientMQAdapter: Build request (requestId: 'S' + counter)
ClientMQAdapter->>ClientMQAdapter: Register in _requests map
alt Socket not found
ClientMQAdapter-->>ClientMQWorker: Error: 'Websocket connection not found'
else Socket not connected
ClientMQAdapter-->>ClientMQWorker: Error: 'Websocket connection not opened'
else Socket valid
ClientMQAdapter->>Client: Request (binary or base64)
ClientMQAdapter->>ClientMQAdapter: Start timeout (15s)
Client->>WebSocketServer: Response message
WebSocketServer->>ClientMQAdapter: messageDelegate (Response)
ClientMQAdapter->>ClientMQAdapter: Resolve request promise
ClientMQAdapter-->>ClientMQWorker: Response message
alt Response timeout
ClientMQAdapter->>ClientMQAdapter: Log warning, cleanup after 2x timeout
end
end
Note over Client,ClientMQWorker: === Heartbeat/Ping Flow ===
loop Every 20 seconds
PingInterval->>ClientMQAdapter: Ping interval trigger
ClientMQAdapter->>ClientMQAdapter: Iterate all sockets
alt Previous ping still pending
ClientMQAdapter->>ClientMQAdapter: Skip ping, log warning
else Socket has ping() method
alt No pong received (lastPong < lastPing)
ClientMQAdapter->>ClientMQAdapter: Detect timeout
ClientMQAdapter->>ClientMQ: emit('timeout', connectionId)
ClientMQ->>ClientMQWorker: emit('timeout', connectionId)
ClientMQWorker->>ClientMQ: close(connectionId, 3334, 'timeout')
Note over ClientMQAdapter: See "Server-Initiated Close" below
else Pong received
ClientMQAdapter->>Client: ping()
Client->>WebSocketServer: pong()
WebSocketServer->>ClientMQAdapter: pongDelegate(socket)
ClientMQAdapter->>ClientMQAdapter: Set __lastPong = Date.now()<br/>Calculate RTT
ClientMQAdapter->>MQWorker: processRoundTripTimeMeasurement(...)
end
else Socket has no ping() method
alt Last ping > 2x heartbeatInterval ago
ClientMQAdapter->>ClientMQAdapter: Detect timeout
ClientMQAdapter->>ClientMQ: emit('timeout', connectionId)
Note over ClientMQAdapter: See timeout flow above
end
end
end
Note over Client,ClientMQWorker: === Server-Initiated Close ===
alt Close reason: timeout
ClientMQWorker->>ClientMQ: close(connectionId, 3334, 'timeout')
else Close reason: drain
ClientMQWorker->>ClientMQ: stop(3333, 'drain')
ClientMQ->>ClientMQAdapter: close(connectionId, code, reason) [for each]
else Close reason: connection limit
ClientMQWorker->>ClientMQ: close(connectionId, 1013, 'concurrency-limit-exceeded')
else Close reason: heartbeat terminate
ClientMQWorker->>ClientMQ: close(connectionId, 3335, 'unexpected-close')
else Close reason: send error
ClientMQWorker->>ClientMQ: close(connectionId, 3335, 'unexpected-close')
end
ClientMQ->>ClientMQAdapter: close(connectionId, code, reason)
alt Socket not found
ClientMQAdapter->>ClientMQAdapter: Log warning, return false
else Connection already closed (>5s ago)
ClientMQAdapter->>ClientMQAdapter: Force disconnect
ClientMQAdapter->>ClientMQAdapter: Set __forcefullyClosed = true
ClientMQAdapter->>ClientMQ: emit('disconnect-after-forcefull-close')
else Connection valid
ClientMQAdapter->>ClientMQAdapter: Set __connectionEnded = now()
ClientMQAdapter->>Client: socket.close(code, reason)
Client->>WebSocketServer: close event
WebSocketServer->>ClientMQAdapter: disconnectDelegate(socket, code, description)
alt Multiple close events
ClientMQAdapter->>ClientMQAdapter: Log warning, return early
else Socket already disconnected
alt Previously forcefully closed
ClientMQAdapter->>ClientMQ: emit('disconnect-after-forcefull-close')
else Already disconnected
ClientMQAdapter->>ClientMQ: emit('disconnect-already-disconnected')
end
else Normal disconnect
ClientMQAdapter->>ClientMQAdapter: Remove from _socketsById
ClientMQAdapter->>ClientMQAdapter: Set __connected = false
ClientMQAdapter->>ClientMQAdapter: Calculate connection duration
ClientMQAdapter->>ClientMQ: emit('disconnect', connectionId, code, description)
ClientMQ->>ClientMQ: Remove from _adaptersById
ClientMQ->>ClientMQWorker: emit('disconnect', connectionId, code, description)
ClientMQWorker->>ClientMQWorker: Decrement _connectionsOpen
ClientMQWorker->>ClientMQWorker: Clean up room subscriptions
ClientMQWorker->>ClientMQWorker: Clean up conversation subscriptions
ClientMQWorker->>MQWorker: publish('pcast.ConnectionDisconnected', ...)
end
end
Note over Client,ClientMQWorker: === Client-Initiated Close ===
Client->>WebSocketServer: close connection
WebSocketServer->>WebSocketServer: close event (reasonCode, description)
WebSocketServer->>ClientMQAdapter: disconnectDelegate(socket, code, description)
Note over ClientMQAdapter: Same disconnect flow as above
Note over Client,ClientMQWorker: === Error Handling ===
alt Socket error
Client->>WebSocketServer: error event
WebSocketServer->>WebSocketServer: Log error
else Handler error
WebSocketServer->>WebSocketServer: Log error, continue
else Send error (not opened)
ClientMQWorker->>ClientMQ: close(connectionId, 3335, 'unexpected-close')
ClientMQWorker->>MQWorker: request('pcast.ConnectionDisconnected', ...)
Note over ClientMQAdapter: See "Server-Initiated Close" above
else Send error (not found/closed)
ClientMQWorker->>ClientMQWorker: Return {status: 'closed'}
end