496 lines
16 KiB
Rust
496 lines
16 KiB
Rust
use std::{sync::Arc, time::Duration};
|
|
|
|
use robotnik::{event::Event, event_manager::EventManager};
|
|
use rstest::rstest;
|
|
use tokio::{
|
|
io::{AsyncBufReadExt, BufReader},
|
|
net::UnixStream,
|
|
time::timeout,
|
|
};
|
|
|
|
const TEST_SOCKET_BASE: &str = "/tmp/robotnik_test";
|
|
|
|
/// Helper to create unique socket paths for parallel tests
|
|
fn test_socket_path(name: &str) -> String {
|
|
format!("{}_{}_{}", TEST_SOCKET_BASE, name, std::process::id())
|
|
}
|
|
|
|
/// Helper to read one JSON event from a stream
|
|
async fn read_event(
|
|
reader: &mut BufReader<UnixStream>,
|
|
) -> Result<Event, Box<dyn std::error::Error>> {
|
|
let mut line = String::new();
|
|
reader.read_line(&mut line).await?;
|
|
let event: Event = serde_json::from_str(&line)?;
|
|
Ok(event)
|
|
}
|
|
|
|
/// Helper to read all available events with a timeout
|
|
async fn read_events_with_timeout(
|
|
reader: &mut BufReader<UnixStream>,
|
|
max_count: usize,
|
|
timeout_ms: u64,
|
|
) -> Vec<String> {
|
|
let mut events = Vec::new();
|
|
for _ in 0..max_count {
|
|
let mut line = String::new();
|
|
match timeout(
|
|
Duration::from_millis(timeout_ms),
|
|
reader.read_line(&mut line),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(0)) => break, // EOF
|
|
Ok(Ok(_)) => events.push(line),
|
|
Ok(Err(_)) => break, // Read error
|
|
Err(_) => break, // Timeout
|
|
}
|
|
}
|
|
events
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_client_connects_and_receives_event() {
|
|
let socket_path = test_socket_path("basic_connect");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
// Give the listener time to start
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Broadcast an event
|
|
let event = Event::new("test_user", "test message");
|
|
manager.broadcast(&event).await.unwrap();
|
|
|
|
// Connect as a client
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader = BufReader::new(stream);
|
|
|
|
// Read the event
|
|
let mut line = String::new();
|
|
reader.read_line(&mut line).await.unwrap();
|
|
|
|
assert!(line.contains("test message"));
|
|
assert!(line.ends_with('\n'));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_client_receives_event_history() {
|
|
let socket_path = test_socket_path("event_history");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Broadcast events BEFORE starting the listener
|
|
for i in 0..5 {
|
|
let event = Event::new("test_user", format!("historical event {}", i));
|
|
manager.broadcast(&event).await.unwrap();
|
|
}
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect as a client
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader = BufReader::new(stream);
|
|
|
|
// Should receive all 5 historical events
|
|
let events = read_events_with_timeout(&mut reader, 5, 100).await;
|
|
|
|
assert_eq!(events.len(), 5);
|
|
assert!(events[0].contains("historical event 0"));
|
|
assert!(events[4].contains("historical event 4"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_multiple_clients_receive_same_events() {
|
|
let socket_path = test_socket_path("multiple_clients");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect 3 clients
|
|
let stream1 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let stream2 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let stream3 = UnixStream::connect(&socket_path).await.unwrap();
|
|
|
|
let mut reader1 = BufReader::new(stream1);
|
|
let mut reader2 = BufReader::new(stream2);
|
|
let mut reader3 = BufReader::new(stream3);
|
|
|
|
// Broadcast a new event
|
|
let event = Event::new("test_user", "broadcast to all");
|
|
manager.broadcast(&event).await.unwrap();
|
|
|
|
// All clients should receive the event
|
|
let mut line1 = String::new();
|
|
let mut line2 = String::new();
|
|
let mut line3 = String::new();
|
|
|
|
timeout(Duration::from_millis(100), reader1.read_line(&mut line1))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
timeout(Duration::from_millis(100), reader2.read_line(&mut line2))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
timeout(Duration::from_millis(100), reader3.read_line(&mut line3))
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
|
|
assert!(line1.contains("broadcast to all"));
|
|
assert!(line2.contains("broadcast to all"));
|
|
assert!(line3.contains("broadcast to all"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_late_joiner_receives_full_history() {
|
|
let socket_path = test_socket_path("late_joiner");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// First client connects
|
|
let stream1 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader1 = BufReader::new(stream1);
|
|
|
|
// Broadcast several events
|
|
for i in 0..10 {
|
|
let event = Event::new("test_user", format!("event {}", i));
|
|
manager.broadcast(&event).await.unwrap();
|
|
}
|
|
|
|
// Consume events from first client
|
|
let _ = read_events_with_timeout(&mut reader1, 10, 100).await;
|
|
|
|
// Late joiner connects
|
|
let stream2 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader2 = BufReader::new(stream2);
|
|
|
|
// Late joiner should receive all 10 events from history
|
|
let events = read_events_with_timeout(&mut reader2, 10, 100).await;
|
|
|
|
assert_eq!(events.len(), 10);
|
|
assert!(events[0].contains("event 0"));
|
|
assert!(events[9].contains("event 9"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_client_receives_events_in_order() {
|
|
let socket_path = test_socket_path("event_order");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect client
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader = BufReader::new(stream);
|
|
|
|
// Broadcast events rapidly
|
|
let count = 50;
|
|
for i in 0..count {
|
|
let event = Event::new("test_user", format!("sequence {}", i));
|
|
manager.broadcast(&event).await.unwrap();
|
|
}
|
|
|
|
// Read all events
|
|
let events = read_events_with_timeout(&mut reader, count, 500).await;
|
|
|
|
assert_eq!(events.len(), count);
|
|
|
|
// Verify order
|
|
for (i, event) in events.iter().enumerate() {
|
|
assert!(
|
|
event.contains(&format!("sequence {}", i)),
|
|
"Event {} out of order: {}",
|
|
i,
|
|
event
|
|
);
|
|
}
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_concurrent_broadcasts_during_client_connections() {
|
|
let socket_path = test_socket_path("concurrent_ops");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect client 1 BEFORE any broadcasts
|
|
let stream1 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader1 = BufReader::new(stream1);
|
|
|
|
// Spawn a task that continuously broadcasts
|
|
let broadcast_manager = Arc::clone(&manager);
|
|
let broadcast_handle = tokio::spawn(async move {
|
|
for i in 0..100 {
|
|
let event = Event::new("test_user", format!("concurrent event {}", i));
|
|
broadcast_manager.broadcast(&event).await.unwrap();
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
}
|
|
});
|
|
|
|
// While broadcasting, connect more clients at different times
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
let stream2 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader2 = BufReader::new(stream2);
|
|
|
|
tokio::time::sleep(Duration::from_millis(150)).await;
|
|
let stream3 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader3 = BufReader::new(stream3);
|
|
|
|
// Wait for broadcasts to complete
|
|
broadcast_handle.await.unwrap();
|
|
|
|
// All clients should have received events
|
|
let events1 = read_events_with_timeout(&mut reader1, 100, 200).await;
|
|
let events2 = read_events_with_timeout(&mut reader2, 100, 200).await;
|
|
let events3 = read_events_with_timeout(&mut reader3, 100, 200).await;
|
|
|
|
// Client 1 connected first (before any broadcasts), should get all 100
|
|
assert_eq!(events1.len(), 100);
|
|
|
|
// Client 2 connected after ~20 events were broadcast
|
|
// Gets ~20 from history + ~80 live = 100
|
|
assert_eq!(events2.len(), 100);
|
|
|
|
// Client 3 connected after ~50 events were broadcast
|
|
// Gets ~50 from history + ~50 live = 100
|
|
assert_eq!(events3.len(), 100);
|
|
|
|
// Verify they all received events in order
|
|
assert!(events1[0].contains("concurrent event 0"));
|
|
assert!(events2[0].contains("concurrent event 0"));
|
|
assert!(events3[0].contains("concurrent event 0"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_buffer_overflow_affects_new_clients() {
|
|
let socket_path = test_socket_path("buffer_overflow");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Broadcast more than buffer max (1000)
|
|
for i in 0..1100 {
|
|
let event = Event::new("test_user", format!("overflow event {}", i));
|
|
manager.broadcast(&event).await.unwrap();
|
|
}
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// New client connects
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader = BufReader::new(stream);
|
|
|
|
// Should receive exactly 1000 events (buffer max)
|
|
let events = read_events_with_timeout(&mut reader, 1100, 500).await;
|
|
|
|
assert_eq!(events.len(), 1000);
|
|
|
|
// First event should be 100 (oldest 100 were evicted)
|
|
assert!(events[0].contains("overflow event 100"));
|
|
|
|
// Last event should be 1099
|
|
assert!(events[999].contains("overflow event 1099"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[rstest]
|
|
#[case(10, 1)]
|
|
#[case(50, 5)]
|
|
#[tokio::test]
|
|
async fn test_client_count_scaling(#[case] num_clients: usize, #[case] events_per_client: usize) {
|
|
let socket_path = test_socket_path(&format!("scaling_{}_{}", num_clients, events_per_client));
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect many clients
|
|
let mut readers = Vec::new();
|
|
for _ in 0..num_clients {
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
readers.push(BufReader::new(stream));
|
|
}
|
|
|
|
// Broadcast events
|
|
for i in 0..events_per_client {
|
|
let event = Event::new("test_user", format!("scale event {}", i));
|
|
manager.broadcast(&event).await.unwrap();
|
|
}
|
|
|
|
// Verify all clients received all events
|
|
for reader in &mut readers {
|
|
let events = read_events_with_timeout(reader, events_per_client, 200).await;
|
|
assert_eq!(events.len(), events_per_client);
|
|
}
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_client_disconnect_doesnt_affect_others() {
|
|
let socket_path = test_socket_path("disconnect");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Connect 3 clients
|
|
let stream1 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let stream2 = UnixStream::connect(&socket_path).await.unwrap();
|
|
let stream3 = UnixStream::connect(&socket_path).await.unwrap();
|
|
|
|
let mut reader1 = BufReader::new(stream1);
|
|
let mut reader2 = BufReader::new(stream2);
|
|
let mut reader3 = BufReader::new(stream3);
|
|
|
|
// Broadcast initial event
|
|
manager
|
|
.broadcast(&Event::new("test_user", "before disconnect"))
|
|
.await
|
|
.unwrap();
|
|
|
|
// All receive it
|
|
let _ = read_events_with_timeout(&mut reader1, 1, 100).await;
|
|
let _ = read_events_with_timeout(&mut reader2, 1, 100).await;
|
|
let _ = read_events_with_timeout(&mut reader3, 1, 100).await;
|
|
|
|
// Drop client 2 (simulates disconnect)
|
|
drop(reader2);
|
|
|
|
// Broadcast another event
|
|
manager
|
|
.broadcast(&Event::new("test_user", "after disconnect"))
|
|
.await
|
|
.unwrap();
|
|
|
|
// Clients 1 and 3 should still receive it
|
|
let events1 = read_events_with_timeout(&mut reader1, 1, 100).await;
|
|
let events3 = read_events_with_timeout(&mut reader3, 1, 100).await;
|
|
|
|
assert_eq!(events1.len(), 1);
|
|
assert_eq!(events3.len(), 1);
|
|
assert!(events1[0].contains("after disconnect"));
|
|
assert!(events3[0].contains("after disconnect"));
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_json_deserialization_of_received_events() {
|
|
let socket_path = test_socket_path("json_deser");
|
|
let manager = Arc::new(EventManager::new().unwrap());
|
|
|
|
// Start the listener
|
|
let listener_manager = Arc::clone(&manager);
|
|
let socket_path_clone = socket_path.clone();
|
|
tokio::spawn(async move {
|
|
listener_manager.start_listening(socket_path_clone).await;
|
|
});
|
|
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
|
|
// Broadcast an event with special characters
|
|
let test_message = "special chars: @#$% newline\\n tab\\t quotes \"test\"";
|
|
manager
|
|
.broadcast(&Event::new("test_user", test_message))
|
|
.await
|
|
.unwrap();
|
|
|
|
// Connect and deserialize
|
|
let stream = UnixStream::connect(&socket_path).await.unwrap();
|
|
let mut reader = BufReader::new(stream);
|
|
|
|
let mut line = String::new();
|
|
reader.read_line(&mut line).await.unwrap();
|
|
|
|
// Should be valid JSON
|
|
let parsed: serde_json::Value = serde_json::from_str(line.trim()).unwrap();
|
|
|
|
assert_eq!(parsed["message"], test_message);
|
|
|
|
// Cleanup
|
|
let _ = std::fs::remove_file(&socket_path);
|
|
}
|