use runa_protocol::Value; use std::io::{BufRead, BufReader, Write}; use std::net::{Shutdown, SocketAddr, TcpStream}; use std::process::{Child, Command as ProcessCommand, Stdio}; use std::time::Duration; use runa_client::{LocalClient, TcpClient}; use runa_protocol::{Command, PROTOCOL_VERSION, Reply, Request, Response}; struct ServerProcess(Child); impl Drop for ServerProcess { fn drop(&mut self) { let _ = self.0.kill(); let _ = self.0.wait(); } } #[test] fn local_client_launches_server_and_operates_the_same_live_image() { let mut client = LocalClient::spawn(env!("CARGO_BIN_EXE_runa-server"), &[] as &[&str]).unwrap(); let published = client .request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(), }) .unwrap(); assert_eq!(published.request_id, "runa-client-1"); assert!(matches!(published.result, Ok(Reply::Behavior(_)))); let spawned = client .request(Command::Spawn { behavior_name: "counter".to_owned(), state: Value::Integer(5), mailbox_capacity: 2, }) .unwrap(); assert_eq!(spawned.request_id, "runa-client-2"); let actor_ref = match spawned.result { Ok(Reply::ActorSpawned { actor_ref }) => actor_ref, other => panic!("unexpected spawn response: {other:?}"), }; assert_eq!(actor_ref.len(), 21); client .request(Command::Send { actor_ref: actor_ref.clone(), message: test_message(0, 7, None), }) .unwrap(); client .request(Command::Tick { ticks: 100, instruction_budget: 10, }) .unwrap(); let inspected = client.request(Command::InspectActor { actor_ref }).unwrap(); assert!(matches!( inspected.result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12) )); assert!(client.shutdown().unwrap().success()); } #[test] fn tcp_server_accepts_local_protocol_requests() { let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server")) .args(["--listen", "127.0.0.1:0"]) .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() .unwrap(); let mut server = ServerProcess(child); let stderr = server.0.stderr.take().unwrap(); let mut stderr = BufReader::new(stderr); let mut startup = String::new(); stderr.read_line(&mut startup).unwrap(); let address = startup .strip_prefix("runa-server: listening on ") .expect("server should report its bound address") .trim() .parse::() .unwrap(); let mut client = TcpStream::connect_timeout(&address, Duration::from_secs(2)).unwrap(); let request = Request { version: PROTOCOL_VERSION, request_id: "tcp-process-test".to_owned(), command: Command::ListActors, }; writeln!(client, "{}", serde_json::to_string(&request).unwrap()).unwrap(); client.shutdown(Shutdown::Write).unwrap(); let mut response_line = String::new(); BufReader::new(client) .read_line(&mut response_line) .unwrap(); let response: Response = serde_json::from_str(&response_line).unwrap(); assert_eq!(response.request_id, "tcp-process-test"); assert_eq!(response.result, Ok(Reply::Actors(Vec::new()))); let mut first = TcpClient::connect(address).unwrap(); let mut second = TcpClient::connect(address).unwrap(); let actor_ref = match first .request(Command::Spawn { behavior_name: "counter".to_owned(), state: Value::Integer(5), mailbox_capacity: 2, }) .unwrap() .result .unwrap() { Reply::ActorSpawned { actor_ref } => actor_ref, other => panic!("unexpected spawn response: {other:?}"), }; assert!(matches!(second.request(Command::InspectActor { actor_ref: actor_ref.clone(), }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(5))); assert!(matches!( first .request(Command::RollbackBehavior { name: "counter".to_owned(), expected_revision: 1, target_revision: 1, }) .unwrap() .result, Ok(Reply::Behavior(behavior)) if behavior.revision == 2 )); drop(first); drop(second); let mut reconnected = TcpClient::connect(address).unwrap(); assert!(matches!( reconnected .request(Command::GetBehavior { name: "counter".to_owned(), }) .unwrap() .result, Ok(Reply::Behavior(behavior)) if behavior.revision == 2 )); reconnected .request(Command::Send { actor_ref: actor_ref.clone(), message: Value::Record(std::collections::BTreeMap::from([ ("op".to_owned(), Value::String("add".to_owned())), ("value".to_owned(), Value::Integer(7)), ])), }) .unwrap() .result .unwrap(); reconnected .request(Command::Tick { ticks: 100, instruction_budget: 10, }) .unwrap() .result .unwrap(); assert!(matches!(reconnected.request(Command::InspectActor { actor_ref: actor_ref.clone(), }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12))); reconnected .request(Command::PublishBehavior { name: "echo".to_owned(), expected_revision: None, source: "pub fn handle(state, message) { (message, []) }".to_owned(), }) .unwrap() .result .unwrap(); let nested = Value::Record(std::collections::BTreeMap::from([( "items".to_owned(), Value::List(vec![ Value::Null, Value::Bool(true), Value::Integer(-42), Value::String("hello world".to_owned()), Value::ActorRef(runa_actor::ActorRef::parse(&actor_ref).unwrap()), ]), )])); let echo = match reconnected .request(Command::Spawn { behavior_name: "echo".to_owned(), state: nested.clone(), mailbox_capacity: 2, }) .unwrap() .result .unwrap() { Reply::ActorSpawned { actor_ref } => actor_ref, other => panic!("unexpected spawn: {other:?}"), }; assert!(matches!(reconnected.request(Command::InspectActor { actor_ref: echo.clone(), }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested)); reconnected .request(Command::Send { actor_ref: echo.clone(), message: nested.clone(), }) .unwrap() .result .unwrap(); reconnected .request(Command::Tick { ticks: 100, instruction_budget: 10, }) .unwrap() .result .unwrap(); assert!(matches!(reconnected.request(Command::InspectActor { actor_ref: echo, }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested && actor.reserved.is_none())); } fn test_message(kind: i64, value: i64, target: Option) -> Value { Value::Record(std::collections::BTreeMap::from([ ("kind".to_owned(), Value::Integer(kind)), ("value".to_owned(), Value::Integer(value)), ( "target".to_owned(), target.map_or(Value::Null, Value::ActorRef), ), ])) }