298 lines
10 KiB
Rust
298 lines
10 KiB
Rust
use citygen_server::{
|
|
gateway::{router, App, Config},
|
|
protocol::{self as p, client_message, server_message},
|
|
};
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use prost::Message;
|
|
use std::{io::Write, time::Duration};
|
|
|
|
fn fixture(dir: &std::path::Path) {
|
|
let folder = dir.join("cities/demo/tiles");
|
|
std::fs::create_dir_all(&folder).unwrap();
|
|
std::fs::write(dir.join("cities/demo/manifest.json"),r#"{"formatVersion":1,"city":{"id":"demo","name":"Demo","bounds":{"minX":0,"minY":0,"maxX":100,"maxY":100}},"tiles":[{"id":"0_0","url":"tiles/0_0.json.gz","bounds":{"minX":0,"minY":0,"maxX":256,"maxY":256}}]}"#).unwrap();
|
|
let file = std::fs::File::create(folder.join("0_0.json.gz")).unwrap();
|
|
let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::default());
|
|
encoder.write_all(br#"{"formatVersion":1,"cityId":"demo","id":"0_0","buildings":[],"roads":[{"id":"road","kind":"residential","pts":[[0,0],[100,0],[100,100]]}]}"#).unwrap();
|
|
encoder.finish().unwrap();
|
|
}
|
|
async fn read(
|
|
socket: &mut tokio_tungstenite::WebSocketStream<
|
|
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
|
|
>,
|
|
) -> p::ServerMessage {
|
|
loop {
|
|
let frame = tokio::time::timeout(Duration::from_secs(5), socket.next())
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.unwrap();
|
|
if let tokio_tungstenite::tungstenite::Message::Binary(bytes) = frame {
|
|
return p::ServerMessage::decode(&bytes[..]).unwrap();
|
|
}
|
|
}
|
|
}
|
|
async fn http_request(
|
|
address: std::net::SocketAddr,
|
|
path: &str,
|
|
method: &str,
|
|
token: &str,
|
|
) -> String {
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
let mut socket = tokio::net::TcpStream::connect(address).await.unwrap();
|
|
socket.write_all(format!("{method} {path} HTTP/1.1\r\nHost: localhost\r\nAuthorization: Bearer {token}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
|
|
let mut response = String::new();
|
|
socket.read_to_string(&mut response).await.unwrap();
|
|
response
|
|
}
|
|
#[tokio::test]
|
|
async fn thirty_observers_snapshot_delta_vote_and_restart() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
fixture(dir.path());
|
|
let app = App::new(Config {
|
|
cities: dir.path().join("cities"),
|
|
saves: dir.path().join("rooms"),
|
|
admin_token: Some("test-only-administrator-token-32-chars".into()),
|
|
..Default::default()
|
|
})
|
|
.unwrap();
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let address = listener.local_addr().unwrap();
|
|
let serve_app = app.clone();
|
|
let server = tokio::spawn(async move {
|
|
axum::serve(
|
|
listener,
|
|
router(serve_app).into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
});
|
|
// Minimal HTTP request avoids adding a second HTTP client dependency.
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
let mut http = tokio::net::TcpStream::connect(address).await.unwrap();
|
|
let body = r#"{"cityId":"demo","seed":1}"#;
|
|
http.write_all(format!("POST /api/rooms HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",body.len()).as_bytes()).await.unwrap();
|
|
let mut response = String::new();
|
|
http.read_to_string(&mut response).await.unwrap();
|
|
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
|
|
let value: serde_json::Value =
|
|
serde_json::from_str(response.split("\r\n\r\n").nth(1).unwrap()).unwrap();
|
|
let id = value["id"].as_str().unwrap().to_string();
|
|
assert!(http_request(address, "/api/admin", "GET", "wrong")
|
|
.await
|
|
.starts_with("HTTP/1.1 401"));
|
|
assert!(http_request(
|
|
address,
|
|
"/api/admin",
|
|
"GET",
|
|
"test-only-administrator-token-32-chars"
|
|
)
|
|
.await
|
|
.starts_with("HTTP/1.1 200"));
|
|
let checkpoint = http_request(
|
|
address,
|
|
&format!("/api/admin/rooms/{id}/checkpoint"),
|
|
"POST",
|
|
"test-only-administrator-token-32-chars",
|
|
)
|
|
.await;
|
|
assert!(checkpoint.starts_with("HTTP/1.1 200"), "{checkpoint}");
|
|
assert!(citygen_server::storage::load(&dir.path().join("rooms"), &id).is_ok());
|
|
let mut sockets = vec![];
|
|
for i in 0..30 {
|
|
let (mut socket, _) =
|
|
tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
|
|
.await
|
|
.unwrap();
|
|
let join = p::ClientMessage {
|
|
body: Some(client_message::Body::Join(p::Join {
|
|
room: id.clone(),
|
|
client_version: 1,
|
|
nickname: format!("Player{i}"),
|
|
})),
|
|
};
|
|
socket
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
join.encode_to_vec().into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert!(matches!(
|
|
read(&mut socket).await.body,
|
|
Some(server_message::Body::Welcome(_))
|
|
));
|
|
sockets.push(socket);
|
|
}
|
|
let s = &mut sockets[0];
|
|
let subscribe = p::ClientMessage {
|
|
body: Some(client_message::Body::Subscribe(p::Subscribe {
|
|
rect: Some(p::Rect {
|
|
min_x: -100,
|
|
min_y: -100,
|
|
max_x: 20000,
|
|
max_y: 20000,
|
|
}),
|
|
lod: 0,
|
|
})),
|
|
};
|
|
s.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
subscribe.encode_to_vec().into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
loop {
|
|
if let Some(server_message::Body::Snapshot(snapshot)) = read(s).await.body {
|
|
assert_eq!(snapshot.entities.len(), 1000);
|
|
break;
|
|
}
|
|
}
|
|
loop {
|
|
if matches!(read(s).await.body, Some(server_message::Body::Delta(_))) {
|
|
break;
|
|
}
|
|
}
|
|
// 31st observer must be rejected without affecting existing players.
|
|
let (mut extra, _) =
|
|
tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
|
|
.await
|
|
.unwrap();
|
|
extra
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
p::ClientMessage {
|
|
body: Some(client_message::Body::Join(p::Join {
|
|
room: id.clone(),
|
|
client_version: 1,
|
|
nickname: "extra".into(),
|
|
})),
|
|
}
|
|
.encode_to_vec()
|
|
.into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
matches!(read(&mut extra).await.body,Some(server_message::Body::Error(e)) if e.code=="ROOM_FULL")
|
|
);
|
|
let proposal = p::ClientMessage {
|
|
body: Some(client_message::Body::Propose(p::Propose { speed: 0 })),
|
|
};
|
|
sockets[0]
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
proposal.encode_to_vec().into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
let yes = p::ClientMessage {
|
|
body: Some(client_message::Body::Vote(p::Vote { yes: true })),
|
|
}
|
|
.encode_to_vec();
|
|
for socket in sockets.iter_mut().take(15).skip(1) {
|
|
socket
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
yes.clone().into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
}
|
|
loop {
|
|
if let Some(server_message::Body::Clock(clock)) = read(&mut sockets[0]).await.body {
|
|
assert_eq!(clock.speed, 1, "15 of 30 is not a majority");
|
|
break;
|
|
}
|
|
}
|
|
sockets[15]
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(yes.into()))
|
|
.await
|
|
.unwrap();
|
|
loop {
|
|
if let Some(server_message::Body::Clock(clock)) = read(&mut sockets[0]).await.body {
|
|
if clock.speed == 0 {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
for socket in &mut sockets {
|
|
let _ = socket.close(None).await;
|
|
}
|
|
app.stop();
|
|
server.abort();
|
|
let saved = citygen_server::storage::load(&dir.path().join("rooms"), &id).unwrap();
|
|
assert_eq!(saved.agents.len(), 1000);
|
|
assert_eq!(saved.speed, 0);
|
|
assert!(!saved.city_revision.is_empty());
|
|
let mut expected = citygen_server::simulation::Simulation::restore(saved.clone()).unwrap();
|
|
let expected_entities = expected
|
|
.visible(
|
|
&p::Rect {
|
|
min_x: -100,
|
|
min_y: -100,
|
|
max_x: 20000,
|
|
max_y: 20000,
|
|
},
|
|
1000,
|
|
)
|
|
.unwrap();
|
|
let restarted = App::new(Config {
|
|
cities: dir.path().join("cities"),
|
|
saves: dir.path().join("rooms"),
|
|
..Default::default()
|
|
})
|
|
.unwrap();
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let address = listener.local_addr().unwrap();
|
|
let serve_app = restarted.clone();
|
|
let server = tokio::spawn(async move {
|
|
axum::serve(
|
|
listener,
|
|
router(serve_app).into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
});
|
|
let (mut socket, _) =
|
|
tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
|
|
.await
|
|
.unwrap();
|
|
socket
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
p::ClientMessage {
|
|
body: Some(client_message::Body::Join(p::Join {
|
|
room: id.clone(),
|
|
client_version: 1,
|
|
nickname: "Restored".into(),
|
|
})),
|
|
}
|
|
.encode_to_vec()
|
|
.into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert!(http_request(
|
|
address,
|
|
"/api/admin",
|
|
"GET",
|
|
"test-only-administrator-token-32-chars"
|
|
)
|
|
.await
|
|
.starts_with("HTTP/1.1 404"));
|
|
assert!(
|
|
matches!(read(&mut socket).await.body,Some(server_message::Body::Welcome(w)) if w.tick>=saved.tick)
|
|
);
|
|
socket
|
|
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
|
subscribe.encode_to_vec().into(),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
loop {
|
|
if let Some(server_message::Body::Snapshot(snapshot)) = read(&mut socket).await.body {
|
|
let entities: std::collections::HashMap<_, _> =
|
|
snapshot.entities.into_iter().map(|a| (a.id, a)).collect();
|
|
assert_eq!(entities, expected_entities);
|
|
break;
|
|
}
|
|
}
|
|
let _ = socket.close(None).await;
|
|
restarted.stop();
|
|
server.abort();
|
|
}
|