Split server and client msgs in libbonknet

bonknet_broker/src/dataconnmanager.rs

@@ -1,17 +1,13 @@
 use actix::prelude::*;
 use uuid::Uuid;
 use std::collections::HashMap;
-use std::io::Error;
-use actix_tls::accept::rustls_0_22::{TlsStream};
-use actix_rt::net::TcpStream;
 use futures::SinkExt;
 use thiserror::Error;
-use tracing::{info, error, warn, debug};
+use tracing::{info, error, warn};
 use libbonknet::ToPeerDataStream;
 use crate::TransportStream;
 
-type RawStream = TlsStream<TcpStream>;
+#[allow(dead_code)]
-
 #[derive(Error, Debug)]
 pub enum DataConnManagerError {
     #[error("Generic Failure")]

bonknet_broker/src/main.rs

@@ -10,15 +10,17 @@ use dataconnmanager::*;
 use actix::prelude::*;
 use std::sync::Arc;
 use libbonknet::*;
+use libbonknet::servermsg::*;
+use libbonknet::clientmsg::*;
 use rustls::{RootCertStore, ServerConfig};
 use rustls::server::WebPkiClientVerifier;
 use actix_tls::accept::rustls_0_22::{Acceptor as RustlsAcceptor, TlsStream};
 use actix_server::Server;
 use actix_rt::net::TcpStream;
-use actix_service::{ServiceFactoryExt as _};
+use actix_service::ServiceFactoryExt as _;
-use futures::{StreamExt, SinkExt};
+use futures::{SinkExt, StreamExt};
 use tokio_util::codec::{Framed, FramedRead, FramedWrite, LengthDelimitedCodec};
-use tracing::{info, error, warn};
+use tracing::{error, info, warn};
 use rcgen::{Certificate, CertificateParams, DnType, KeyPair};
 use tokio::io::{ReadHalf, WriteHalf};
 
@@ -125,7 +127,8 @@ async fn main() {
                                 }
                                 Some(item) => match item {
                                     Ok(buf) => {
-                                        use FromServerConnTypeMessage::*;
+                                        use libbonknet::servermsg::{FromServerConnTypeMessage, ToServerConnTypeReply};
+                                        use libbonknet::servermsg::FromServerConnTypeMessage::*;
                                         let msg: FromServerConnTypeMessage = rmp_serde::from_slice(&buf).unwrap();
                                         info!("{:?}", msg);
                                         match msg {
@@ -207,7 +210,7 @@ async fn server_command_handler(mut transport: TransportStream, peer_cert_bytes:
             }
             Some(item) => match item {
                 Ok(buf) => {
-                    use FromServerCommandMessage::*;
+                    use libbonknet::servermsg::FromServerCommandMessage::*;
                     let msg: FromServerCommandMessage = rmp_serde::from_slice(&buf).unwrap();
                     info!("{:?}", msg);
                     match msg {
@@ -269,7 +272,7 @@ async fn guestserver_handler(mut transport: TransportStream, server_db_addr: Add
             Some(item) => {
                 match item {
                     Ok(buf) => {
-                        use FromGuestServerMessage::*;
+                        use libbonknet::servermsg::FromGuestServerMessage::*;
                         let msg: FromGuestServerMessage = rmp_serde::from_slice(&buf).unwrap();
                         info!("{:?}", msg);
                         match msg {

bonknet_broker/src/pendingdataconndb.rs

@@ -160,8 +160,8 @@ impl Handler<TryStartDataStream> for PendingDataConnManager {
     fn handle(&mut self, msg: TryStartDataStream, _ctx: &mut Self::Context) -> Self::Result {
         use RegisterKind::*;
         let idx = match msg.kind {
-            Server => self.db.iter().enumerate().find(|(i, x)| x.server.conn_id == msg.conn_id),
+            Server => self.db.iter().enumerate().find(|(_i, x)| x.server.conn_id == msg.conn_id),
-            Client => self.db.iter().enumerate().find(|(i, x)| x.client.conn_id == msg.conn_id),
+            Client => self.db.iter().enumerate().find(|(_i, x)| x.client.conn_id == msg.conn_id),
         };
         if let Some((idx, record)) = idx {
             if record.client.transport.is_some() && record.server.transport.is_some() {

bonknet_broker/src/servermanager.rs

@@ -1,26 +1,20 @@
 use std::collections::HashMap;
 use std::io::Error;
-use std::sync::{Arc};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 use actix::prelude::*;
 use rand::random;
 use thiserror::Error;
-use futures::{SinkExt};
+use futures::SinkExt;
 use tokio::sync::{Mutex, oneshot};
 use tokio_util::bytes::{Bytes, BytesMut};
 use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
 use tracing::{debug, error, info};
-use libbonknet::*;
 use crate::{TransportStream, TransportStreamRx, TransportStreamTx};
 use uuid::Uuid;
+use libbonknet::servermsg::*;
 use crate::pendingdataconndb::*;
 
-#[derive(MessageResponse)]
-enum SendMsgResult {
-    Accepted,
-    Rejected,
-}
-
 #[derive(Error, Debug)]
 pub enum SendMsgError {
     #[error("Generic Failure")]
@@ -35,7 +29,7 @@ struct SendMsg {
 
 struct ServerTransporter {
     rx: Option<TransportStreamRx>,
-    tx: Arc<tokio::sync::Mutex<TransportStreamTx>>,
+    tx: Arc<Mutex<TransportStreamTx>>,
     timeout: Option<SpawnHandle>,
     reply_channels: HashMap<u64, oneshot::Sender<FromServerReplyBody>>,
 }
@@ -49,7 +43,7 @@ impl ServerTransporter {
         let tx = FramedWrite::new(stx, codec.clone());
         ServerTransporter {
             rx: Some(rx),
-            tx: Arc::new(tokio::sync::Mutex::new(tx)),
+            tx: Arc::new(Mutex::new(tx)),
             timeout: None,
             reply_channels: HashMap::new(),
         }
@@ -78,7 +72,7 @@ impl Actor for ServerTransporter {
             let arc_tx = Arc::clone(&s.tx);
             c.spawn(async move {
                 arc_tx.lock().await.send(payload).await
-            }.into_actor(s).map(|res, a, ctx| {
+            }.into_actor(s).map(|res, _a, ctx| {
                 match res {
                     Ok(_) => {
                         info!("Ping sent!");
@@ -115,7 +109,7 @@ impl Handler<SendMsg> for ServerTransporter {
         let payload: Bytes = rmp_serde::to_vec(&msg).unwrap().into();
         let arc_tx = self.tx.clone();
         Box::pin(async move {
-            arc_tx.lock().await.send(payload).await.map_err(|e| SendMsgError::GenericFailure)?;
+            arc_tx.lock().await.send(payload).await.map_err(|_e| SendMsgError::GenericFailure)?;
             info!("msg sent");
             let r = reply_channel_rx.await.unwrap();
             info!("reply received");
@@ -128,7 +122,7 @@ impl StreamHandler<Result<BytesMut, Error>> for ServerTransporter {
     fn handle(&mut self, item: Result<BytesMut, Error>, ctx: &mut Self::Context) {
         match item {
             Ok(buf) => {
-                use FromServerReply::*;
+                use libbonknet::servermsg::FromServerReply::*;
                 let msg: FromServerReply = rmp_serde::from_slice(&buf).unwrap();
                 match msg {
                     Pong => {
@@ -226,7 +220,7 @@ impl Handler<StartTransporter> for ServerManager {
     type Result = ResponseFuture<Result<(), ServerManagerError>>;
 
     fn handle(&mut self, msg: StartTransporter, _ctx: &mut Self::Context) -> Self::Result {
-        let mut entries = Arc::clone(&self.entries);
+        let entries = Arc::clone(&self.entries);
         Box::pin(async move {
             let mut entries_mg = entries.lock().await;
             if entries_mg.contains_key(&msg.name) {

bonknet_client/src/bin/client.rs

@@ -1,15 +1,16 @@
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 use std::time::Duration;
-use futures::{StreamExt, SinkExt};
+use futures::{SinkExt, StreamExt};
 use tokio::net::TcpStream;
 use tokio_rustls::rustls::{ClientConfig, RootCertStore};
-use tokio_rustls::rustls::pki_types::{ServerName};
+use tokio_rustls::rustls::pki_types::ServerName;
 use tokio_rustls::TlsConnector;
 use tokio_util::codec::{Framed, LengthDelimitedCodec};
 use libbonknet::*;
+use libbonknet::clientmsg::*;
 use uuid::Uuid;
-use tracing::{info, error};
+use tracing::{error, info};
 
 
 async fn datastream(tlsconfig: ClientConfig, conn_id: Uuid) -> std::io::Result<()> {
@@ -129,7 +130,8 @@ async fn main() -> std::io::Result<()> {
         None => panic!("None in the transport"),
         Some(item) => match item {
             Ok(buf) => {
-                use ToClientResponse::*;
+                use libbonknet::clientmsg::ToClientResponse;
+                use libbonknet::clientmsg::ToClientResponse::*;
                 let msg: ToClientResponse = rmp_serde::from_slice(&buf).unwrap();
                 match msg {
                     OkServerList { data } => info!("{}", data.join("\n")),
@@ -151,7 +153,8 @@ async fn main() -> std::io::Result<()> {
         None => panic!("None in the transport"),
         Some(item) => match item {
             Ok(buf) => {
-                use ToClientResponse::*;
+                use libbonknet::clientmsg::ToClientResponse;
+                use libbonknet::clientmsg::ToClientResponse::*;
                 let msg: ToClientResponse = rmp_serde::from_slice(&buf).unwrap();
                 match msg {
                     OkRequest { conn_id } => {

bonknet_server/src/bin/server.rs

@@ -1,16 +1,15 @@
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
-use futures::{StreamExt, SinkExt};
+use futures::{SinkExt, StreamExt};
 use tokio::net::TcpStream;
 use tokio_rustls::rustls::{ClientConfig, RootCertStore};
-use tokio_rustls::rustls::pki_types::{ServerName, CertificateDer, PrivatePkcs8KeyDer};
+use tokio_rustls::rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer, ServerName};
 use tokio_rustls::TlsConnector;
-use tokio_util::bytes::BytesMut;
 use tokio_util::codec::{Framed, LengthDelimitedCodec};
 use libbonknet::*;
+use libbonknet::servermsg::*;
 use uuid::Uuid;
-use tracing::{info, error};
+use tracing::{error, info};
-use libbonknet::ToPeerDataStream::{OkDataStreamOpen, OkDataStreamRequestAccepted, Refused, Revoked};
 
 
 async fn datastream(tlsconfig: Arc<ClientConfig>, conn_id: Uuid) -> std::io::Result<()> {
@@ -125,7 +124,8 @@ async fn main() -> std::io::Result<()> {
         }
         Some(item) => match item {
             Ok(buf) => {
-                use ToGuestServerMessage::*;
+                use libbonknet::servermsg::{okannounce_to_cert, ToGuestServerMessage};
+                use libbonknet::servermsg::ToGuestServerMessage::*;
                 let msg: ToGuestServerMessage = rmp_serde::from_slice(&buf).unwrap();
                 info!("{:?}", msg);
                 match msg {
@@ -166,7 +166,8 @@ async fn main() -> std::io::Result<()> {
             }
             Some(item) => match item {
                 Ok(buf) => {
-                    use ToServerConnTypeReply::*;
+                    use libbonknet::servermsg::ToServerConnTypeReply;
+                    use libbonknet::servermsg::ToServerConnTypeReply::*;
                     let msg: ToServerConnTypeReply = rmp_serde::from_slice(&buf).unwrap();
                     info!("{:?}", msg);
                     match msg {
@@ -195,7 +196,8 @@ async fn main() -> std::io::Result<()> {
             }
             Some(item) => match item {
                 Ok(buf) => {
-                    use ToServerCommandReply::*;
+                    use libbonknet::servermsg::ToServerCommandReply;
+                    use libbonknet::servermsg::ToServerCommandReply::*;
                     let msg: ToServerCommandReply = rmp_serde::from_slice(&buf).unwrap();
                     info!("{:?}", msg);
                     match msg {
@@ -231,7 +233,8 @@ async fn main() -> std::io::Result<()> {
             }
             Some(item) => match item {
                 Ok(buf) => {
-                    use ToServerConnTypeReply::*;
+                    use libbonknet::servermsg::ToServerConnTypeReply;
+                    use libbonknet::servermsg::ToServerConnTypeReply::*;
                     let msg: ToServerConnTypeReply = rmp_serde::from_slice(&buf).unwrap();
                     info!("{:?}", msg);
                     match msg {
@@ -261,11 +264,13 @@ async fn main() -> std::io::Result<()> {
                     let mut out: Option<FromServerReply> = None;
                     match item {
                         Ok(buf) => {
-                            use ToServerMessage::*;
+                            use libbonknet::servermsg::ToServerMessage;
+                            use libbonknet::servermsg::ToServerMessage::*;
                             let msg: ToServerMessage = rmp_serde::from_slice(&buf).unwrap();
                             match msg {
                                 Msg { reply_id, body } => {
-                                    use ToServerMessageBody::*;
+                                    use libbonknet::servermsg::FromServerReplyBody;
+                                    use libbonknet::servermsg::ToServerMessageBody::*;
                                     match body {
                                         Request { conn_id } => {
                                             info!("I'm required with Connection ID {}", conn_id);

libbonknet/src/clientmsg.rs

@@ -0,0 +1,23 @@
+pub use crate::ToPeerDataStream;
+
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+// Client things
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromClientCommand {
+    RequestServer { name: String },
+    ServerList,
+    UpgradeToDataStream(Uuid),
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToClientResponse {
+    OkRequest { conn_id: Uuid },
+    OkServerList { data: Vec<String> },
+    // You are now a DataStream, wait the Open message
+    OkDataStreamRequestAccepted,
+    // The stream is open, you can pipe in-out the content you want!
+    OkDataStreamOpen,
+    GenericError,
+}

libbonknet/src/lib.rs

@@ -1,8 +1,10 @@
+pub mod servermsg;
+pub mod clientmsg;
+
 use std::io::{BufReader, Error, ErrorKind};
-use rustls_pemfile::{read_one, Item};
+use rustls_pemfile::{Item, read_one};
 use serde::{Deserialize, Serialize};
 use tokio_rustls::rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer};
-use uuid::Uuid;
 
 pub fn load_cert(filename: &str) -> std::io::Result<CertificateDer> {
     let cert_file = std::fs::File::open(filename).unwrap();
@@ -26,119 +28,6 @@ pub fn load_prkey(filename: &str) -> std::io::Result<PrivatePkcs8KeyDer> {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromServerConnTypeMessage {
-    SendCommand,
-    Subscribe,
-    OpenDataStream(Uuid),
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToServerConnTypeReply {
-    OkSendCommand,
-    OkSubscribe,
-    // You are now a DataStream, wait the Open message
-    OkDataStreamRequestAccepted,
-    // The stream is open, you can pipe in-out the content you want!
-    OkDataStreamOpen,
-    GenericFailure,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromServerCommandMessage {
-    ChangeName { name: String },
-    WhoAmI,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToServerCommandReply {
-    NameChanged,
-    NameNotAvailable,
-    YouAre { name: String },
-    GenericFailure,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum YouAreValues {
-    Registered { name: String },
-    NotRegistered,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToServerMessageBody {
-    Request { conn_id: Uuid },
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToServerMessage {
-    Ping,
-    Msg {
-        reply_id: u64,
-        body: ToServerMessageBody,
-    },
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromServerReplyBody {
-    RequestAccepted,
-    RequestFailed,
-    Pong,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromServerReply {
-    Pong,
-    Msg {
-        reply_id: u64,
-        body: FromServerReplyBody
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromGuestServerMessage {
-    Announce { name: String }
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToGuestServerMessage {
-    OkAnnounce { server_cert: Vec<u8>, server_prkey: Vec<u8> },
-    FailedNameAlreadyOccupied,
-}
-
-pub fn okannounce_to_cert<'a>(server_cert: Vec<u8>, server_prkey: Vec<u8>) -> (CertificateDer<'a>, PrivatePkcs8KeyDer<'a>) {
-    let server_cert = CertificateDer::from(server_cert);
-    let server_prkey = PrivatePkcs8KeyDer::from(server_prkey);
-    (server_cert, server_prkey)
-}
-
-impl ToGuestServerMessage {
-    pub fn make_okannounce(server_cert: CertificateDer, server_prkey: PrivatePkcs8KeyDer) -> Self {
-        ToGuestServerMessage::OkAnnounce {
-            server_cert: server_cert.to_vec(),
-            server_prkey: server_prkey.secret_pkcs8_der().to_vec()
-        }
-    }
-}
-
-// Client things
-#[derive(Debug, Serialize, Deserialize)]
-pub enum FromClientCommand {
-    RequestServer { name: String },
-    ServerList,
-    UpgradeToDataStream(Uuid),
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub enum ToClientResponse {
-    OkRequest { conn_id: Uuid },
-    OkServerList { data: Vec<String> },
-    // You are now a DataStream, wait the Open message
-    OkDataStreamRequestAccepted,
-    // The stream is open, you can pipe in-out the content you want!
-    OkDataStreamOpen,
-    GenericError,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub enum ToPeerDataStream {
     // You are now a DataStream, wait the Open message

libbonknet/src/servermsg.rs

@@ -0,0 +1,93 @@
+pub use crate::ToPeerDataStream;
+
+use tokio_rustls::rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromServerConnTypeMessage {
+    SendCommand,
+    Subscribe,
+    OpenDataStream(Uuid),
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToServerConnTypeReply {
+    OkSendCommand,
+    OkSubscribe,
+    // You are now a DataStream, wait the Open message
+    OkDataStreamRequestAccepted,
+    // The stream is open, you can pipe in-out the content you want!
+    OkDataStreamOpen,
+    GenericFailure,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromServerCommandMessage {
+    ChangeName { name: String },
+    WhoAmI,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToServerCommandReply {
+    NameChanged,
+    NameNotAvailable,
+    YouAre { name: String },
+    GenericFailure,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToServerMessageBody {
+    Request { conn_id: Uuid },
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToServerMessage {
+    Ping,
+    Msg {
+        reply_id: u64,
+        body: ToServerMessageBody,
+    },
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromServerReplyBody {
+    RequestAccepted,
+    RequestFailed,
+    Pong,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromServerReply {
+    Pong,
+    Msg {
+        reply_id: u64,
+        body: FromServerReplyBody
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FromGuestServerMessage {
+    Announce { name: String }
+}
+
+pub fn okannounce_to_cert<'a>(server_cert: Vec<u8>, server_prkey: Vec<u8>) -> (CertificateDer<'a>, PrivatePkcs8KeyDer<'a>) {
+    let server_cert = CertificateDer::from(server_cert);
+    let server_prkey = PrivatePkcs8KeyDer::from(server_prkey);
+    (server_cert, server_prkey)
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ToGuestServerMessage {
+    OkAnnounce { server_cert: Vec<u8>, server_prkey: Vec<u8> },
+    FailedNameAlreadyOccupied,
+}
+
+impl ToGuestServerMessage {
+    pub fn make_okannounce(server_cert: CertificateDer, server_prkey: PrivatePkcs8KeyDer) -> Self {
+        ToGuestServerMessage::OkAnnounce {
+            server_cert: server_cert.to_vec(),
+            server_prkey: server_prkey.secret_pkcs8_der().to_vec()
+        }
+    }
+}