physnet: actually connect to manager (#112)

I've updated the code to properly spawn a future that will start the
tokio runtime and perform the necessary steps to connect to the
replicate manager.

Next step will be to use the newly connected manager resource to spawn
an instance. The video showcases the UI flows and how connecting to a
server that doesn't exist fails, but connecting to the one I start from
the command line with a valid url succeeds.

Rebased on #111 



https://github.com/NexusSocial/nexus-vr/assets/6969415/c2d1ee2f-593a-4ee4-a3a7-bd39ccdcc569

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();
+		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(_)) => (),
+			}
+		}))
+		// We don't need to explicitly retrieve the return value.
+		.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;