Add async to console commands

Cargo.toml

@@ -11,7 +11,7 @@ repository = "https://forgejo.fireturlte.net"
 [workspace.dependencies]
 lazy-supplements.path = "lazy-supplements"
 lazy-supplements-migration.path = "lazy-supplements-migration"
-libp2p = { version = "0.55.0", features = ["noise", "ping", "tcp", "tokio", "yamux" ] }
+libp2p = { version = "0.55.0", features = ["macros", "mdns", "noise", "ping", "tcp", "tokio", "yamux" ] }
 
 [workspace.dependencies.sea-orm-migration]
 version = "1.1.0"

lazy-supplements/Cargo.toml

@@ -18,9 +18,11 @@ clap = { version = "4.5.38", features = ["derive"] }
 dirs = "6.0.0"
 futures = "0.3.31"
 libp2p.workspace = true
+rustyline = "16.0.0"
 sea-orm = { version = "1.1.11", features = ["sqlx-sqlite", "runtime-tokio-native-tls", "macros", "with-chrono", "with-uuid"] } 
 sea-orm-migration.workspace = true
 serde = { version = "1.0.219", features = ["derive"] }
+shell-words = "1.1.0"
 tempfile = { version = "3.20.0", optional = true }
 thiserror = "2.0.12"
 tokio = { version = "1.45.0", features = ["macros", "rt"] }

lazy-supplements/src/cli/console.rs

@@ -0,0 +1,91 @@
+use std::{collections::HashMap, ffi::OsString, hash::Hash, time::Duration};
+
+use clap::{Args, Parser};
+use futures::{future::BoxFuture, StreamExt};
+use libp2p::{noise, ping, swarm::{NetworkBehaviour, SwarmEvent}, tcp, yamux, Swarm};
+use tokio::time::sleep;
+use tracing_subscriber::EnvFilter;
+
+use crate::{error::Error, global::GLOBAL};
+
+use super::{node::parse_and_run_console_node_command, ConfigArgs};
+
+pub trait Executable {
+    fn execute(self) -> Result<(), Error>;
+}
+
+pub trait ConsoleCommand {
+    fn execute_line(&self, line: String) -> Result<(), Error>;
+}
+pub struct ConsoleCommands {
+    content: HashMap<&'static str, Box<dyn Fn(Vec<String>) -> BoxFuture<'static, Result<(), Error>>>>,
+}
+impl ConsoleCommands {
+    pub fn new() -> Self {
+        Self{content: HashMap::new()}
+    }
+    pub fn insert<F, Fut>(&mut self,name: &'static str, f: F)
+    where 
+        F: Fn(Vec<String>) -> Fut + 'static,
+        Fut: Future<Output = Result<(), Error>> + Send + 'static,
+    {
+        if let Some(_) = self.content.insert(name, Box::new(move |v| {
+            Box::pin(f(v))
+        })){
+            unreachable!();
+        };
+    }
+    pub async fn parse_and_run(&self, line: String) -> Result<(), Error>{
+        let args = shell_words::split(&line)?;
+        if let Some(command_name) = args.first().map(|s| {s.clone()}) {
+            if let Some(command) = self.content.get(command_name.as_str()) {
+                command(args).await
+            } else {
+                println!("Invalid command: {command_name}");
+                self.print_commands();
+                Ok(())
+            }
+        } else {
+            Ok(())
+        }
+    }
+    pub fn print_commands(&self) {
+        for key in self.content.keys(){
+            println!("{key}");
+        }
+    }
+}
+
+impl Default for ConsoleCommands {
+    fn default() -> Self {
+        let mut commands = Self::new();
+        commands.insert("node", parse_and_run_console_node_command);
+        commands
+    }
+}
+
+#[derive(Args, Debug)]
+pub struct ConsoleArgs {
+    #[command(flatten)]
+    config: ConfigArgs,
+}
+
+impl ConsoleArgs {
+    pub async fn start_console(self, commands: ConsoleCommands) -> Result<(), Error>
+    {
+        let _ = crate::global::GLOBAL.get_or_init_node_config(self.config.try_into_node_config().await?).await;
+        tokio::spawn( async {
+            GLOBAL.launch_swarm().await
+        });
+        let mut rl = rustyline::DefaultEditor::new()?;
+        loop {
+            match rl.readline(">> ") {
+                Ok(line) => {
+                    commands.parse_and_run(line).await?
+                },
+                Err(x) => Err(x)?,
+            };
+        }
+    }
+}
+

lazy-supplements/src/cli/mod.rs

@@ -1,11 +1,13 @@
 use std::path::PathBuf;
 
 mod config;
+mod console;
 mod init;
 mod node;
 mod server;
 
 pub use config::ConfigArgs;
+pub use console::{ConsoleArgs, ConsoleCommands};
 pub use init::InitArgs;
-pub use node::{ NodeArgs, NodeCommand, JoinNodeArgs };
+pub use node::{ NodeArgs, NodeCommand, JoinNodeArgs , ConsoleNodeArgs};
 pub use server::ServerArgs;

lazy-supplements/src/cli/node.rs

@@ -1,6 +1,6 @@
 use std::{net::IpAddr, path::PathBuf};
 
-use clap::{Args, Subcommand};
+use clap::{Args, Parser, Subcommand};
 use futures::StreamExt;
 use libp2p::{
     multiaddr::Protocol, noise, ping, swarm::SwarmEvent, tcp, yamux, Multiaddr
@@ -11,12 +11,30 @@ use crate::{cli::ServerArgs, error::Error};
 
 use super::ConfigArgs;
 
-#[derive(Args, Debug)]
+#[derive(Debug, Args)]
 pub struct NodeArgs {
     #[command(subcommand)]
     pub command: NodeCommand
 }
 
+#[derive(Debug, Parser)]
+pub struct ConsoleNodeArgs {
+    #[command(flatten)]
+    pub args: NodeArgs,
+}
+
+impl ConsoleNodeArgs {
+    pub async fn run(self) -> Result<(), Error> {
+        println!("{self:?}");
+        Ok(())
+    }
+}
+
+pub async fn parse_and_run_console_node_command(s:Vec<String>) -> Result<(), Error> {
+    let args = ConsoleNodeArgs::parse_from(s);
+    args.run().await
+}
+
 #[derive(Args, Debug)]
 pub struct JoinNodeArgs {
     #[arg(long)]
@@ -53,7 +71,10 @@ impl JoinNodeArgs {
         loop{
             match swarm.select_next_some().await {
                 SwarmEvent::NewListenAddr { address, .. } => println!("Listening on {address:?}"),
-                SwarmEvent::Behaviour(event) => println!("{event:?}"),
+                SwarmEvent::Behaviour(event) => {
+                    println!("{event:?}");
+                    event.run().await;
+                },
                 _ => {}
             }
         }

lazy-supplements/src/cli/server.rs

@@ -3,7 +3,7 @@ use futures::StreamExt;
 use libp2p::{noise, ping, swarm::{NetworkBehaviour, SwarmEvent}, tcp, yamux, Swarm};
 use tracing_subscriber::EnvFilter;
 
-use crate::error::Error;
+use crate::{error::Error, global::GLOBAL};
 
 use super::ConfigArgs;
 
@@ -14,13 +14,7 @@ pub struct ServerArgs {
 }
 impl ServerArgs {
     pub async fn start_server(self) -> Result<(), Error>{
-        let mut swarm = self.config.try_into_node_config().await?.try_into_swarm().await?;
-        loop{
-            match swarm.select_next_some().await {
-                SwarmEvent::NewListenAddr { address, .. } => println!("Listening on {address:?}"),
-                SwarmEvent::Behaviour(event) => println!("{event:?}"),
-                _ => {}
-            }
-        }
+        let _ = crate::global::GLOBAL.get_or_init_node_config(self.config.try_into_node_config().await?).await;
+        GLOBAL.launch_swarm().await
     }
 }

lazy-supplements/src/config/node.rs

@@ -8,7 +8,7 @@ use tokio::{fs::File, io::{AsyncReadExt, AsyncWriteExt}};
 use tracing_subscriber::EnvFilter;
 
 
-use crate::{cli::ConfigArgs, error::Error, global::DEFAULT_DATABASE_FILE_PATH};
+use crate::{cli::ConfigArgs, error::Error, global::DEFAULT_DATABASE_FILE_PATH, p2p};
 
 use super::{PartialConfig, DEFAULT_LISTEN_IPS, DEFAULT_PORT};
 
@@ -23,7 +23,7 @@ fn base64_to_keypair(base64: &str) -> Result<Keypair, Error>  {
         Ok(Keypair::from_protobuf_encoding(&vec)?)
 }
 
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct NodeConfig {
     #[serde(with = "keypair_parser")]
     pub secret: Keypair,
@@ -33,7 +33,7 @@ pub struct NodeConfig {
 }
 
 impl NodeConfig {
-    pub async fn try_into_swarm (self) -> Result<Swarm<ping::Behaviour>, Error> {
+    pub async fn try_into_swarm (self) -> Result<Swarm<p2p::Behaviour>, Error> {
         let _ = tracing_subscriber::fmt()
             .with_env_filter(EnvFilter::from_default_env())
             .try_init();
@@ -44,7 +44,7 @@ impl NodeConfig {
                 noise::Config::new,
                 yamux::Config::default,
             )?
-            .with_behaviour(|_| ping::Behaviour::default())?
+            .with_behaviour(|keypair| p2p::Behaviour::try_from(keypair).unwrap())?
             .build();
         swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
         Ok(swarm)

lazy-supplements/src/error.rs

@@ -18,6 +18,10 @@ pub enum Error {
     Multiaddr(#[from] libp2p::multiaddr::Error),
     #[error("Noise error: {0}")]
     Noise(#[from] libp2p::noise::Error),
+    #[error("Readline error: {0}")]
+    Readline(#[from] rustyline::error::ReadlineError),
+    #[error("Shell word split error: {0}")]
+    ShellWord(#[from] shell_words::ParseError),
     #[error("toml deserialization error: {0}")]
     TomlDe(#[from] toml::de::Error),
     #[error("toml serialization error: {0}")]

lazy-supplements/src/global/mod.rs

@@ -1,8 +1,10 @@
-use std::{net::{IpAddr, Ipv4Addr}, path::PathBuf, sync::LazyLock};
+use std::{collections::HashMap, net::{IpAddr, Ipv4Addr}, path::PathBuf, sync::LazyLock};
 
-use crate::config::{NodeConfig, RawNodeConfig};
+use crate::{config::{NodeConfig, RawNodeConfig}, error::Error};
+use futures::StreamExt;
+use libp2p::{swarm::SwarmEvent, Multiaddr, PeerId};
 use sea_orm::DatabaseConnection;
-use tokio::sync::OnceCell;
+use tokio::sync::{OnceCell, RwLock};
 
 mod database;
 
@@ -53,10 +55,12 @@ pub static DEFAULT_DATABASE_FILE_PATH: LazyLock<PathBuf> = LazyLock::new(|| {
 pub static GLOBAL: Global = Global{
     node_config: OnceCell::const_new(),
     database: OnceCell::const_new(),
+    peers: OnceCell::const_new(),
 };
 pub struct Global {
     pub node_config: OnceCell<NodeConfig>,
     pub database: OnceCell<DatabaseConnection>,
+    pub peers: OnceCell<RwLock<HashMap<PeerId, Multiaddr>>>,
 }
 
 #[cfg(test)]
@@ -66,9 +70,36 @@ impl Global {
     pub fn get_node_config(&self) -> Option<&NodeConfig> {
         self.node_config.get()
     }
-    pub async fn get_or_try_init_node_config(&self, config: NodeConfig) -> &NodeConfig {
+    pub async fn get_or_init_node_config(&self, config: NodeConfig) -> &NodeConfig {
         self.node_config.get_or_init(|| async {config}).await
     }
+    pub async fn get_or_init_peers(&self) -> &RwLock<HashMap<PeerId, Multiaddr>> {
+        self.peers.get_or_init(|| async {
+            RwLock::new(HashMap::new())
+        }).await
+    }
+    pub async fn read_peers(&self) -> tokio::sync::RwLockReadGuard<'_, HashMap<PeerId, Multiaddr>>{
+        self.get_or_init_peers().await.read().await
+    }
+    pub async fn write_peers(&self) -> tokio::sync::RwLockWriteGuard<'_, HashMap<PeerId, Multiaddr>>{
+        self.get_or_init_peers().await.write().await
+    }
+    pub async fn launch_swarm(&self) -> Result<(), Error> {
+        let mut swarm = self.get_node_config().unwrap().clone().try_into_swarm().await?;
+        loop{
+            let swarm_event = swarm.select_next_some().await;
+            tokio::spawn(async move{
+                match swarm_event {
+                    SwarmEvent::NewListenAddr { address, .. } => println!("Listening on {address:?}"),
+                    SwarmEvent::Behaviour(event) => {
+                        println!("{event:?}");
+                        event.run().await;
+                    },
+                    _ => {}
+                }
+            });
+        }
+    }
 }
 
 pub static DEFAULT_RAW_NODE_CONFIG: LazyLock<RawNodeConfig> = LazyLock::new(|| {

lazy-supplements/src/lib.rs

@@ -3,5 +3,6 @@ pub mod config;
 pub mod entity;
 pub mod error;
 pub mod global;
+pub mod p2p;
 #[cfg(any(test, feature="test"))]
 pub mod tests;

lazy-supplements/src/main.rs

@@ -1,5 +1,5 @@
 use clap::{Parser, Subcommand};
-use lazy_supplements::{cli::{InitArgs, NodeArgs, NodeCommand, ServerArgs}, *};
+use lazy_supplements::{cli::{ConsoleArgs, ConsoleCommands, InitArgs, NodeArgs, NodeCommand, ServerArgs}, *};
 
 #[derive(Debug, Parser)]
 struct Cli {
@@ -9,6 +9,7 @@ struct Cli {
 
 #[derive(Debug, Subcommand)]
 enum Command {
+    Console(ConsoleArgs),
     Node(NodeArgs),
     Init(InitArgs),
     Server(ServerArgs),
@@ -24,5 +25,6 @@ async fn main() {
         },
         Command::Init(x) => x.init_config().await,
         Command::Server(x) => x.start_server().await.unwrap(),
+        Command::Console(x) => x.start_console(ConsoleCommands::default()).await.unwrap(),
     }
 }

lazy-supplements/src/p2p/mod.rs

@@ -0,0 +1,60 @@
+use libp2p::{ identity::Keypair, mdns, ping, swarm};
+
+use crate::error::Error;
+
+#[derive(swarm::NetworkBehaviour)]
+#[behaviour(to_swarm = "Event")]
+pub struct Behaviour {
+    pub mdns: mdns::tokio::Behaviour,
+    pub ping: ping::Behaviour,    
+}
+
+impl TryFrom<&Keypair> for Behaviour {
+    type Error = Error;
+    fn try_from(keypair: &Keypair) -> Result<Self, Error> {
+        Ok(Self {
+            mdns: mdns::tokio::Behaviour::new(
+                mdns::Config::default(),
+                keypair.public().into(),
+            )?,
+            ping: libp2p::ping::Behaviour::new(ping::Config::new()),
+        })
+    }
+}
+
+#[derive(Debug)]
+pub enum Event {
+    Mdns(mdns::Event),
+    Ping(ping::Event),
+}
+
+impl Event {
+    pub async fn run(self) {
+        match self {
+            Self::Mdns(x) => {
+                match x {
+                    mdns::Event::Discovered(e) => {
+                        for peer in e {
+                            let mut peers = crate::global::GLOBAL.write_peers().await;
+                            peers.insert(peer.0, peer.1);
+                        }
+                        let peers = crate::global::GLOBAL.read_peers().await;
+                        println!("Peers: {peers:?}");
+                    },
+                    _ => {},
+                }
+            },
+            _ => {}
+        }
+    }
+}
+impl From<mdns::Event> for Event {
+    fn from(event: mdns::Event) -> Self {
+        Self::Mdns(event)
+    }
+}
+impl From<ping::Event> for Event {
+    fn from(event: ping::Event) -> Self {
+        Self::Ping(event)
+    }
+}