From e022b2b77e3e98f0f444f637f96e74e6a6f990cf Mon Sep 17 00:00:00 2001 From: hozan23 Date: Sat, 25 May 2024 07:15:02 +0200 Subject: p2p: rename connection.rs to conn_queue.rs --- p2p/src/backend.rs | 2 +- p2p/src/conn_queue.rs | 70 ++++++++++++++++++++++++++++++++++++++++++++++++ p2p/src/connection.rs | 70 ------------------------------------------------ p2p/src/discovery/mod.rs | 2 +- p2p/src/lib.rs | 2 +- p2p/src/peer/mod.rs | 2 +- p2p/src/peer_pool.rs | 8 +++--- 7 files changed, 78 insertions(+), 78 deletions(-) create mode 100644 p2p/src/conn_queue.rs delete mode 100644 p2p/src/connection.rs (limited to 'p2p/src') diff --git a/p2p/src/backend.rs b/p2p/src/backend.rs index 98297e5..16cc20b 100644 --- a/p2p/src/backend.rs +++ b/p2p/src/backend.rs @@ -6,7 +6,7 @@ use karyon_core::{async_runtime::Executor, crypto::KeyPair}; use crate::{ config::Config, - connection::ConnQueue, + conn_queue::ConnQueue, discovery::{ArcDiscovery, Discovery}, monitor::Monitor, peer_pool::PeerPool, diff --git a/p2p/src/conn_queue.rs b/p2p/src/conn_queue.rs new file mode 100644 index 0000000..9a153f3 --- /dev/null +++ b/p2p/src/conn_queue.rs @@ -0,0 +1,70 @@ +use std::{collections::VecDeque, fmt, sync::Arc}; + +use async_channel::Sender; + +use karyon_core::{async_runtime::lock::Mutex, async_util::CondVar}; +use karyon_net::Conn; + +use crate::{message::NetMsg, Result}; + +/// Defines the direction of a network connection. +#[derive(Clone, Debug)] +pub enum ConnDirection { + Inbound, + Outbound, +} + +impl fmt::Display for ConnDirection { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match self { + ConnDirection::Inbound => write!(f, "Inbound"), + ConnDirection::Outbound => write!(f, "Outbound"), + } + } +} + +pub struct NewConn { + pub direction: ConnDirection, + pub conn: Conn, + pub disconnect_signal: Sender>, +} + +/// Connection queue +pub struct ConnQueue { + queue: Mutex>, + conn_available: CondVar, +} + +impl ConnQueue { + pub fn new() -> Arc { + Arc::new(Self { + queue: Mutex::new(VecDeque::new()), + conn_available: CondVar::new(), + }) + } + + /// Push a connection into the queue and wait for the disconnect signal + pub async fn handle(&self, conn: Conn, direction: ConnDirection) -> Result<()> { + let (disconnect_signal, chan) = async_channel::bounded(1); + let new_conn = NewConn { + direction, + conn, + disconnect_signal, + }; + self.queue.lock().await.push_back(new_conn); + self.conn_available.signal(); + if let Ok(result) = chan.recv().await { + return result; + } + Ok(()) + } + + /// Receive the next connection in the queue + pub async fn next(&self) -> NewConn { + let mut queue = self.queue.lock().await; + while queue.is_empty() { + queue = self.conn_available.wait(queue).await; + } + queue.pop_front().unwrap() + } +} diff --git a/p2p/src/connection.rs b/p2p/src/connection.rs deleted file mode 100644 index 9a153f3..0000000 --- a/p2p/src/connection.rs +++ /dev/null @@ -1,70 +0,0 @@ -use std::{collections::VecDeque, fmt, sync::Arc}; - -use async_channel::Sender; - -use karyon_core::{async_runtime::lock::Mutex, async_util::CondVar}; -use karyon_net::Conn; - -use crate::{message::NetMsg, Result}; - -/// Defines the direction of a network connection. -#[derive(Clone, Debug)] -pub enum ConnDirection { - Inbound, - Outbound, -} - -impl fmt::Display for ConnDirection { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - match self { - ConnDirection::Inbound => write!(f, "Inbound"), - ConnDirection::Outbound => write!(f, "Outbound"), - } - } -} - -pub struct NewConn { - pub direction: ConnDirection, - pub conn: Conn, - pub disconnect_signal: Sender>, -} - -/// Connection queue -pub struct ConnQueue { - queue: Mutex>, - conn_available: CondVar, -} - -impl ConnQueue { - pub fn new() -> Arc { - Arc::new(Self { - queue: Mutex::new(VecDeque::new()), - conn_available: CondVar::new(), - }) - } - - /// Push a connection into the queue and wait for the disconnect signal - pub async fn handle(&self, conn: Conn, direction: ConnDirection) -> Result<()> { - let (disconnect_signal, chan) = async_channel::bounded(1); - let new_conn = NewConn { - direction, - conn, - disconnect_signal, - }; - self.queue.lock().await.push_back(new_conn); - self.conn_available.signal(); - if let Ok(result) = chan.recv().await { - return result; - } - Ok(()) - } - - /// Receive the next connection in the queue - pub async fn next(&self) -> NewConn { - let mut queue = self.queue.lock().await; - while queue.is_empty() { - queue = self.conn_available.wait(queue).await; - } - queue.pop_front().unwrap() - } -} diff --git a/p2p/src/discovery/mod.rs b/p2p/src/discovery/mod.rs index 64e5c14..529469e 100644 --- a/p2p/src/discovery/mod.rs +++ b/p2p/src/discovery/mod.rs @@ -16,7 +16,7 @@ use karyon_net::{Conn, Endpoint}; use crate::{ config::Config, - connection::{ConnDirection, ConnQueue}, + conn_queue::{ConnDirection, ConnQueue}, connector::Connector, listener::Listener, message::NetMsg, diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs index c873980..546353a 100644 --- a/p2p/src/lib.rs +++ b/p2p/src/lib.rs @@ -40,7 +40,7 @@ mod backend; mod codec; mod config; -mod connection; +mod conn_queue; mod connector; mod discovery; mod error; diff --git a/p2p/src/peer/mod.rs b/p2p/src/peer/mod.rs index f0f6f17..b4cfd3a 100644 --- a/p2p/src/peer/mod.rs +++ b/p2p/src/peer/mod.rs @@ -18,7 +18,7 @@ use karyon_core::{ use karyon_net::{Conn, Endpoint}; use crate::{ - connection::ConnDirection, + conn_queue::ConnDirection, message::{NetMsg, NetMsgCmd, ProtocolMsg, ShutdownMsg}, peer_pool::{ArcPeerPool, WeakPeerPool}, protocol::{Protocol, ProtocolEvent, ProtocolID}, diff --git a/p2p/src/peer_pool.rs b/p2p/src/peer_pool.rs index 07bb73d..79fd4b4 100644 --- a/p2p/src/peer_pool.rs +++ b/p2p/src/peer_pool.rs @@ -21,7 +21,7 @@ use karyon_net::Conn; use crate::{ config::Config, - connection::{ConnDirection, ConnQueue}, + conn_queue::{ConnDirection, ConnQueue}, message::{get_msg_payload, NetMsg, NetMsgCmd, VerAckMsg, VerMsg}, monitor::{Monitor, PeerPoolEvent}, peer::{ArcPeer, Peer, PeerID}, @@ -99,11 +99,11 @@ impl PeerPool { /// Listens to a new connection from the connection queue pub async fn listen_loop(self: Arc) { loop { - let new_conn = self.conn_queue.next().await; - let signal = new_conn.disconnect_signal; + let conn = self.conn_queue.next().await; + let signal = conn.disconnect_signal; let result = self - .new_peer(new_conn.conn, &new_conn.direction, signal.clone()) + .new_peer(conn.conn, &conn.direction, signal.clone()) .await; // Only send a disconnect signal if there is an error when adding a peer. -- cgit v1.2.3