physnet: actually connect to manager

Cargo.toml

@@ -31,6 +31,7 @@ edition = "2021"
 rust-version = "1.78.0"
 
 [workspace.dependencies]
+async-compat = "0.2.4"
 base64 = "0.21.7"
 bevy = { version = "0.13", features = ["serialize"] }
 bevy-inspector-egui = "0.23.4"

apps/networked_physics_demo/client/Cargo.toml

@@ -23,3 +23,5 @@ rand.workspace = true
 rand_xoshiro.workspace = true
 replicate-client.workspace = true
 serde_json.workspace = true
+tokio.workspace = true
+async-compat.workspace = true

apps/networked_physics_demo/client/src/netcode.rs

@@ -5,47 +5,110 @@ use bevy::{
 	ecs::{
 		component::Component,
 		entity::Entity,
-		event::{Event, EventReader},
+		event::{Event, EventReader, EventWriter},
 		query::{Added, With, Without},
-		schedule::NextState,
-		system::{Commands, Query, Res, ResMut, Resource},
+		system::{CommandQueue, Commands, Query, Res, ResMut, Resource},
+		world::World,
 	},
-	log::trace,
-	reflect::Reflect,
+	log::{error, trace},
+	tasks::IoTaskPool,
 	transform::components::{GlobalTransform, Transform},
 };
+use color_eyre::eyre::{Result, WrapErr as _};
 use replicate_client::common::data_model::{DataModel, Entity as DmEntity};
+use tokio::sync::mpsc;
 
-use crate::GameModeState;
+const BOUNDED_CHAN_COMMAND_QUEUE_SIZE: usize = 16;
 
-#[derive(Debug)]
-pub struct NetcodePlugin;
+#[derive(Debug, Default)]
+pub struct NetcodePlugin {}
 
 impl Plugin for NetcodePlugin {
 	fn build(&self, app: &mut bevy::prelude::App) {
-		app.register_type::<ConnectToManager>()
-			.add_event::<ConnectToManager>()
+		app.add_event::<ConnectToManagerRequest>()
+			.add_event::<ConnectToManagerResponse>()
+			.init_resource::<CommandQueueChannel>()
 			.init_resource::<NetcodeDataModel>()
-			.add_systems(PreUpdate, from_data_model)
+			.add_systems(PreUpdate, (apply_queued_commands, from_data_model))
 			.add_systems(PostUpdate, (spawn_entities, to_data_model))
 			.add_systems(Update, on_connect_to_manager_evt);
 	}
 }
 
+#[derive(Debug, Resource)]
+struct NetcodeManager(#[allow(unused)] replicate_client::manager::Manager);
+
+/// Convenient way to receive commands sent from the async tasks.
+#[derive(Debug, Resource)]
+struct CommandQueueChannel {
+	tx: mpsc::Sender<CommandQueue>,
+	rx: mpsc::Receiver<CommandQueue>,
+}
+
+impl Default for CommandQueueChannel {
+	fn default() -> Self {
+		let (tx, rx) = mpsc::channel(BOUNDED_CHAN_COMMAND_QUEUE_SIZE);
+		Self { tx, rx }
+	}
+}
+
+fn apply_queued_commands(
+	mut commands: Commands,
+	mut chan: ResMut<CommandQueueChannel>,
+) {
+	while let Ok(mut command_queue) = chan.rx.try_recv() {
+		commands.append(&mut command_queue)
+	}
+}
+
 /// Other plugins create this when they want to connect to a manager.
-#[derive(Debug, Reflect, Event, Eq, PartialEq)]
-pub struct ConnectToManager {
-	/// The URL of the manager to connect to
-	pub manager_url: String,
+#[derive(Debug, Event, Eq, PartialEq)]
+pub struct ConnectToManagerRequest {
+	/// The URL of the manager to connect to. If `None`, locally host.
+	pub manager_url: Option<replicate_client::url::Url>,
 }
 
+/// Produced in response to [`ConnectToManagerRequest`].
+#[derive(Debug, Event)]
+pub struct ConnectToManagerResponse(pub Result<()>);
+
 fn on_connect_to_manager_evt(
-	mut connect_to_manager: EventReader<ConnectToManager>,
-	mut next_state: ResMut<NextState<GameModeState>>,
+	command_queue: Res<CommandQueueChannel>,
+	mut request: EventReader<ConnectToManagerRequest>,
+	mut response: EventWriter<ConnectToManagerResponse>,
 ) {
-	for ConnectToManager { manager_url: _ } in connect_to_manager.read() {
-		// TODO: Actually connect to the manager instead of faking it
-		next_state.set(GameModeState::InMinecraft);
+	for ConnectToManagerRequest { manager_url } in request.read() {
+		let Some(manager_url) = manager_url else {
+			response.send(ConnectToManagerResponse(Ok(())));
+			continue;
+		};
+		let pool = IoTaskPool::get();
+		let manager_url = manager_url.to_owned();
+		let tx = command_queue.tx.clone();
+		// We don't need to explicitly retrieve the return value.
+		pool.spawn(async_compat::Compat::new(async move {
+			let connect_result =
+				replicate_client::manager::Manager::connect(manager_url, None)
+					.await
+					.wrap_err("failed to connect to manager server");
+			if let Err(ref err) = connect_result {
+				error!("{err:?}");
+			}
+
+			// We use a command queue to enqueue commands back to bevy from the
+			// async code.
+			let mut queue = CommandQueue::default();
+			let response_event = ConnectToManagerResponse(connect_result.map(|mngr| {
+				queue.push(|w: &mut World| w.insert_resource(NetcodeManager(mngr)));
+			}));
+			queue.push(|w: &mut World| {
+				w.send_event(response_event).expect("failed to send event");
+			});
+			match tx.send(queue).await {
+				Ok(()) | Err(mpsc::error::SendError(_)) => (),
+			}
+		}))
+		.detach();
 	}
 }
 

apps/networked_physics_demo/client/src/title_screen.rs

@@ -3,9 +3,10 @@
 use bevy::{
 	app::{Plugin, Update},
 	ecs::{
-		event::{Event, EventWriter},
+		event::{Event, EventReader, EventWriter},
 		schedule::{
-			common_conditions::in_state, IntoSystemConfigs as _, NextState, OnEnter,
+			common_conditions::in_state, IntoSystemConfigs as _,
+			IntoSystemSetConfigs as _, NextState, OnEnter,
 		},
 		system::{Commands, Res, ResMut},
 	},
@@ -14,7 +15,10 @@ use bevy::{
 };
 use bevy_inspector_egui::bevy_egui::{egui, EguiContexts};
 
-use crate::{netcode::ConnectToManager, AppExt, GameModeState};
+use crate::{
+	netcode::{ConnectToManagerRequest, ConnectToManagerResponse},
+	AppExt, GameModeState,
+};
 
 use self::ui::EventWriters;
 
@@ -24,13 +28,20 @@ pub struct TitleScreenPlugin;
 impl Plugin for TitleScreenPlugin {
 	fn build(&self, app: &mut bevy::prelude::App) {
 		app.add_if_not_added(bevy_inspector_egui::bevy_egui::EguiPlugin)
-			.add_if_not_added(crate::netcode::NetcodePlugin)
+			.add_if_not_added(crate::netcode::NetcodePlugin::default())
 			.register_type::<ui::TitleScreen>()
+			.configure_sets(
+				Update,
+				UiStateSystems.run_if(in_state(GameModeState::TitleScreen)),
+			)
 			.add_systems(OnEnter(GameModeState::TitleScreen), setup)
 			.add_systems(
 				Update,
-				(should_transition, draw_title_screen)
-					.run_if(in_state(GameModeState::TitleScreen)),
+				(
+					handle_connect_to_manager_response.in_set(UiStateSystems),
+					(should_transition, draw_ui.after(UiStateSystems))
+						.run_if(in_state(GameModeState::TitleScreen)),
+				),
 			);
 	}
 }
@@ -39,11 +50,13 @@ impl Plugin for TitleScreenPlugin {
 mod ui {
 	use bevy::{ecs::system::Resource, prelude::default};
 
+	use crate::netcode::ConnectToManagerRequest;
+
 	use super::*;
 
 	/// [`EventWriter`]s needed by the UI.
 	pub(super) struct EventWriters<'a> {
-		pub connect_to_manager: EventWriter<'a, ConnectToManager>,
+		pub connect_to_manager: EventWriter<'a, ConnectToManagerRequest>,
 	}
 
 	impl EventWriters<'_> {
@@ -59,7 +72,7 @@ mod ui {
 		fn send(self, evw: &mut EventWriters<'_>);
 	}
 
-	impl UiEvent for ConnectToManager {
+	impl UiEvent for ConnectToManagerRequest {
 		fn send(self, evw: &mut EventWriters<'_>) {
 			evw.connect_to_manager.send(self);
 		}
@@ -94,8 +107,15 @@ mod ui {
 	/// User has chosen to create an instance.
 	#[derive(Debug, Reflect, Eq, PartialEq)]
 	pub enum CreateInstance {
-		Initial { manager_url: String },
-		WaitingForConnection,
+		Initial {
+			manager_url: String,
+			/// Non-empty will display this as the error.
+			error_msg: String,
+		},
+		WaitingForConnection {
+			/// The url given in `Initial`
+			manager_url: String,
+		},
 		Connected,
 		InstanceCreated,
 	}
@@ -105,31 +125,59 @@ mod ui {
 			match self {
 				CreateInstance::Initial {
 					ref mut manager_url,
+					ref mut error_msg,
 				} => {
-					ui.add(egui::TextEdit::singleline(manager_url).hint_text(
-						"Manager Url (leave blank to spin up server locally)",
-					));
+					ui.add(
+						egui::TextEdit::singleline(manager_url)
+							.hint_text(
+								"Manager Url (leave blank to spin up server locally)",
+							)
+							.text_color_opt(
+								(!error_msg.is_empty()).then_some(egui::Color32::RED),
+							),
+					);
 					let text = if manager_url.is_empty() {
+						error_msg.clear();
 						"Locally Host"
+					} else if !error_msg.is_empty() {
+						error_msg.as_str()
 					} else {
 						"Remotely Host"
 					};
 					if ui.button(text).clicked() {
-						evw.send(ConnectToManager {
-							manager_url: manager_url.to_owned(),
-						});
-						return CreateInstance::WaitingForConnection.into();
+						match (!manager_url.is_empty())
+							.then(|| manager_url.parse())
+							.transpose()
+						{
+							Ok(parsed_manager_url) => {
+								evw.send(ConnectToManagerRequest {
+									manager_url: parsed_manager_url,
+								});
+								return CreateInstance::WaitingForConnection {
+									manager_url: std::mem::take(manager_url),
+								}
+								.into();
+							}
+							Err(_parse_err) => {
+								error_msg.clear();
+								error_msg.push_str("Invalid URL");
+							}
+						}
 					}
 					if ui.button("Back").clicked() {
 						return default();
 					}
 					self.into()
 				}
-				CreateInstance::WaitingForConnection => {
+				CreateInstance::WaitingForConnection { .. } => {
+					ui.spinner();
+					self.into()
+				}
+				CreateInstance::Connected => {
+					ui.label("Connected to Manager, creating instance...");
 					ui.spinner();
 					self.into()
 				}
-				CreateInstance::Connected => todo!(),
 				CreateInstance::InstanceCreated => todo!(),
 			}
 		}
@@ -139,6 +187,7 @@ mod ui {
 		fn default() -> Self {
 			Self::Initial {
 				manager_url: default(),
+				error_msg: String::new(),
 			}
 		}
 	}
@@ -191,10 +240,38 @@ fn setup(mut commands: Commands) {
 	commands.init_resource::<ui::TitleScreen>()
 }
 
-fn draw_title_screen(
+/// Systems related to transition of UI state
+#[derive(bevy::prelude::SystemSet, Hash, Debug, Eq, PartialEq, Clone)]
+struct UiStateSystems;
+
+fn handle_connect_to_manager_response(
+	mut ui_state: ResMut<ui::TitleScreen>,
+	mut connect_to_manager_response: EventReader<ConnectToManagerResponse>,
+) {
+	let ui::TitleScreen::Create(ref mut create_state) = *ui_state else {
+		return;
+	};
+	for response in connect_to_manager_response.read() {
+		let Err(ref _response_err) = response.0 else {
+			*create_state = ui::CreateInstance::Connected;
+			continue;
+		};
+		if let ui::CreateInstance::WaitingForConnection {
+			ref mut manager_url,
+		} = *create_state
+		{
+			*create_state = ui::CreateInstance::Initial {
+				error_msg: "Could not connect to server".to_owned(),
+				manager_url: std::mem::take(manager_url),
+			}
+		}
+	}
+}
+
+fn draw_ui(
 	mut state: ResMut<ui::TitleScreen>,
 	mut contexts: EguiContexts,
-	connect_to_manager: EventWriter<ConnectToManager>,
+	connect_to_manager: EventWriter<ConnectToManagerRequest>,
 ) {
 	egui::Window::new("Instances")
 		.resizable(false)

crates/replicate/client/src/lib.rs

@@ -3,9 +3,11 @@ use eyre::Result;
 use tracing::warn;
 use url::Url;
 
-pub use replicate_common as common;
 use wtransport::{endpoint::ConnectOptions, ClientConfig, Endpoint};
 
+pub use replicate_common as common;
+pub use url;
+
 pub mod instance;
 pub mod manager;