# WebSocket and Server-Sent Events Guide
This guide explains how to use WebSockets and Server-Sent Events (SSE) in Armature for real-time communication.
## Overview
Armature provides built-in support for two real-time communication patterns:
1. **WebSockets** - Full-duplex, bidirectional communication
2. **Server-Sent Events (SSE)** - Server-to-client streaming
## WebSocket Support
### Core Components
#### WebSocketMessage
Enum representing different message types:
```rust
pub enum WebSocketMessage {
Text(String),
Binary(Vec<u8>),
Ping(Vec<u8>),
Pong(Vec<u8>),
Close,
}
```
#### WebSocketConnection
Handle for a single WebSocket connection:
```rust
let (connection, receiver) = WebSocketConnection::new("conn-123".to_string());
// Send text message
connection.send_text("Hello!".to_string()).await?;
// Send JSON
connection.send_json(&data).await?;
```
#### WebSocketRoom
Broadcast to multiple connections in a room:
```rust
let room = WebSocketRoom::new("chat-room".to_string());
// Add connection
room.add_connection(id, tx).await;
// Broadcast to all
room.broadcast_text("Hello everyone!".to_string()).await?;
room.broadcast_json(&data).await?;
// Get connection count
let count = room.connection_count().await;
```
#### WebSocketManager
Manage multiple rooms:
```rust
let manager = WebSocketManager::new();
// Get or create room
let room = manager.get_or_create_room("lobby").await;
// Get existing room
if let Some(room) = manager.get_room("lobby").await {
// Use room
}
// Remove room
manager.remove_room("lobby").await;
```
### Usage Example
#### 1. Create a WebSocket Service
```rust
#[injectable]
#[derive(Clone)]
struct ChatService {
manager: Arc<WebSocketManager>,
}
impl Default for ChatService {
fn default() -> Self {
Self {
manager: Arc::new(WebSocketManager::new()),
}
}
}
impl ChatService {
async fn get_room(&self, name: &str) -> Arc<WebSocketRoom> {
self.manager.get_or_create_room(name).await
}
async fn broadcast(&self, room: &str, msg: &str) -> Result<(), Error> {
let room = self.get_room(room).await;
room.broadcast_text(msg.to_string()).await
}
}
```
#### 2. Create a WebSocket Controller
```rust
#[controller("/ws")]
#[derive(Default, Clone)]
struct WebSocketController {
chat_service: ChatService,
}
impl WebSocketController {
async fn handle_chat(&self, room_name: String) -> Result<(), Error> {
let room = self.chat_service.get_room(&room_name).await;
// In full implementation, this would:
// 1. Upgrade HTTP connection to WebSocket
// 2. Handle incoming messages
// 3. Broadcast to room
Ok(())
}
}
```
#### 3. Register in Module
```rust
#[module(
providers: [ChatService],
controllers: [WebSocketController]
)]
#[derive(Default)]
struct AppModule;
```
### Broadcasting Patterns
#### Room-based Broadcasting
```rust
// Send to specific room
let room = manager.get_or_create_room("room-1").await;
room.broadcast_json(&message).await?;
```
#### Targeted Messaging
```rust
// Send to specific connection
let (connection, _rx) = WebSocketConnection::new(id);
connection.send_text("Private message".to_string()).await?;
```
#### Global Broadcasting
```rust
// Broadcast to all rooms
for room_name in ["room-1", "room-2", "room-3"] {
if let Some(room) = manager.get_room(room_name).await {
room.broadcast_json(&message).await?;
}
}
```
## Server-Sent Events (SSE)
### Core Components
#### ServerSentEvent
Represents an SSE event:
```rust
// Simple message
let event = ServerSentEvent::new("Hello".to_string());
// Typed event
let event = ServerSentEvent::with_event(
"user_joined".to_string(),
"Alice joined".to_string()
);
// Full event
let event = ServerSentEvent::full(
"123".to_string(), // ID
"notification".to_string(), // Event type
"You have mail".to_string(), // Data
5000 // Retry (ms)
);
```
#### SseStream
Single client SSE stream:
```rust
let (stream, receiver) = SseStream::new();
// Send events
stream.send_message("Hello".to_string()).await?;
stream.send_event("update".to_string(), "Data".to_string()).await?;
stream.send_json(&data).await?;
// Keep-alive
stream.send_keep_alive().await?;
```
#### SseBroadcaster
Broadcast SSE to multiple clients:
```rust
let broadcaster = Arc::new(SseBroadcaster::new());
// Register clients
let stream1 = broadcaster.register().await;
let stream2 = broadcaster.register().await;
// Broadcast to all
broadcaster.broadcast_message("Update".to_string()).await?;
broadcaster.broadcast_json(&data).await?;
// Get client count
let count = broadcaster.client_count().await;
// Auto keep-alive
broadcaster.clone().start_keep_alive(Duration::from_secs(30));
```
### Usage Example
#### 1. Create an SSE Service
```rust
#[injectable]
#[derive(Clone)]
struct StockTickerService {
broadcaster: Arc<SseBroadcaster>,
}
impl Default for StockTickerService {
fn default() -> Self {
let broadcaster = Arc::new(SseBroadcaster::new());
// Start broadcasting stock prices
let broadcaster_clone = broadcaster.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(1));
loop {
interval.tick().await;
let price = StockPrice {
symbol: "AAPL".to_string(),
price: 150.0,
};
let event = ServerSentEvent::with_event(
"price".to_string(),
serde_json::to_string(&price).unwrap(),
);
let _ = broadcaster_clone.broadcast(event).await;
}
});
// Start keep-alive
broadcaster.clone().start_keep_alive(Duration::from_secs(15));
Self { broadcaster }
}
}
```
#### 2. Create an SSE Controller
```rust
#[controller("/events")]
#[derive(Default, Clone)]
struct EventsController {
stock_ticker: StockTickerService,
}
impl EventsController {
async fn subscribe_stocks(&self) -> ReceiverStream<Result<String, Error>> {
self.stock_ticker.broadcaster.register().await
}
}
```
### Client-Side Usage
#### JavaScript
```javascript
// Connect to SSE endpoint
const source = new EventSource('http://localhost:3000/events/stocks');
// Listen for specific events
source.addEventListener('price', (event) => {
const data = JSON.parse(event.data);
console.log('Stock price:', data);
});
// Listen for all events
source.onmessage = (event) => {
console.log('Message:', event.data);
};
// Handle errors
source.onerror = (error) => {
console.error('SSE error:', error);
};
// Close connection
source.close();
```
#### curl
```bash
# Subscribe to SSE stream (-N for no buffering)
curl -N http://localhost:3000/events/stocks
```
## Comparison: WebSocket vs SSE
### Use WebSockets When:
✅ You need bidirectional communication
✅ Low latency is critical
✅ Binary data transfer is required
✅ Building chat applications
✅ Real-time collaborative editing
✅ Gaming or interactive applications
### Use SSE When:
✅ Server → Client communication only
✅ Text-based data is sufficient
✅ Simple implementation needed
✅ Automatic reconnection desired
✅ Event streaming (stocks, notifications)
✅ Progress updates
✅ Live feeds
## Best Practices
### WebSocket Best Practices
1. **Connection Management**
```rust
let mut connections = HashMap::new();
connections.remove(&conn_id);
```
2. **Error Handling**
```rust
if let Err(e) = connection.send_text(msg).await {
eprintln!("Failed to send: {}", e);
}
```
3. **Rate Limiting**
```rust
let mut last_message = Instant::now();
if last_message.elapsed() < Duration::from_millis(100) {
return Err(Error::Validation("Too many messages".into()));
}
```
4. **Authentication**
```rust
fn validate_connection(req: &HttpRequest) -> Result<UserId, Error> {
let token = req.headers.get("Authorization")?;
verify_token(token)
}
```
### SSE Best Practices
1. **Keep-Alive**
```rust
broadcaster.start_keep_alive(Duration::from_secs(30));
```
2. **Event IDs**
```rust
let mut id = 0;
loop {
id += 1;
let event = ServerSentEvent {
id: Some(id.to_string()),
event: Some("update".into()),
data: "...".into(),
retry: None,
};
}
```
3. **Retry Configuration**
```rust
let event = ServerSentEvent {
retry: Some(5000), ..Default::default()
};
```
4. **Clean Up Disconnected Clients**
```rust
let mut clients = broadcaster.clients.write().await;
clients.retain(|tx| !tx.is_closed());
```
## Performance Considerations
### WebSocket Performance
- **Memory:** ~2KB per connection
- **CPU:** Minimal when idle
- **Latency:** < 1ms typically
- **Throughput:** Depends on message size
### SSE Performance
- **Memory:** ~1KB per client
- **CPU:** Low overhead
- **Latency:** 1-2ms typically
- **Throughput:** Limited by HTTP/1.1
### Scaling
#### Horizontal Scaling
Use Redis pub/sub for multi-server setups:
```rust
// Subscribe to Redis channel
let mut pubsub = redis_client.get_async_connection().await?;
pubsub.subscribe("chat:room1").await?;
// Forward to WebSocket/SSE
while let Some(msg) = pubsub.on_message().next().await {
room.broadcast_text(msg.get_payload()).await?;
}
```
#### Connection Limits
```rust
// Limit connections per room
const MAX_CONNECTIONS: usize = 1000;
if room.connection_count().await >= MAX_CONNECTIONS {
return Err(Error::Internal("Room full".into()));
}
```
## Testing
### Unit Testing
```rust
#[tokio::test]
async fn test_websocket_broadcast() {
let room = WebSocketRoom::new("test".to_string());
let (conn1, _) = WebSocketConnection::new("1".to_string());
let (conn2, _) = WebSocketConnection::new("2".to_string());
// Test broadcast
room.broadcast_text("Hello".to_string()).await.unwrap();
assert_eq!(room.connection_count().await, 2);
}
```
### Integration Testing
```rust
#[tokio::test]
async fn test_sse_stream() {
let broadcaster = Arc::new(SseBroadcaster::new());
let mut stream = broadcaster.register().await;
// Broadcast message
broadcaster.broadcast_message("Test".to_string()).await.unwrap();
// Receive message
let msg = stream.next().await.unwrap().unwrap();
assert!(msg.contains("Test"));
}
```
## Troubleshooting
### WebSocket Issues
**Connection fails:**
- Check HTTP → WebSocket upgrade
- Verify protocol headers
- Check firewall/proxy settings
**Messages not received:**
- Verify connection is open
- Check error logs
- Test with simple echo server
### SSE Issues
**Stream disconnects:**
- Implement keep-alive
- Check server timeout settings
- Verify HTTP/1.1 support
**Messages delayed:**
- Disable response buffering
- Use `Content-Type: text/event-stream`
- Check network conditions
## Future Enhancements
Planned features:
- [ ] `#[websocket]` decorator for automatic upgrade
- [ ] `#[sse]` decorator for streaming responses
- [ ] Built-in authentication middleware
- [ ] Rate limiting decorators
- [ ] Redis adapter for clustering
- [ ] Compression support
- [ ] Binary WebSocket frames
- [ ] GraphQL subscription support
## Summary
Armature provides comprehensive real-time communication support:
✅ **WebSocket** - Full-duplex communication with rooms and broadcasting
✅ **SSE** - Efficient server-to-client streaming
✅ **Type-Safe** - Compile-time verified message types
✅ **Easy to Use** - Simple, intuitive APIs
✅ **Scalable** - Designed for production use
✅ **Well-Tested** - Comprehensive test coverage
Both WebSocket and SSE integrate seamlessly with Armature's dependency injection system, making it easy to build real-time features into your applications.