baby steps towards a working bot api

This commit is contained in:
Ilion Beyst 2022-06-05 21:22:38 +02:00
parent c3d32e051c
commit 90ecb13a17
4 changed files with 165 additions and 19 deletions

View file

@ -7,6 +7,7 @@ edition = "2021"
[dependencies] [dependencies]
tokio = { version = "1.15", features = ["full"] } tokio = { version = "1.15", features = ["full"] }
tokio-stream = "0.1.9"
prost = "0.10" prost = "0.10"
tonic = "0.7.2" tonic = "0.7.2"

View file

@ -2,20 +2,34 @@ pub mod pb {
tonic::include_proto!("grpc.planetwars.bot_api"); tonic::include_proto!("grpc.planetwars.bot_api");
} }
use pb::test_service_client::TestServiceClient; use pb::bot_api_service_client::BotApiServiceClient;
use pb::{Hello, HelloResponse}; use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic::Response;
use tokio::sync::mpsc;
#[tokio::main] #[tokio::main]
async fn main() { async fn main() {
let mut client = TestServiceClient::connect("http://localhost:50051") let mut client = BotApiServiceClient::connect("http://localhost:50051")
.await .await
.unwrap(); .unwrap();
let response: Response<HelloResponse> = client
.greet(Hello { let (tx, rx) = mpsc::unbounded_channel();
hello_message: "robbe".to_string(), let mut stream = client
.connect_bot(UnboundedReceiverStream::new(rx))
.await
.unwrap()
.into_inner();
while let Some(message) = stream.message().await.unwrap() {
let state = String::from_utf8(message.content).unwrap();
println!("{}", state);
let response = r#"{ moves: [] }"#;
tx.send(pb::PlayerRequestResponse {
request_id: message.request_id,
content: response.as_bytes().to_vec(),
}) })
.await
.unwrap(); .unwrap();
println!("{}", response.get_ref().response); }
std::mem::drop(tx);
// for clean exit
std::mem::drop(client);
} }

View file

@ -7,6 +7,7 @@ edition = "2021"
[dependencies] [dependencies]
tokio = { version = "1.15", features = ["full"] } tokio = { version = "1.15", features = ["full"] }
tokio-stream = "0.1.9"
hyper = "0.14" hyper = "0.14"
axum = { version = "0.4", features = ["json", "headers", "multipart"] } axum = { version = "0.4", features = ["json", "headers", "multipart"] }
diesel = { version = "1.4.4", features = ["postgres", "chrono"] } diesel = { version = "1.4.4", features = ["postgres", "chrono"] }

View file

@ -3,27 +3,157 @@ pub mod pb {
} }
use std::net::SocketAddr; use std::net::SocketAddr;
use std::ops::DerefMut;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use runner::match_context::{EventBus, PlayerHandle, RequestMessage};
use runner::match_log::MatchLogger;
use tokio::sync::{mpsc, oneshot};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic; use tonic;
use tonic::transport::Server; use tonic::transport::Server;
use tonic::{Request, Response, Status}; use tonic::{Request, Response, Status, Streaming};
pub struct BotApiServer {} use planetwars_matchrunner as runner;
use crate::db;
use crate::{ConnectionPool, MAPS_DIR, MATCHES_DIR};
use super::matches::code_bundle_to_botspec;
pub struct BotApiServer {
sync_thing: ServerSyncThing,
}
#[tonic::async_trait] #[tonic::async_trait]
impl pb::test_service_server::TestService for BotApiServer { impl pb::bot_api_service_server::BotApiService for BotApiServer {
async fn greet(&self, req: Request<pb::Hello>) -> Result<Response<pb::HelloResponse>, Status> { type ConnectBotStream = UnboundedReceiverStream<Result<pb::PlayerRequest, Status>>;
Ok(Response::new(pb::HelloResponse {
response: format!("hallo {}", req.get_ref().hello_message), async fn connect_bot(
})) &self,
req: Request<Streaming<pb::PlayerRequestResponse>>,
) -> Result<Response<Self::ConnectBotStream>, Status> {
println!("bot connected");
let stream = req.into_inner();
let sync_data = self.sync_thing.streams.lock().unwrap().take().unwrap();
sync_data.tx.send(stream).unwrap();
Ok(Response::new(UnboundedReceiverStream::new(
sync_data.server_messages,
)))
} }
} }
pub async fn run_bot_api() { #[derive(Clone)]
let server = BotApiServer {}; struct ServerSyncThing {
streams: Arc<Mutex<Option<SyncThingData>>>,
}
struct SyncThingData {
tx: oneshot::Sender<Streaming<pb::PlayerRequestResponse>>,
server_messages: mpsc::UnboundedReceiver<Result<pb::PlayerRequest, Status>>,
}
impl ServerSyncThing {
fn new() -> Self {
ServerSyncThing {
streams: Arc::new(Mutex::new(None)),
}
}
}
struct RemoteBotSpec {
sync_thing: ServerSyncThing,
}
#[tonic::async_trait]
impl runner::BotSpec for RemoteBotSpec {
async fn run_bot(
&self,
player_id: u32,
event_bus: Arc<Mutex<EventBus>>,
_match_logger: MatchLogger,
) -> Box<dyn PlayerHandle> {
let (tx, rx) = oneshot::channel();
let (server_msg_snd, server_msg_recv) = mpsc::unbounded_channel();
*self.sync_thing.streams.lock().unwrap().deref_mut() = Some(SyncThingData {
tx,
server_messages: server_msg_recv,
});
let client_messages = rx.await.unwrap();
tokio::spawn(handle_bot_messages(player_id, event_bus, client_messages));
Box::new(RemoteBotHandle {
sender: server_msg_snd,
})
}
}
async fn handle_bot_messages(
player_id: u32,
event_bus: Arc<Mutex<EventBus>>,
mut messages: Streaming<pb::PlayerRequestResponse>,
) {
while let Some(message) = messages.message().await.unwrap() {
let request_id = (player_id, message.request_id as u32);
event_bus
.lock()
.unwrap()
.resolve_request(request_id, Ok(message.content));
}
}
struct RemoteBotHandle {
sender: mpsc::UnboundedSender<Result<pb::PlayerRequest, Status>>,
}
impl PlayerHandle for RemoteBotHandle {
fn send_request(&mut self, r: RequestMessage) {
self.sender
.send(Ok(pb::PlayerRequest {
request_id: r.request_id as i32,
content: r.content,
}))
.unwrap();
}
}
async fn run_match(sync_thing: ServerSyncThing, pool: ConnectionPool) {
let conn = pool.get().await.unwrap();
let opponent = db::bots::find_bot_by_name("simplebot", &conn).unwrap();
let opponent_code_bundle = db::bots::active_code_bundle(opponent.id, &conn).unwrap();
let log_file_name = "remote_match.log";
let remote_bot_spec = RemoteBotSpec { sync_thing };
let match_config = runner::MatchConfig {
map_path: PathBuf::from(MAPS_DIR).join("hex.json"),
map_name: "hex".to_string(),
log_path: PathBuf::from(MATCHES_DIR).join(&log_file_name),
players: vec![
runner::MatchPlayer {
bot_spec: Box::new(remote_bot_spec),
},
runner::MatchPlayer {
bot_spec: code_bundle_to_botspec(&opponent_code_bundle),
},
],
};
runner::run_match(match_config).await;
}
pub async fn run_bot_api(pool: ConnectionPool) {
let sync_thing = ServerSyncThing::new();
tokio::spawn(run_match(sync_thing.clone(), pool));
let server = BotApiServer { sync_thing };
let addr = SocketAddr::from(([127, 0, 0, 1], 50051)); let addr = SocketAddr::from(([127, 0, 0, 1], 50051));
Server::builder() Server::builder()
.add_service(pb::test_service_server::TestServiceServer::new(server)) .add_service(pb::bot_api_service_server::BotApiServiceServer::new(server))
.serve(addr) .serve(addr)
.await .await
.unwrap() .unwrap()