hotshot_libp2p_networking/network/
def.rs

1// Copyright (c) 2021-2024 Espresso Systems (espressosys.com)
2// This file is part of the HotShot repository.
3
4// You should have received a copy of the MIT License
5// along with the HotShot repository. If not, see <https://mit-license.org/>.
6
7use hotshot_types::traits::signature_key::SignatureKey;
8use libp2p::{
9    Multiaddr,
10    gossipsub::{Behaviour as GossipBehaviour, Event as GossipEvent, IdentTopic},
11    identify::{Behaviour as IdentifyBehaviour, Event as IdentifyEvent},
12    kad::store::MemoryStore,
13    request_response::{OutboundRequestId, ResponseChannel},
14};
15use libp2p_identity::PeerId;
16use libp2p_swarm_derive::NetworkBehaviour;
17use tracing::{error, info};
18
19use super::{
20    NetworkEventInternal,
21    behaviours::dht::store::{
22        persistent::{DhtPersistentStorage, PersistentStore},
23        validated::ValidatedStore,
24    },
25    cbor,
26    log_summary::LogEvent,
27};
28
29/// Overarching network behaviour performing:
30/// - network topology discovery
31/// - direct messaging
32/// - p2p broadcast
33/// - connection management
34#[derive(NetworkBehaviour, derive_more::Debug)]
35#[behaviour(to_swarm = "NetworkEventInternal")]
36pub struct NetworkDef<K: SignatureKey + 'static, D: DhtPersistentStorage> {
37    /// purpose: broadcasting messages to many peers
38    /// NOTE gossipsub works ONLY for sharing messages right now
39    /// in the future it may be able to do peer discovery and routing
40    /// <https://github.com/libp2p/rust-libp2p/issues/2398>
41    #[debug(skip)]
42    gossipsub: GossipBehaviour,
43
44    /// The DHT store. We use a `PersistentStore` to occasionally save the DHT to
45    /// some persistent store and a `ValidatedStore` to validate the records stored.
46    #[debug(skip)]
47    pub dht: libp2p::kad::Behaviour<PersistentStore<ValidatedStore<MemoryStore, K>, D>>,
48
49    /// purpose: identifying the addresses from an outside POV
50    #[debug(skip)]
51    identify: IdentifyBehaviour,
52
53    /// purpose: directly messaging peer
54    #[debug(skip)]
55    pub direct_message: cbor::Behaviour<Vec<u8>, Vec<u8>>,
56}
57
58impl<K: SignatureKey + 'static, D: DhtPersistentStorage> NetworkDef<K, D> {
59    /// Create a new instance of a `NetworkDef`
60    #[must_use]
61    pub fn new(
62        gossipsub: GossipBehaviour,
63        dht: libp2p::kad::Behaviour<PersistentStore<ValidatedStore<MemoryStore, K>, D>>,
64        identify: IdentifyBehaviour,
65        direct_message: super::cbor::Behaviour<Vec<u8>, Vec<u8>>,
66    ) -> NetworkDef<K, D> {
67        Self {
68            gossipsub,
69            dht,
70            identify,
71            direct_message,
72        }
73    }
74}
75
76/// Address functions
77impl<K: SignatureKey + 'static, D: DhtPersistentStorage> NetworkDef<K, D> {
78    /// Add an address
79    pub fn add_address(&mut self, peer_id: &PeerId, address: Multiaddr) {
80        // NOTE to get this address to play nice with the other
81        // behaviours using the DHT for routing
82        // we only need to add this address to the DHT since it
83        // is always enabled. If it were not always enabled,
84        // we would need to manually add the address to
85        // the direct message behaviour
86        self.dht.add_address(peer_id, address);
87    }
88}
89
90/// Gossip functions
91impl<K: SignatureKey + 'static, D: DhtPersistentStorage> NetworkDef<K, D> {
92    /// Publish a given gossip
93    pub fn publish_gossip(&mut self, topic: IdentTopic, contents: Vec<u8>) {
94        if let Err(e) = self.gossipsub.publish(topic, contents) {
95            LogEvent::GossipPublishFailure.record();
96            tracing::debug!("Failed to publish gossip message. Error: {:?}", e);
97        }
98    }
99    /// Subscribe to a given topic
100    pub fn subscribe_gossip(&mut self, t: &str) {
101        if let Err(e) = self.gossipsub.subscribe(&IdentTopic::new(t)) {
102            error!("Failed to subscribe to topic {:?}. Error: {:?}", t, e);
103        }
104    }
105
106    /// Unsubscribe from a given topic
107    pub fn unsubscribe_gossip(&mut self, t: &str) {
108        if !self.gossipsub.unsubscribe(&IdentTopic::new(t)) {
109            info!("We were not subscribed to topic {:?}.", t);
110        }
111    }
112}
113
114/// Request/response functions
115impl<K: SignatureKey + 'static, D: DhtPersistentStorage> NetworkDef<K, D> {
116    /// Add a direct request for a given peer
117    pub fn add_direct_request(&mut self, peer_id: PeerId, data: Vec<u8>) -> OutboundRequestId {
118        self.direct_message.send_request(&peer_id, data)
119    }
120
121    /// Add a direct response for a channel
122    pub fn add_direct_response(&mut self, chan: ResponseChannel<Vec<u8>>, msg: Vec<u8>) {
123        let _ = self.direct_message.send_response(chan, msg);
124    }
125}
126
127impl From<GossipEvent> for NetworkEventInternal {
128    fn from(event: GossipEvent) -> Self {
129        Self::GossipEvent(Box::new(event))
130    }
131}
132
133impl From<libp2p::kad::Event> for NetworkEventInternal {
134    fn from(event: libp2p::kad::Event) -> Self {
135        Self::DHTEvent(event)
136    }
137}
138
139impl From<IdentifyEvent> for NetworkEventInternal {
140    fn from(event: IdentifyEvent) -> Self {
141        Self::IdentifyEvent(Box::new(event))
142    }
143}
144impl From<libp2p::request_response::Event<Vec<u8>, Vec<u8>>> for NetworkEventInternal {
145    fn from(value: libp2p::request_response::Event<Vec<u8>, Vec<u8>>) -> Self {
146        Self::DMEvent(value)
147    }
148}
hotshot_libp2p_networking/network/def.rs

hotshot_libp2p_networking/network/
def.rs
