ab_networking/
shared.rs

1//! Data structures shared between node and node runner, facilitating exchange and creation of
2//! queries, subscriptions, various events and shared information.
3
4use crate::protocols::request_response::request_response_factory::RequestFailure;
5use crate::utils::Handler;
6use crate::utils::multihash::Multihash;
7use crate::utils::rate_limiter::RateLimiter;
8use bytes::Bytes;
9use futures::channel::{mpsc, oneshot};
10use libp2p::gossipsub::{PublishError, Sha256Topic, SubscriptionError};
11use libp2p::kad::{PeerRecord, RecordKey};
12use libp2p::{Multiaddr, PeerId};
13use parking_lot::Mutex;
14use std::sync::Arc;
15use std::sync::atomic::AtomicUsize;
16use tokio::sync::OwnedSemaphorePermit;
17
18/// Represents Kademlia events (RoutablePeer, PendingRoutablePeer, UnroutablePeer).
19#[derive(Clone, Debug)]
20pub enum PeerDiscovered {
21    /// Kademlia's unroutable peer event.
22    UnroutablePeer {
23        /// Peer ID
24        peer_id: PeerId,
25    },
26
27    /// Kademlia's routable or pending routable peer event.
28    RoutablePeer {
29        /// Peer ID
30        peer_id: PeerId,
31        /// Peer address
32        address: Multiaddr,
33    },
34}
35
36impl PeerDiscovered {
37    /// Extracts peer ID from event.
38    pub fn peer_id(&self) -> PeerId {
39        #[expect(
40            clippy::rest_pattern_accessible_field,
41            reason = "Do not need other fields"
42        )]
43        match self {
44            PeerDiscovered::UnroutablePeer { peer_id } => *peer_id,
45            PeerDiscovered::RoutablePeer { peer_id, .. } => *peer_id,
46        }
47    }
48}
49
50#[derive(Debug)]
51pub(crate) struct CreatedSubscription {
52    /// Subscription ID to be used for unsubscribing.
53    pub(crate) subscription_id: usize,
54    /// Receiver side of the channel with new messages.
55    pub(crate) receiver: mpsc::UnboundedReceiver<Bytes>,
56}
57
58#[derive(Debug)]
59pub(crate) enum Command {
60    GetValue {
61        key: Multihash,
62        result_sender: mpsc::UnboundedSender<PeerRecord>,
63        permit: OwnedSemaphorePermit,
64    },
65    PutValue {
66        key: Multihash,
67        value: Vec<u8>,
68        result_sender: mpsc::UnboundedSender<()>,
69        permit: OwnedSemaphorePermit,
70    },
71    Subscribe {
72        topic: Sha256Topic,
73        result_sender: oneshot::Sender<Result<CreatedSubscription, SubscriptionError>>,
74    },
75    Unsubscribe {
76        topic: Sha256Topic,
77        subscription_id: usize,
78    },
79    Publish {
80        topic: Sha256Topic,
81        message: Vec<u8>,
82        result_sender: oneshot::Sender<Result<(), PublishError>>,
83    },
84    GetClosestPeers {
85        key: Multihash,
86        result_sender: mpsc::UnboundedSender<PeerId>,
87        permit: Option<OwnedSemaphorePermit>,
88    },
89    GetClosestLocalPeers {
90        key: Multihash,
91        source: Option<PeerId>,
92        result_sender: oneshot::Sender<Vec<(PeerId, Vec<Multiaddr>)>>,
93    },
94    GenericRequest {
95        peer_id: PeerId,
96        addresses: Vec<Multiaddr>,
97        protocol_name: &'static str,
98        request: Vec<u8>,
99        result_sender: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
100    },
101    GetProviders {
102        key: RecordKey,
103        result_sender: mpsc::UnboundedSender<PeerId>,
104        permit: Option<OwnedSemaphorePermit>,
105    },
106    BanPeer {
107        peer_id: PeerId,
108    },
109    Dial {
110        address: Multiaddr,
111    },
112    ConnectedPeers {
113        result_sender: oneshot::Sender<Vec<PeerId>>,
114    },
115    ConnectedServers {
116        result_sender: oneshot::Sender<Vec<PeerId>>,
117    },
118    Bootstrap {
119        // No result sender means background async bootstrapping
120        result_sender: Option<mpsc::UnboundedSender<()>>,
121    },
122}
123
124#[derive(Default, Debug)]
125pub(crate) struct Handlers {
126    pub(crate) new_listener: Handler<Multiaddr>,
127    pub(crate) num_established_peer_connections_change: Handler<usize>,
128    pub(crate) connected_peer: Handler<PeerId>,
129    pub(crate) disconnected_peer: Handler<PeerId>,
130    pub(crate) peer_discovered: Handler<PeerDiscovered>,
131}
132
133#[derive(Debug)]
134pub(crate) struct Shared {
135    pub(crate) handlers: Handlers,
136    pub(crate) id: PeerId,
137    /// Addresses on which node is listening for incoming requests.
138    pub(crate) listeners: Mutex<Vec<Multiaddr>>,
139    pub(crate) external_addresses: Mutex<Vec<Multiaddr>>,
140    pub(crate) num_established_peer_connections: Arc<AtomicUsize>,
141    /// Sender end of the channel for sending commands to the swarm.
142    pub(crate) command_sender: mpsc::Sender<Command>,
143    pub(crate) rate_limiter: RateLimiter,
144}
145
146impl Shared {
147    pub(crate) fn new(
148        id: PeerId,
149        command_sender: mpsc::Sender<Command>,
150        rate_limiter: RateLimiter,
151    ) -> Self {
152        Self {
153            handlers: Handlers::default(),
154            id,
155            listeners: Mutex::default(),
156            external_addresses: Mutex::default(),
157            num_established_peer_connections: Arc::new(AtomicUsize::new(0)),
158            command_sender,
159            rate_limiter,
160        }
161    }
162}
ab_networking/shared.rs

ab_networking/
shared.rs
