From ddbaf0588f4210459ceab2b0b58de72f2c3e7fa3 Mon Sep 17 00:00:00 2001 From: Josiah Glosson Date: Tue, 28 Jan 2025 14:25:17 -0600 Subject: [PATCH] Implement client-side except for socket loop --- apps/app-playground/src/main.rs | 108 ++++++++++-------- apps/labrinth/src/routes/internal/statuses.rs | 2 +- packages/app-lib/src/config.rs | 8 +- packages/app-lib/src/state/friends.rs | 83 +++++++++++--- packages/app-lib/src/state/tunnel.rs | 63 +++++++++- .../rust-common/src/networking/message.rs | 2 +- 6 files changed, 190 insertions(+), 76 deletions(-) diff --git a/apps/app-playground/src/main.rs b/apps/app-playground/src/main.rs index a50002dc7..549d638cc 100644 --- a/apps/app-playground/src/main.rs +++ b/apps/app-playground/src/main.rs @@ -3,9 +3,12 @@ windows_subsystem = "windows" )] +use std::env::args; +use std::net::SocketAddr; use theseus::prelude::*; - -use theseus::profile::create::profile_create; +use tokio::net::TcpListener; +use tokio::signal::ctrl_c; +use uuid::Uuid; // A simple Rust implementation of the authentication run // 1) call the authenticate_begin_flow() function to get the URL to open (like you would in the frontend) @@ -41,54 +44,61 @@ async fn main() -> theseus::Result<()> { // Initialize state State::init().await?; - if minecraft_auth::users().await?.is_empty() { - println!("No users found, authenticating."); - authenticate_run().await?; // could take credentials from here direct, but also deposited in state users + // if minecraft_auth::users().await?.is_empty() { + // println!("No users found, authenticating."); + // authenticate_run().await?; // could take credentials from here direct, but also deposited in state users + // } + + match args().nth(1).as_deref() { + Some("host") => main_host().await?, + Some("client") => main_client().await?, + Some(other) => tracing::error!( + "'host' or 'client' expected as first CLI arg, but found '{other}'" + ), + None => tracing::error!("Expected first CLI arg 'host' or 'client'"), } - // - // st.settings - // .write() - // .await - // .java_globals - // .insert(JAVA_8_KEY.to_string(), check_jre(path).await?.unwrap()); - // Clear profiles - println!("Clearing profiles."); - { - let h = profile::list().await?; - for profile in h.into_iter() { - profile::remove(&profile.path).await?; - } - } - - println!("Creating/adding profile."); - - let name = "Example".to_string(); - let game_version = "1.16.1".to_string(); - let modloader = ModLoader::Forge; - let loader_version = "stable".to_string(); - - let profile_path = profile_create( - name, - game_version, - modloader, - Some(loader_version), - None, - None, - None, - ) - .await?; - - println!("running"); - // Run a profile, running minecraft and store the RwLock to the process - let process = profile::run(&profile_path).await?; - - println!("Minecraft UUID: {}", process.uuid); - - println!("All running process UUID {:?}", process::get_all().await?); - - // hold the lock to the process until it ends - println!("Waiting for process to end..."); - process::wait_for(process.uuid).await?; + + Ok(()) +} + +async fn main_host() -> theseus::Result<()> { + tracing::info!("Starting host"); + + let socket = State::get().await?.friends_socket.open_port(25565).await?; + tracing::info!("Running host on socket {}", socket.socket_id()); + + ctrl_c().await?; + tracing::info!("Stopping host"); + socket.shutdown().await?; + + Ok(()) +} + +async fn main_client() -> theseus::Result<()> { + tracing::info!("Starting client"); + + let socket_id = args() + .nth(2) + .expect("Expected second CLI arg to be socket ID") + .parse::()?; + + tracing::info!("Listening on port 25565 to connect to {socket_id}"); + let tcp_stream = + TcpListener::bind(SocketAddr::new("127.0.0.1".parse().unwrap(), 25585)) + .await? + .accept() + .await? + .0; + tracing::info!("Connecting to {socket_id}"); + let socket = State::get() + .await? + .friends_socket + .connect_to_socket(socket_id, tcp_stream) + .await?; + + ctrl_c().await?; + tracing::info!("Stopping client"); + socket.shutdown().await?; Ok(()) } diff --git a/apps/labrinth/src/routes/internal/statuses.rs b/apps/labrinth/src/routes/internal/statuses.rs index bf24649e7..9a48851f9 100644 --- a/apps/labrinth/src/routes/internal/statuses.rs +++ b/apps/labrinth/src/routes/internal/statuses.rs @@ -282,7 +282,7 @@ pub async fn ws_init( TunnelSocketType::Listening => { let _ = broadcast_friends( user.id, - ServerToClientMessage::FriendSocketStoppedListening { socket }, + ServerToClientMessage::FriendSocketStoppedListening { user: user.id }, &pool, &db, None, diff --git a/packages/app-lib/src/config.rs b/packages/app-lib/src/config.rs index bf7e47e1e..ad59e2d0e 100644 --- a/packages/app-lib/src/config.rs +++ b/packages/app-lib/src/config.rs @@ -4,10 +4,10 @@ // pub const MODRINTH_API_URL: &str = "https://staging-api.modrinth.com/v2/"; // pub const MODRINTH_API_URL_V3: &str = "https://staging-api.modrinth.com/v3/"; -pub const MODRINTH_URL: &str = "https://modrinth.com/"; -pub const MODRINTH_API_URL: &str = "https://api.modrinth.com/v2/"; -pub const MODRINTH_API_URL_V3: &str = "https://api.modrinth.com/v3/"; +pub const MODRINTH_URL: &str = "http://localhost:3000/"; +pub const MODRINTH_API_URL: &str = "http://localhost:8000/v2/"; +pub const MODRINTH_API_URL_V3: &str = "http://localhost:8000/v3/"; -pub const MODRINTH_SOCKET_URL: &str = "wss://api.modrinth.com/"; +pub const MODRINTH_SOCKET_URL: &str = "ws://localhost:8000/"; pub const META_URL: &str = "https://launcher-meta.modrinth.com/"; diff --git a/packages/app-lib/src/state/friends.rs b/packages/app-lib/src/state/friends.rs index 332323fff..698ec1ac2 100644 --- a/packages/app-lib/src/state/friends.rs +++ b/packages/app-lib/src/state/friends.rs @@ -2,8 +2,8 @@ use crate::config::{MODRINTH_API_URL_V3, MODRINTH_SOCKET_URL}; use crate::data::ModrinthCredentials; use crate::event::emit::emit_friend; use crate::event::FriendPayload; -use crate::state::tunnel::TunnelSocket; -use crate::state::{ProcessManager, Profile}; +use crate::state::tunnel::InternalTunnelSocket; +use crate::state::{ProcessManager, Profile, TunnelSocket}; use crate::util::fetch::{fetch_advanced, fetch_json, FetchSemaphore}; use async_tungstenite::tokio::{connect_async, ConnectStream}; use async_tungstenite::tungstenite::client::IntoClientRequest; @@ -21,19 +21,22 @@ use rust_common::networking::message::{ }; use rust_common::users::{UserId, UserStatus}; use serde::{Deserialize, Serialize}; +use std::net::SocketAddr; +use std::ops::Deref; use std::sync::Arc; use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use uuid::Uuid; -type WriteSocket = +pub(super) type WriteSocket = Arc, Message>>>>; +pub(super) type TunnelSockets = Arc>>; pub struct FriendsSocket { write: WriteSocket, user_statuses: Arc>, - tunnel_sockets: Arc>, + tunnel_sockets: TunnelSockets, } #[derive(Deserialize, Serialize)] @@ -175,26 +178,22 @@ impl FriendsSocket { ServerToClientMessage::SocketConnected { to_socket, new_socket } => { if let Some(connected_to) = sockets.get(&to_socket) { - if let TunnelSocket::Listening(connected_to) = connected_to.value() { - if let Ok(local_addr) = connected_to.local_addr() { - if let Ok(new_stream) = TcpStream::connect(local_addr).await { - sockets.insert(new_socket, TunnelSocket::Connected(new_stream)); - continue; - } + if let InternalTunnelSocket::Listening(local_addr) = *connected_to.value().clone() { + if let Ok(new_stream) = TcpStream::connect(local_addr).await { + sockets.insert(new_socket, Arc::new(InternalTunnelSocket::Connected(Mutex::new(new_stream)))); + continue; } } } let _ = Self::send_message(&write_handle, ClientToServerMessage::SocketClose { socket: new_socket }).await; }, ServerToClientMessage::SocketClosed { socket } => { - if let Some((_, TunnelSocket::Connected(mut stream))) = sockets.remove(&socket) { - let _ = stream.shutdown().await; - } + sockets.remove_if(&socket, |_, x| matches!(*x.clone(), InternalTunnelSocket::Connected(_))); }, ServerToClientMessage::SocketData { socket, data } => { if let Some(mut socket) = sockets.get_mut(&socket) { - if let TunnelSocket::Connected(ref mut stream) = socket.value_mut() { - let _ = stream.write_all(&data).await; + if let InternalTunnelSocket::Connected(ref stream) = *socket.value_mut().clone() { + let _ = stream.lock().await.write_all(&data).await; } } }, @@ -358,8 +357,58 @@ impl FriendsSocket { Ok(()) } + #[tracing::instrument(skip(self))] + pub async fn open_port(&self, port: u16) -> crate::Result { + let socket_id = Uuid::new_v4(); + let socket = self.tunnel_sockets.entry(socket_id).insert(Arc::new( + InternalTunnelSocket::Listening(SocketAddr::new( + "127.0.0.1".parse().unwrap(), + port, + )), + )); + Self::send_message( + &self.write, + ClientToServerMessage::SocketListen { socket: socket_id }, + ) + .await?; + self.create_tunnel_socket(socket_id, socket) + } + + pub async fn connect_to_socket( + &self, + to_socket: Uuid, + stream: TcpStream, + ) -> crate::Result { + let socket_id = Uuid::new_v4(); + let socket = self.tunnel_sockets.entry(socket_id).insert(Arc::new( + InternalTunnelSocket::Connected(Mutex::new(stream)), + )); + Self::send_message( + &self.write, + ClientToServerMessage::SocketConnect { + from_socket: socket_id, + to_socket, + }, + ) + .await?; + self.create_tunnel_socket(socket_id, socket) + } + + fn create_tunnel_socket( + &self, + socket_id: Uuid, + socket: impl Deref>, + ) -> crate::Result { + Ok(TunnelSocket { + socket_id, + write: self.write.clone(), + sockets: self.tunnel_sockets.clone(), + internal: socket.clone(), + }) + } + #[tracing::instrument(skip(write))] - async fn send_message( + pub(super) async fn send_message( write: &WriteSocket, message: ClientToServerMessage, ) -> crate::Result<()> { diff --git a/packages/app-lib/src/state/tunnel.rs b/packages/app-lib/src/state/tunnel.rs index c76d04bc8..6f1dbee7b 100644 --- a/packages/app-lib/src/state/tunnel.rs +++ b/packages/app-lib/src/state/tunnel.rs @@ -1,6 +1,61 @@ -use tokio::net::{TcpListener, TcpStream}; +use crate::state::friends::{TunnelSockets, WriteSocket}; +use crate::state::FriendsSocket; +use rust_common::networking::message::ClientToServerMessage; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpStream; +use tokio::sync::Mutex; +use uuid::Uuid; -pub enum TunnelSocket { - Listening(TcpListener), - Connected(TcpStream), +pub(super) enum InternalTunnelSocket { + Listening(SocketAddr), + Connected(Mutex), +} + +pub struct TunnelSocket { + pub(super) socket_id: Uuid, + pub(super) write: WriteSocket, + pub(super) sockets: TunnelSockets, + pub(super) internal: Arc, +} + +impl TunnelSocket { + pub fn socket_id(&self) -> Uuid { + self.socket_id + } + + pub async fn shutdown(self) -> crate::Result<()> { + if self.sockets.remove(&self.socket_id).is_some() { + FriendsSocket::send_message( + &self.write, + ClientToServerMessage::SocketClose { + socket: self.socket_id, + }, + ) + .await?; + if let InternalTunnelSocket::Connected(ref stream) = + *self.internal.clone() + { + stream.lock().await.shutdown().await? + } + } + Ok(()) + } +} + +impl Drop for TunnelSocket { + fn drop(&mut self) { + if self.sockets.remove(&self.socket_id).is_some() { + let write = self.write.clone(); + let socket_id = self.socket_id; + tokio::spawn(async move { + let _ = FriendsSocket::send_message( + &write, + ClientToServerMessage::SocketClose { socket: socket_id }, + ) + .await; + }); + } + } } diff --git a/packages/rust-common/src/networking/message.rs b/packages/rust-common/src/networking/message.rs index 951948bb1..85bbee267 100644 --- a/packages/rust-common/src/networking/message.rs +++ b/packages/rust-common/src/networking/message.rs @@ -51,7 +51,7 @@ pub enum ServerToClientMessage { socket: Uuid, }, FriendSocketStoppedListening { - socket: Uuid, + user: UserId, }, SocketConnected {