basic networking

Cargo.toml

@@ -13,5 +13,6 @@ serde_json = "1.0.82"
 serde = {version = "1.0.140", features = ["derive"]}
 bevy = "0.7.0"
 
+
 [workspace]
-members = ["server", "client"]
+members = ["server", "client", "bevy-add-events-macro"]

client/Cargo.toml

@@ -17,4 +17,5 @@ bevy = "0.7.0"
 color-eyre = "0.6.2"
 bytes = "1.2.0"
 tokio = { version = "1.20.0", features = ["full"] }
-qp2p = "0.28.7"
+qp2p = "0.28.7"
+bevy-add-events-macro = { path = "../bevy-add-events-macro"}

client/src/main.rs

@@ -1,3 +1,25 @@
+use std::{
+    env,
+    net::{Ipv4Addr, SocketAddr},
+    time::Duration,
+};
+use bevy::app::App;
+use bevy::DefaultPlugins;
+use bevy::prelude::{Commands, EventWriter};
+use tokio::runtime::Runtime;
+use crate::networking::event;
+
+mod networking;
+
 pub fn main() {
+    let mut app = App::new();
+    app.add_plugins(DefaultPlugins);
+    app.add_plugin(networking::plugin::NetworkingPlugin);
+    app.add_startup_system(my_test);
+    app.run();
     println!("hello world!")
 }
+
+pub fn my_test(mut commands: Commands, mut writer: EventWriter<event::ConnectToServer>) {
+    writer.send(event::ConnectToServer(SocketAddr::from((Ipv4Addr::LOCALHOST, 3030))));
+}

client/src/networking/event.rs

@@ -0,0 +1,3 @@
+use std::net::SocketAddr;
+
+pub struct ConnectToServer(pub(crate) SocketAddr);

client/src/networking/mod.rs

@@ -0,0 +1,3 @@
+pub mod plugin;
+pub mod event;
+pub mod resource;

client/src/networking/plugin.rs

@@ -1,4 +1,3 @@
-use std::net::SocketAddr;
 use bevy::app::{App, Plugin};
 use bevy::prelude::{Commands, EventReader, ResMut};
 use bytes::Bytes;
@@ -7,22 +6,96 @@ use qp2p::{Config, Endpoint};
 use tokio::runtime::Runtime;
 use bevy_add_events_macro::add_events;
 use crate::networking::{event, resource};
+use crate::networking::resource::NetworkingRuntimeHandlerSignals::KILL;
+use std::{
+    env,
+    net::{Ipv4Addr, SocketAddr},
+    time::Duration,
+};
+use std::sync::Arc;
+use bincode::config;
+use serde_json::Value;
 
 pub struct NetworkingPlugin;
 
 impl Plugin for NetworkingPlugin {
     fn build(&self, app: &mut App) {
-        add_events!(app, ServerConnection, ServerSocketAddr);
-        app.add_startup_system(setup_networking_runtime());
+        add_events!(app, event::ConnectToServer);
+        app.add_startup_system(setup_networking_runtime);
+        app.add_system(connection_added);
+        app.add_system(server_client_connection_handler);
     }
 }
 
 pub fn setup_networking_runtime(mut commands: Commands) {
     commands.insert_resource(resource::NetworkingRuntime(Runtime::new().expect("networking runtime needs to be init'd to work!")));
+    commands.insert_resource(Arc::new(resource::ServerConnection::new()));
+    commands.insert_resource(Arc::new(resource::NetworkingRuntimeMessenger::new()));
 }
 
-pub fn connection_added(mut commands: Commands, mut socket_addr: EventReader<event::ServerSocketAddr>, mut networking_runtime: ResMut<resource::NetworkingRuntime>) {
+/// sends events based on things received from the server connection
+pub fn server_client_connection_handler(mut commands: Commands, mut server_cx: ResMut<Arc<resource::ServerConnection>>) {
+    for message in server_cx.rx.try_iter() {
+        let (string, size) = bincode::decode_from_slice(message.as_ref(), config::standard()).unwrap();
+        let str: &str = string;
+        println!("message is: {}", str);
+    }
+}
+
+pub fn connection_added(
+    mut server_cx_events: EventReader<event::ConnectToServer>,
+    mut server_connection: ResMut<Arc<resource::ServerConnection>>,
+    mut net_runtime: ResMut<resource::NetworkingRuntime>,
+    mut net_runtime_messenger: ResMut<Arc<resource::NetworkingRuntimeMessenger>>) {
+    //We wanna get the last connection request, or just end this function if we don't have any.
+    let socket_address = match server_cx_events.iter().last() {
+        None => {return;}
+        Some(sock) => {sock.clone()}
+    }.0.clone();
+    //Then we wanna kill whatever is currently running
+    net_runtime_messenger.cx.send(KILL).expect("unable to kill networking runtime");
+    //Then we setup our new connection
     let mut config = Config::default();
-    config.idle_timeout = None;
-    let endpoint = Endpoint::new_client(SocketAddr::from)
-}
+    config.idle_timeout = None; //change this so we can actually do something based on it.
+    let server_connection = server_connection.clone();
+    let net_runtime_messenger = net_runtime_messenger.clone();
+    net_runtime.0.spawn(async move {
+        let endpoint = Endpoint::new_client(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), config)
+            .expect("unable to setup endpoint for client connection");
+        println!("setup endpoint");
+        let (_connection, mut incoming_connection) = endpoint.connect_to(&socket_address).await.expect("error connecting to server");
+        println!("connected");
+        let _x = net_runtime_messenger.rx.try_recv().unwrap(); //have to flush.
+        while net_runtime_messenger.rx.is_empty() {
+            let recived = incoming_connection.next().await.unwrap().unwrap();
+            server_connection.cx.send(recived).unwrap();
+        }
+    });
+}
+
+
+
+
+// match socket_addr.iter().last().clone() {
+//     None => {}
+//     Some(sock_addr) => {
+//         let sock_addr = sock_addr.0.clone();
+//         //First we want to shutdown the currently running network if there is one.
+//         networking_runtime_handler.cx.send(KILL).expect("unable to kill networking runtime");
+//         //setup the client endpoint with no timeout so it can wait as long as needed
+//         let mut config = Config::default();
+//         config.idle_timeout = None; //change this so we can actually do something based on it.
+//         let endpoint = Endpoint::new_client(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), config)
+//             .expect("unable to setup endpoint for client connection");
+//         let net_runtime_handler = networking_runtime_handler.clone();
+//         let server_cx = server_connection.clone();
+//         //Then we spawn the new runtimes
+//         networking_runtime.0.spawn(async move {
+//             let future = endpoint.connect_to(&sock_addr);
+//             let (connection, mut incoming_connection) = future.await.unwrap();
+//             while net_runtime_handler.rx.is_empty() {
+//                 server_cx.cx.send(incoming_connection.next().await.unwrap().unwrap()).unwrap();
+//             }
+//         });
+//     }
+// }

client/src/networking/resource.rs

@@ -0,0 +1,39 @@
+use bytes::Bytes;
+use crossbeam_channel::{Receiver, Sender};
+use tokio::runtime::Runtime;
+
+pub struct ServerConnection {
+    pub rx: Receiver<Bytes>,
+    pub cx: Sender<Bytes>,
+}
+
+impl ServerConnection {
+    pub fn new() -> Self {
+        let (cx, rx) = crossbeam_channel::unbounded();
+        ServerConnection {
+            cx,
+            rx,
+        }
+    }
+}
+
+pub struct NetworkingRuntime(pub(crate) Runtime);
+
+pub struct NetworkingRuntimeMessenger {
+    pub rx: Receiver<NetworkingRuntimeHandlerSignals>,
+    pub cx: Sender<NetworkingRuntimeHandlerSignals>,
+}
+
+impl NetworkingRuntimeMessenger {
+    pub fn new() -> Self {
+        let (cx, rx) = crossbeam_channel::unbounded();
+        NetworkingRuntimeMessenger {
+            cx,
+            rx
+        }
+    }
+}
+
+pub enum NetworkingRuntimeHandlerSignals {
+    KILL, //Don't have any other signals as of now, may want more later
+}

server/src/main.rs

@@ -1,3 +1,37 @@
+use std::net::{Ipv4Addr, SocketAddr};
+use std::time::Duration;
+use bincode::config;
+use color_eyre::eyre::Result;
+use qp2p::{Config, Endpoint};
+use tokio::runtime::Runtime;
+
 pub fn main() {
-    println!("hello world!");
+    let runtime = Runtime::new().unwrap();
+
+    runtime.block_on(async {
+        testing_connection().await.unwrap();
+    });
 }
+
+async fn testing_connection() -> Result<()> {
+    color_eyre::install()?;
+
+    let (node, mut incoming_cx, _contacs) = Endpoint::new_peer(
+        SocketAddr::from((Ipv4Addr::LOCALHOST, 3030)),
+        &[],
+        Config {
+            idle_timeout: Duration::from_secs(600000).into(),
+            ..Default::default()
+        },
+    )
+        .await?;
+
+    while let Some((connection, mut incoming_connection)) = incoming_cx.next().await {
+        loop {
+            let bytes = bincode::encode_to_vec("hello world!", config::standard())?;
+            connection.send(bytes.into()).await?;
+        }
+    };
+
+    Ok(())
+}

src/lib.rs

@@ -1 +1,2 @@
+pub mod networking;
 //this is a lib

src/networking/mod.rs

@@ -0,0 +1,2 @@
+pub const SERVER_CLOSE_CONNECTION: &str = "{payload_header: \"server.close_connection\"}";
+pub const CLIENT_CLOSE_CONNECTION: &str = "{payload_header: \"client.close_connection\"}";