Implement tiled multiplayer city simulation and scaling stages
This commit is contained in:
mrleo1nid committed
1 parent
3d4b29fedf
commit
3b1fcd996a
53 files changed
+7354
-13
No files matched your search
@@ -0,0 +1,121 @@
|
||||
use citygen_server::{
|
||||
gateway::{router, App, Config},
|
||||
generation::{cache_id, Request},
|
||||
protocol::{CityJob, CreateCity},
|
||||
};
|
||||
use prost::Message;
|
||||
use std::time::Duration;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
async fn request(
|
||||
address: std::net::SocketAddr,
|
||||
method: &str,
|
||||
path: &str,
|
||||
body: &[u8],
|
||||
) -> (u16, Vec<u8>) {
|
||||
let mut stream = tokio::net::TcpStream::connect(address).await.unwrap();
|
||||
stream.write_all(format!("{method} {path} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/x-protobuf\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",body.len()).as_bytes()).await.unwrap();
|
||||
stream.write_all(body).await.unwrap();
|
||||
let mut bytes = vec![];
|
||||
stream.read_to_end(&mut bytes).await.unwrap();
|
||||
let split = bytes.windows(4).position(|b| b == b"\r\n\r\n").unwrap();
|
||||
let status = std::str::from_utf8(&bytes[..split])
|
||||
.unwrap()
|
||||
.split_whitespace()
|
||||
.nth(1)
|
||||
.unwrap()
|
||||
.parse()
|
||||
.unwrap();
|
||||
(status, bytes[split + 4..].to_vec())
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn isolated_typescript_worker_exports_city_and_opens_room_without_network() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let fixture = dir.path().join("osm.json");
|
||||
std::fs::write(&fixture,r#"{"elements":[{"type":"way","id":1,"tags":{"highway":"residential"},"geometry":[{"lat":47.001,"lon":9.0},{"lat":47.001,"lon":9.004}]},{"type":"way","id":2,"tags":{"building":"house","building:levels":"1"},"geometry":[{"lat":47.0011,"lon":9.001},{"lat":47.0011,"lon":9.0012},{"lat":47.0013,"lon":9.0012},{"lat":47.0013,"lon":9.001},{"lat":47.0011,"lon":9.001}]},{"type":"way","id":3,"tags":{"building":"industrial","building:levels":"1"},"geometry":[{"lat":47.0006,"lon":9.002},{"lat":47.0006,"lon":9.0023},{"lat":47.0009,"lon":9.0023},{"lat":47.0009,"lon":9.002},{"lat":47.0006,"lon":9.002}]}]}"#).unwrap();
|
||||
let app = App::new(Config {
|
||||
cities: dir.path().join("cities"),
|
||||
saves: dir.path().join("rooms"),
|
||||
osm_fixture: Some(fixture),
|
||||
project_root: std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.parent()
|
||||
.unwrap()
|
||||
.to_owned(),
|
||||
..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();
|
||||
});
|
||||
let bbox = [47., 9., 47.002, 9.004];
|
||||
let message = CreateCity {
|
||||
bbox: bbox.to_vec(),
|
||||
name: "Test town".into(),
|
||||
seed: 1,
|
||||
};
|
||||
let (status, body) = request(
|
||||
address,
|
||||
"POST",
|
||||
"/api/cities/jobs",
|
||||
&message.encode_to_vec(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, 200, "{}", String::from_utf8_lossy(&body));
|
||||
let job = CityJob::decode(&body[..]).unwrap();
|
||||
assert_eq!(job.city_id, cache_id(bbox, 1));
|
||||
let ready = tokio::time::timeout(Duration::from_secs(30), async {
|
||||
loop {
|
||||
let (status, body) =
|
||||
request(address, "GET", &format!("/api/cities/jobs/{}", job.id), &[]).await;
|
||||
assert_eq!(status, 200);
|
||||
let state = CityJob::decode(&body[..]).unwrap();
|
||||
assert_ne!(state.status, "failed", "{}", state.error);
|
||||
if state.status == "ready" {
|
||||
break state;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(!ready.room_id.is_empty());
|
||||
let city = citygen_server::city::City::load(&app.config.cities, &ready.city_id).unwrap();
|
||||
assert!(!city.graph.points.is_empty());
|
||||
assert!(city.manifest.navigation.is_some());
|
||||
let duplicate = app
|
||||
.jobs
|
||||
.submit(
|
||||
app.clone(),
|
||||
Request {
|
||||
bbox: [47.0000001, 9., 47.002, 9.004],
|
||||
name: "Same area".into(),
|
||||
seed: 1,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(duplicate.id, job.id);
|
||||
let (status, _) = request(
|
||||
address,
|
||||
"POST",
|
||||
"/api/cities/jobs",
|
||||
&message.encode_to_vec(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, 429);
|
||||
app.stop();
|
||||
server.abort();
|
||||
assert_eq!(
|
||||
citygen_server::storage::load(&app.config.saves, &ready.room_id)
|
||||
.unwrap()
|
||||
.city_id,
|
||||
ready.city_id
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,255 @@
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
#[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"),
|
||||
..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();
|
||||
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!(
|
||||
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();
|
||||
}
|
||||
Reference in new issue
Block a user