letsplay_runner_core: Slowly bring up rpc transport

as well as detail outstanding additions to the client main to make it speak RPC properly

Author: modeco80

Cargo.lock

@@ -45,6 +45,7 @@ clap = { version = "4.5.6", features = ["cargo"] }
 # Async stuff
 tokio = { version = "1.37.0", features = [ "full" ] }
 tokio-util = { version = "0.7.13" }
+futures = "0.3.31"
 futures-util = "0.3.31"
 
 # Tracing

Cargo.toml

@@ -81,10 +81,10 @@ impl Frontend {
 	/// # Notes
 	///
 	/// As mentioned before, only one frontend can be created at once in a application.
-	/// 
+	///
 	/// The provided [FrontendInterface] implementation *must* last at least as long as the
 	/// frontend itself.
-	/// 
+	///
 	/// The returned [Box] *must* be held until the frontend is no longer used.
 	pub fn new(interface: *mut dyn FrontendInterface) -> Box<Self> {
 		let mut boxed = Box::new(Self {
@@ -132,6 +132,16 @@ impl Frontend {
 		self.core_library.is_some() && self.core_api.is_some()
 	}
 
+	pub fn game_loaded(&self) -> bool {
+		// It doesn't exactly make sense for a game to be considered
+		// loaded if we don't have a core.
+		if !self.core_loaded() {
+			return false;
+		}
+
+		self.game_loaded
+	}
+
 	/// Plugs in an input device to the specified port.
 	pub fn plug_input_device(&mut self, port: u32, device: *mut dyn InputDevice) {
 		if self.core_loaded() {

crates/letsplay_retro_frontend/src/frontend/frontend_impl.rs

@@ -17,6 +17,7 @@ bytes.workspace = true
 tokio.workspace = true
 tokio-util = { workspace = true, features = ["codec"] }
 futures-util = { workspace = true, features = [ "sink" ] }
+futures.workspace = true
 
 tracing.workspace = true
 tracing-subscriber.workspace = true
@@ -47,3 +48,6 @@ client = [ ]
 
 # Build runner server code
 server = [ ]
+
+# bleh
+temp = []

crates/letsplay_runner_core/Cargo.toml

@@ -3,6 +3,13 @@ use letsplay_av_ffmpeg::encoder_thread::Control;
 use super::GraphicsContexts;
 use std::time::Instant;
 
+#[repr(u32)]
+#[derive(Eq, PartialEq, PartialOrd, Ord)]
+pub enum ConfigurationState {
+	ConfigurationNeeded,
+	ConfigurationComplete
+}
+
 /// A game that should be run by the runner core.
 /// 
 /// # Implementation notes
@@ -18,6 +25,9 @@ pub trait Game {
 	// Not needed per se since we will just exit after shutdown,
 	// but cleaning up after ourselves isn't bad programming practice
 
+	/// Gets the configuration state of the game.
+	fn get_configuration_state(&mut self) -> ConfigurationState;
+
 	/// Set a named property. Failable.
 	fn set_property(&mut self, key: &str, value: &str) -> anyhow::Result<()>;
 

crates/letsplay_runner_core/src/client/game.rs

@@ -11,6 +11,8 @@ use tokio::sync::{
 	oneshot,
 };
 
+use crate::client::ConfigurationState;
+
 use super::{Game, GraphicsContexts};
 
 use letsplay_av_ffmpeg::encoder_thread;
@@ -59,56 +61,70 @@ fn main(mut rx: mpsc::UnboundedReceiver<GameThreadMessage>, mut game: Box<dyn Ga
 
 	game.init(&contexts, &video_encoder_control);
 
-	// TEMP CODE: This accepts a single tcp connection and broadcasts packets to it
-	// this is temporary as all hell
-	// yes I know sync io but its running on another thread anyways so its fine.
-	let server = std::net::TcpListener::bind("0.0.0.0:6040").expect("rrr");
-	let mut clients = Vec::new();
-
-	while clients.len() != 1 {
-		let client = server.accept().expect("baned");
-		clients.push(client.0);
-	}
+	#[cfg(feature = "temp")]
+	{
+		// TEMP CODE: This accepts a single tcp connection and broadcasts NALU packets to it
+		// this is temporary as all hell
+		// yes I know sync io but its running on another thread anyways so its fine.
+		let server = std::net::TcpListener::bind("0.0.0.0:6040").expect("rrr");
+		let mut clients = Vec::new();
+
+		while clients.len() != 1 {
+			let client = server.accept().expect("baned");
+			clients.push(client.0);
+		}
 
-	tracing::info!("all clients accepted - unblocking and completing intialization");
+		tracing::info!("all clients accepted - unblocking and completing intialization");
 
-	// Helper thread
-	std::thread::spawn(move || loop {
-		let frame = packet_waiter.wait_for_packet();
-		for client in &mut clients {
-			let _ = client.write_all(frame.data().unwrap());
-		}
-	});
+		// Helper thread
+		std::thread::spawn(move || loop {
+			let frame = packet_waiter.wait_for_packet();
+			for client in &mut clients {
+				let _ = client.write_all(frame.data().unwrap());
+			}
+		});
+	}
 
 	loop {
 		match rx.try_recv() {
-			Ok(message) => match message {
-				GameThreadMessage::Shutdown => break,
-				GameThreadMessage::Suspend { suspend, tx } => {
-					if suspended != suspend {
-						suspended = suspend
-					}
+			Ok(message) => {
+				match message {
+					GameThreadMessage::Shutdown => break,
+					GameThreadMessage::Suspend { suspend, tx } => {
+						if suspended != suspend {
+							// TODO: This is temporary, since we probably should instead shutdown or something
+							// since a unconfigured game indicates a JSON misconfiguration.
+							if suspend == false
+								&& game.get_configuration_state()
+									== ConfigurationState::ConfigurationNeeded
+							{
+								tracing::error!("Attempting to unsuspend a game that hasn't been fully configured!");
+								continue;
+							}
+							suspended = suspend
+						}
 
-					let _ = tx.send(());
-				}
+						let _ = tx.send(());
+					}
 
-				GameThreadMessage::Reset { tx } => {
-					game.reset();
-					let _ = tx.send(());
-				}
+					GameThreadMessage::Reset { tx } => {
+						game.reset();
+						let _ = tx.send(());
+					}
 
-				GameThreadMessage::SetProperty { key, value, tx } => {
-					// TODO: we should send the error result to the given tx
-					// so that we can propegate errors to the main thread
-					match game.set_property(&key, &value) {
-						Ok(_) => {}
-						Err(err) => {
-							tracing::error!("Error setting property {key} to {value}: {}", err);
-						}
-					};
-					let _ = tx.send(());
+					GameThreadMessage::SetProperty { key, value, tx } => {
+						// TODO: we should send the error result to the given tx
+						// so that we can propegate errors to the main thread
+						match game.set_property(&key, &value) {
+							Ok(_) => {}
+							Err(err) => {
+								tracing::error!("Error setting property {key} to {value}: {}", err);
+							}
+						};
+						let _ = tx.send(());
+					}
 				}
-			},
+			}
 
 			Err(TryRecvError::Empty) => {}
 			Err(TryRecvError::Disconnected) => break,
@@ -197,8 +213,8 @@ impl GameThread {
 		let _ = rx.await;
 	}
 
-	/// Shuts down the game thread.
-	pub async fn shutdown(self) {
+	/// Shuts down and waits for the game thread to exit.
+	pub fn shutdown(self) {
 		let _ = self.tx.send(GameThreadMessage::Shutdown);
 		self.join_handle.join().expect("Failed to join game thread");
 	}

crates/letsplay_runner_core/src/client/game_thread.rs

@@ -17,23 +17,24 @@ use tracing_subscriber::FmtSubscriber;
 
 use thiserror::Error;
 
+// TODO: Use this
 #[derive(Error, Debug)]
 pub enum RunnerError {}
 
 /// The main Let's Play runners using the letsplay_runner_core crate utilize.
-pub async fn main(game: Box<dyn Game + Send>) -> Result<(), RunnerError> {
+pub async fn main(game: Box<dyn Game + Send>) -> anyhow::Result<()> {
 	let subscriber = FmtSubscriber::builder()
 		.with_max_level(Level::INFO)
 		.with_thread_names(true)
 		.finish();
 
 	tracing::subscriber::set_global_default(subscriber).unwrap();
 
-	let game_thread = GameThread::spawn(game);
+	// DOGFOOD: 
+	//	- Implement JSON configuration
+	//	- Implement RPC client (including both local and remote modes)
 
-	// just spin forever for now
-	//
-	// TODO: implement RPC client
+	let game_thread = GameThread::spawn(game);
 
 	// TEMP: Just for testing libretro runner bringup
 	game_thread
@@ -49,14 +50,15 @@ pub async fn main(game: Box<dyn Game + Send>) -> Result<(), RunnerError> {
 		)
 		.await;
 
+	// FIXME: Remove this when RPC client is implemented
+	// (this is temporary for bringup)
 	game_thread.set_suspend(false).await;
 
 	loop {
 		time::sleep(Duration::from_secs(1)).await;
 	}
 
-	// Once we exit the loop, we should shutdown the game thread
-	// and all our resources
+	game_thread.shutdown();
 
 	Ok(())
 }

crates/letsplay_runner_core/src/client/mod.rs

@@ -1,15 +1,21 @@
+use bytes::{Bytes, BytesMut};
+use futures::{
+	stream::{SplitSink, SplitStream},
+	Stream,
+};
 use letsplay_core::si_unit::MB;
+use protobuf::{Message, Serialize};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio_util::codec::{Framed, LengthDelimitedCodec};
 
 use crate::shared::proto::{ClientMessage, ServerMessage};
 
 use futures_util::{SinkExt, StreamExt};
 
-// FIXME(s): 
+// FIXME(s):
 //  - DO NOT USE ANYHOW! DO NOT! SIMPLY DO NOT
 //  - Instead of providing a static read_xxx helper
-//	  I wonder if it would be slightly more ergonomic to 
+//	  I wonder if it would be slightly more ergonomic to
 //	  return a futures map() or whatever which parses
 //	  the message. It would allow us to handle hangup
 //	  a BIT easier, and tackle the first fixme too.
@@ -23,27 +29,50 @@ pub struct RpcTransport<RW>
 where
 	RW: AsyncReadExt + AsyncWriteExt + Unpin,
 {
-	transport: Framed<RW, LengthDelimitedCodec>,
+	read: SplitStream<Framed<RW, LengthDelimitedCodec>>,
+	write: SplitSink<Framed<RW, LengthDelimitedCodec>, Bytes>,
 }
 
 impl<RW> RpcTransport<RW>
 where
 	RW: AsyncReadExt + AsyncWriteExt + Unpin,
 {
 	pub fn from_stream(stream: RW) -> Self {
-		Self {
-			transport: LengthDelimitedCodec::builder()
-				.length_field_type::<u32>()
-				.max_frame_length(MAX_FRAME_SIZE)
-				.new_framed(stream),
-		}
+		let framed = LengthDelimitedCodec::builder()
+			.length_field_type::<u32>()
+			.max_frame_length(MAX_FRAME_SIZE)
+			.new_framed(stream);
+
+		let (write, read) = framed.split();
+		Self { read, write }
 	}
 
-	#[cfg(feature = "client")]
-	pub async fn read_server_message(&mut self) -> anyhow::Result<ServerMessage> {
-		if let Some(framed) = self.transport.next().await {
-			let bytes = framed?;
+	/* this MIGHT be better but doesnt compile
+	pub async fn sink_server_messages(&self) -> impl Stream<Item = anyhow::Result<ServerMessage>> {
+		self.read.map(|res| {
+			let bytes = res?;
 			Ok(ServerMessage::parse(&bytes[..])?)
+		})
+	}
+	*/
+
+	/// Writes a single protobuf message.
+	pub async fn write_message<T: Message>(
+		&mut self,
+		message: &T,
+	) -> anyhow::Result<()> {
+		let bytes = Bytes::from(message.serialize()?);
+		self.write.send(bytes).await?;
+		Ok(())
+	}
+
+	/// Read a protobuf message.
+	pub async fn read_message<T: Message>(
+		&mut self
+	) -> anyhow::Result<T> {
+		if let Some(framed) = self.read.next().await {
+			let bytes = framed?;
+			Ok(T::parse(&bytes[..])?)
 		} else {
 			Err(anyhow::anyhow!("End of stream"))
 		}

crates/letsplay_runner_core/src/shared/rpc_transport.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use anyhow::Context;
-use client::GraphicsContexts;
+use client::{ConfigurationState, GraphicsContexts};
 use letsplay_core::sleep;
 use letsplay_gpu::{self as gpu, GlFramebuffer};
 use letsplay_retro_frontend::{
@@ -84,6 +84,14 @@ impl client::Game for RetroGame {
 		self.get_frontend().reset();
 	}
 
+	fn get_configuration_state(&mut self) -> ConfigurationState {
+		if self.get_frontend().game_loaded() {
+			return ConfigurationState::ConfigurationComplete;
+		}
+
+		ConfigurationState::ConfigurationNeeded
+	}
+
 	fn set_property(&mut self, key: &str, value: &str) -> anyhow::Result<()> {
 		match key {
 			"libretro.core" => {