reth_network/
network.rs

1use crate::{
2    config::NetworkMode, message::PeerMessage, protocol::RlpxSubProtocol,
3    swarm::NetworkConnectionState, transactions::TransactionsHandle, FetchClient,
4};
5use alloy_primitives::B256;
6use enr::Enr;
7use futures::StreamExt;
8use parking_lot::Mutex;
9use reth_discv4::{Discv4, NatResolver};
10use reth_discv5::Discv5;
11use reth_eth_wire::{
12    BlockRangeUpdate, BroadcastPoolTransactions, DisconnectReason, EthNetworkPrimitives,
13    NetworkPrimitives, NewPooledTransactionHashes, SharedTransactions,
14};
15use reth_ethereum_forks::Head;
16use reth_network_api::{
17    events::{NetworkPeersEvents, PeerEvent, PeerEventStream},
18    test_utils::{PeersHandle, PeersHandleProvider},
19    BlockDownloaderProvider, CellCustody, DiscoveryEvent, NetworkError, NetworkEvent,
20    NetworkEventListenerProvider, NetworkInfo, NetworkStatus, PeerInfo, PeerRequest, Peers,
21    PeersInfo,
22};
23use reth_network_p2p::sync::{NetworkSyncUpdater, SyncState, SyncStateProvider};
24use reth_network_peers::{NodeRecord, PeerId, TrustedPeer};
25use reth_network_types::{PeerAddr, PeerKind, Reputation, ReputationChangeKind};
26use reth_tokio_util::{EventSender, EventStream};
27use secp256k1::SecretKey;
28use std::{
29    net::SocketAddr,
30    sync::{
31        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
32        Arc,
33    },
34};
35use tokio::sync::{
36    mpsc::{self, UnboundedSender},
37    oneshot,
38};
39use tokio_stream::wrappers::UnboundedReceiverStream;
40
41/// A _shareable_ network frontend. Used to interact with the network.
42///
43/// See also [`NetworkManager`](crate::NetworkManager).
44#[derive(Clone, Debug)]
45pub struct NetworkHandle<N: NetworkPrimitives = EthNetworkPrimitives> {
46    /// The Arc'ed delegate that contains the state.
47    inner: Arc<NetworkInner<N>>,
48}
49
50// === impl NetworkHandle ===
51
52impl<N: NetworkPrimitives> NetworkHandle<N> {
53    /// Creates a single new instance.
54    #[expect(clippy::too_many_arguments)]
55    pub(crate) fn new(
56        num_active_peers: Arc<AtomicUsize>,
57        listener_address: Arc<Mutex<SocketAddr>>,
58        to_manager_tx: UnboundedSender<NetworkHandleMessage<N>>,
59        secret_key: SecretKey,
60        local_peer_id: PeerId,
61        peers: PeersHandle,
62        network_mode: NetworkMode,
63        chain_id: Arc<AtomicU64>,
64        tx_gossip_disabled: bool,
65        discv4: Option<Discv4>,
66        discv5: Option<Discv5>,
67        event_sender: EventSender<NetworkEvent<PeerRequest<N>>>,
68        nat: Option<NatResolver>,
69    ) -> Self {
70        let inner = NetworkInner {
71            num_active_peers,
72            to_manager_tx,
73            listener_address,
74            secret_key,
75            local_peer_id,
76            peers,
77            network_mode,
78            is_syncing: Arc::new(AtomicBool::new(false)),
79            initial_sync_done: Arc::new(AtomicBool::new(false)),
80            chain_id,
81            cell_custody: CellCustody::default(),
82            tx_gossip_disabled,
83            discv4,
84            discv5,
85            event_sender,
86            nat,
87        };
88        Self { inner: Arc::new(inner) }
89    }
90
91    /// Returns the [`PeerId`] used in the network.
92    pub fn peer_id(&self) -> &PeerId {
93        &self.inner.local_peer_id
94    }
95
96    fn manager(&self) -> &UnboundedSender<NetworkHandleMessage<N>> {
97        &self.inner.to_manager_tx
98    }
99
100    /// Returns the mode of the network, either pow, or pos
101    pub fn mode(&self) -> &NetworkMode {
102        &self.inner.network_mode
103    }
104
105    /// Sends a [`NetworkHandleMessage`] to the manager
106    pub(crate) fn send_message(&self, msg: NetworkHandleMessage<N>) {
107        let _ = self.inner.to_manager_tx.send(msg);
108    }
109
110    /// Update the status of the node.
111    pub fn update_status(&self, head: Head) {
112        self.send_message(NetworkHandleMessage::StatusUpdate { head });
113    }
114
115    /// Announce a block over devp2p
116    ///
117    /// Caution: in `PoS` this is a noop because new blocks are no longer announced over devp2p.
118    /// Instead they are sent to the node by CL and can be requested over devp2p.
119    /// Broadcasting new blocks is considered a protocol violation.
120    pub fn announce_block(&self, block: N::NewBlockPayload, hash: B256) {
121        self.send_message(NetworkHandleMessage::AnnounceBlock(block, hash))
122    }
123
124    /// Sends a [`PeerRequest`] to the given peer's session.
125    pub fn send_request(&self, peer_id: PeerId, request: PeerRequest<N>) {
126        self.send_message(NetworkHandleMessage::EthRequest { peer_id, request })
127    }
128
129    /// Send transactions hashes to the peer.
130    pub fn send_transactions_hashes(&self, peer_id: PeerId, msg: NewPooledTransactionHashes) {
131        self.send_message(NetworkHandleMessage::SendPooledTransactionHashes { peer_id, msg })
132    }
133
134    /// Send full transactions to the peer
135    pub fn send_transactions(&self, peer_id: PeerId, msg: Vec<Arc<N::BroadcastedTransaction>>) {
136        self.send_message(NetworkHandleMessage::SendTransaction {
137            peer_id,
138            msg: SharedTransactions(msg),
139        })
140    }
141
142    /// Send cached full pool transactions to the peer.
143    pub(crate) fn send_broadcast_pool_transactions(
144        &self,
145        peer_id: PeerId,
146        msg: BroadcastPoolTransactions,
147    ) {
148        self.send_message(NetworkHandleMessage::SendBroadcastPoolTransactions { peer_id, msg })
149    }
150
151    /// Send eth message to the peer.
152    pub fn send_eth_message(&self, peer_id: PeerId, message: PeerMessage<N>) {
153        self.send_message(NetworkHandleMessage::EthMessage { peer_id, message })
154    }
155
156    /// Send message to get the [`TransactionsHandle`].
157    ///
158    /// Returns `None` if no transaction task is installed.
159    pub async fn transactions_handle(&self) -> Option<TransactionsHandle<N>> {
160        let (tx, rx) = oneshot::channel();
161        let _ = self.manager().send(NetworkHandleMessage::GetTransactionsHandle(tx));
162        rx.await.unwrap()
163    }
164
165    /// Send message to gracefully shutdown node.
166    ///
167    /// This will disconnect all active and pending sessions and prevent
168    /// new connections to be established.
169    pub async fn shutdown(&self) -> Result<(), oneshot::error::RecvError> {
170        let (tx, rx) = oneshot::channel();
171        self.send_message(NetworkHandleMessage::Shutdown(tx));
172        rx.await
173    }
174
175    /// Set network connection state to Active.
176    ///
177    /// New outbound connections will be established if there's capacity.
178    pub fn set_network_active(&self) {
179        self.set_network_conn(NetworkConnectionState::Active);
180    }
181
182    /// Set network connection state to Hibernate.
183    ///
184    /// No new outbound connections will be established.
185    pub fn set_network_hibernate(&self) {
186        self.set_network_conn(NetworkConnectionState::Hibernate);
187    }
188
189    /// Set network connection state.
190    fn set_network_conn(&self, network_conn: NetworkConnectionState) {
191        self.send_message(NetworkHandleMessage::SetNetworkState(network_conn));
192    }
193
194    /// Whether tx gossip is disabled
195    pub fn tx_gossip_disabled(&self) -> bool {
196        self.inner.tx_gossip_disabled
197    }
198
199    /// Returns the secret key used for authenticating sessions.
200    pub fn secret_key(&self) -> &SecretKey {
201        &self.inner.secret_key
202    }
203
204    /// Returns the [`Discv4`] handle if discv4 is enabled.
205    pub fn discv4(&self) -> Option<&Discv4> {
206        self.inner.discv4.as_ref()
207    }
208
209    /// Returns the [`Discv5`] handle if discv5 is enabled.
210    pub fn discv5(&self) -> Option<&Discv5> {
211        self.inner.discv5.as_ref()
212    }
213}
214
215// === API Implementations ===
216
217impl<N: NetworkPrimitives> NetworkPeersEvents for NetworkHandle<N> {
218    /// Returns an event stream of peer-specific network events.
219    fn peer_events(&self) -> PeerEventStream {
220        let peer_events = self.inner.event_sender.new_listener().map(|event| match event {
221            NetworkEvent::Peer(peer_event) => peer_event,
222            NetworkEvent::ActivePeerSession { info, .. } => PeerEvent::SessionEstablished(info),
223        });
224        PeerEventStream::new(peer_events)
225    }
226}
227
228impl<N: NetworkPrimitives> NetworkEventListenerProvider for NetworkHandle<N> {
229    type Primitives = N;
230
231    fn event_listener(&self) -> EventStream<NetworkEvent<PeerRequest<Self::Primitives>>> {
232        self.inner.event_sender.new_listener()
233    }
234
235    fn discovery_listener(&self) -> UnboundedReceiverStream<DiscoveryEvent> {
236        let (tx, rx) = mpsc::unbounded_channel();
237        let _ = self.manager().send(NetworkHandleMessage::DiscoveryListener(tx));
238        UnboundedReceiverStream::new(rx)
239    }
240}
241
242impl<N: NetworkPrimitives> NetworkProtocols for NetworkHandle<N> {
243    fn add_rlpx_sub_protocol(&self, protocol: RlpxSubProtocol) {
244        self.send_message(NetworkHandleMessage::AddRlpxSubProtocol(protocol))
245    }
246}
247
248impl<N: NetworkPrimitives> PeersInfo for NetworkHandle<N> {
249    fn num_connected_peers(&self) -> usize {
250        self.inner.num_active_peers.load(Ordering::Relaxed)
251    }
252
253    fn local_node_record(&self) -> NodeRecord {
254        if let Some(discv4) = &self.inner.discv4 {
255            // Note: the discv4 services uses the same `nat` so we can directly return the node
256            // record here
257            discv4.node_record()
258        } else if let Some(discv5) = self.inner.discv5.as_ref() {
259            // for disv5 we must check if we have an external ip configured
260            if let Some(external) =
261                self.inner.nat.clone().and_then(|nat| nat.as_external_ip(discv5.local_port()))
262            {
263                NodeRecord::new((external, discv5.local_port()).into(), *self.peer_id())
264            } else {
265                // use the node record that discv5 tracks or use localhost
266                self.inner.discv5.as_ref().and_then(|d| d.node_record()).unwrap_or_else(|| {
267                    NodeRecord::new(
268                        (std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), discv5.local_port())
269                            .into(),
270                        *self.peer_id(),
271                    )
272                })
273            }
274            // also use the tcp port
275            .with_tcp_port(self.inner.listener_address.lock().port())
276        } else {
277            let mut socket_addr = *self.inner.listener_address.lock();
278
279            let external_ip =
280                self.inner.nat.clone().and_then(|nat| nat.as_external_ip(socket_addr.port()));
281
282            if let Some(ip) = external_ip {
283                // if able to resolve external ip, use it instead and also set the local address
284                socket_addr.set_ip(ip)
285            } else if socket_addr.ip().is_unspecified() {
286                // zero address is invalid
287                if socket_addr.ip().is_ipv4() {
288                    socket_addr.set_ip(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST));
289                } else {
290                    socket_addr.set_ip(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST));
291                }
292            }
293
294            NodeRecord::new(socket_addr, *self.peer_id())
295        }
296    }
297
298    fn local_enr(&self) -> Enr<SecretKey> {
299        let local_node_record = self.local_node_record();
300        let mut builder = Enr::builder();
301        builder.ip(local_node_record.address);
302        if local_node_record.address.is_ipv4() {
303            builder.udp4(local_node_record.udp_port);
304            builder.tcp4(local_node_record.tcp_port);
305
306            // add IPv6 fields from discv5 for dual-stack support
307            if let Some(discv5) = self.inner.discv5.as_ref() {
308                let discv5_enr = discv5.local_enr();
309                if let Some(ip6) = discv5_enr.ip6() {
310                    builder.ip6(ip6);
311                }
312                if let Some(udp6) = discv5_enr.udp6() {
313                    builder.udp6(udp6);
314                }
315                if let Some(tcp6) = discv5_enr.tcp6() {
316                    builder.tcp6(tcp6);
317                }
318            }
319        } else {
320            builder.udp6(local_node_record.udp_port);
321            builder.tcp6(local_node_record.tcp_port);
322        }
323
324        builder.build(&self.inner.secret_key).expect("valid enr")
325    }
326}
327
328impl<N: NetworkPrimitives> Peers for NetworkHandle<N> {
329    fn add_trusted_peer_id(&self, peer: PeerId) {
330        self.send_message(NetworkHandleMessage::AddTrustedPeerId(peer));
331    }
332
333    fn add_trusted_peer_node(&self, peer: TrustedPeer) {
334        self.send_message(NetworkHandleMessage::AddTrustedPeerNode(peer));
335    }
336
337    /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to add a peer to the known
338    /// set, with the given kind.
339    fn add_peer_kind(
340        &self,
341        peer: PeerId,
342        kind: Option<PeerKind>,
343        tcp_addr: SocketAddr,
344        udp_addr: Option<SocketAddr>,
345    ) {
346        let addr = PeerAddr::new(tcp_addr, udp_addr);
347        self.send_message(NetworkHandleMessage::AddPeerAddress(peer, kind, addr));
348    }
349
350    async fn get_peers_by_kind(&self, kind: PeerKind) -> Result<Vec<PeerInfo>, NetworkError> {
351        let (tx, rx) = oneshot::channel();
352        let _ = self.manager().send(NetworkHandleMessage::GetPeerInfosByPeerKind(kind, tx));
353        Ok(rx.await?)
354    }
355
356    async fn get_all_peers(&self) -> Result<Vec<PeerInfo>, NetworkError> {
357        let (tx, rx) = oneshot::channel();
358        let _ = self.manager().send(NetworkHandleMessage::GetPeerInfos(tx));
359        Ok(rx.await?)
360    }
361
362    async fn get_peer_by_id(&self, peer_id: PeerId) -> Result<Option<PeerInfo>, NetworkError> {
363        let (tx, rx) = oneshot::channel();
364        let _ = self.manager().send(NetworkHandleMessage::GetPeerInfoById(peer_id, tx));
365        Ok(rx.await?)
366    }
367
368    async fn get_peers_by_id(&self, peer_ids: Vec<PeerId>) -> Result<Vec<PeerInfo>, NetworkError> {
369        let (tx, rx) = oneshot::channel();
370        let _ = self.manager().send(NetworkHandleMessage::GetPeerInfosByIds(peer_ids, tx));
371        Ok(rx.await?)
372    }
373
374    /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to remove a peer from the
375    /// set corresponding to given kind.
376    fn remove_peer(&self, peer: PeerId, kind: PeerKind) {
377        self.send_message(NetworkHandleMessage::RemovePeer(peer, kind))
378    }
379
380    /// Sends a message to the [`NetworkManager`](crate::NetworkManager)  to disconnect an existing
381    /// connection to the given peer.
382    fn disconnect_peer(&self, peer: PeerId) {
383        self.send_message(NetworkHandleMessage::DisconnectPeer(peer, None))
384    }
385
386    /// Sends a message to the [`NetworkManager`](crate::NetworkManager)  to disconnect an existing
387    /// connection to the given peer using the provided reason
388    fn disconnect_peer_with_reason(&self, peer: PeerId, reason: DisconnectReason) {
389        self.send_message(NetworkHandleMessage::DisconnectPeer(peer, Some(reason)))
390    }
391
392    /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to ban the given peer and
393    /// disconnect an active non-trusted session if one exists.
394    fn ban_peer(&self, peer: PeerId) {
395        self.send_message(NetworkHandleMessage::BanPeer(peer))
396    }
397
398    /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to unban the given peer.
399    fn unban_peer(&self, peer: PeerId) {
400        self.send_message(NetworkHandleMessage::UnbanPeer(peer))
401    }
402
403    /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to connect to the given
404    /// peer.
405    ///
406    /// This will add a new entry for the given peer if it isn't tracked yet.
407    /// If it is tracked then the peer is updated with the given information.
408    fn connect_peer_kind(
409        &self,
410        peer_id: PeerId,
411        kind: PeerKind,
412        tcp_addr: SocketAddr,
413        udp_addr: Option<SocketAddr>,
414    ) {
415        self.send_message(NetworkHandleMessage::ConnectPeer(
416            peer_id,
417            kind,
418            PeerAddr::new(tcp_addr, udp_addr),
419        ))
420    }
421
422    /// Send a reputation change for the given peer.
423    fn reputation_change(&self, peer_id: PeerId, kind: ReputationChangeKind) {
424        self.send_message(NetworkHandleMessage::ReputationChange(peer_id, kind));
425    }
426
427    async fn reputation_by_id(&self, peer_id: PeerId) -> Result<Option<Reputation>, NetworkError> {
428        let (tx, rx) = oneshot::channel();
429        let _ = self.manager().send(NetworkHandleMessage::GetReputationById(peer_id, tx));
430        Ok(rx.await?)
431    }
432}
433
434impl<N: NetworkPrimitives> PeersHandleProvider for NetworkHandle<N> {
435    fn peers_handle(&self) -> &PeersHandle {
436        &self.inner.peers
437    }
438}
439
440impl<N: NetworkPrimitives> NetworkInfo for NetworkHandle<N> {
441    fn local_addr(&self) -> SocketAddr {
442        *self.inner.listener_address.lock()
443    }
444
445    async fn network_status(&self) -> Result<NetworkStatus, NetworkError> {
446        let (tx, rx) = oneshot::channel();
447        let _ = self.manager().send(NetworkHandleMessage::GetStatus(tx));
448        rx.await.map_err(Into::into)
449    }
450
451    fn chain_id(&self) -> u64 {
452        self.inner.chain_id.load(Ordering::Relaxed)
453    }
454
455    fn cell_custody(&self) -> &CellCustody {
456        &self.inner.cell_custody
457    }
458
459    fn is_syncing(&self) -> bool {
460        SyncStateProvider::is_syncing(self)
461    }
462
463    fn is_initially_syncing(&self) -> bool {
464        SyncStateProvider::is_initially_syncing(self)
465    }
466}
467
468impl<N: NetworkPrimitives> SyncStateProvider for NetworkHandle<N> {
469    fn is_syncing(&self) -> bool {
470        self.inner.is_syncing.load(Ordering::Relaxed)
471    }
472    // used to guard the txpool
473    fn is_initially_syncing(&self) -> bool {
474        if self.inner.initial_sync_done.load(Ordering::Relaxed) {
475            return false
476        }
477        self.inner.is_syncing.load(Ordering::Relaxed)
478    }
479}
480
481impl<N: NetworkPrimitives> NetworkSyncUpdater for NetworkHandle<N> {
482    fn update_sync_state(&self, state: SyncState) {
483        let future_state = state.is_syncing();
484        let prev_state = self.inner.is_syncing.swap(future_state, Ordering::Relaxed);
485        let syncing_to_idle_state_transition = prev_state && !future_state;
486        if syncing_to_idle_state_transition {
487            self.inner.initial_sync_done.store(true, Ordering::Relaxed);
488        }
489    }
490
491    /// Update the status of the node.
492    fn update_status(&self, head: Head) {
493        self.send_message(NetworkHandleMessage::StatusUpdate { head });
494    }
495
496    /// Updates the advertised block range.
497    fn update_block_range(&self, update: reth_eth_wire::BlockRangeUpdate) {
498        self.send_message(NetworkHandleMessage::InternalBlockRangeUpdate(update));
499    }
500}
501
502impl<N: NetworkPrimitives> BlockDownloaderProvider for NetworkHandle<N> {
503    type Client = FetchClient<N>;
504
505    async fn fetch_client(&self) -> Result<Self::Client, oneshot::error::RecvError> {
506        let (tx, rx) = oneshot::channel();
507        let _ = self.manager().send(NetworkHandleMessage::FetchClient(tx));
508        rx.await
509    }
510}
511
512#[derive(Debug)]
513struct NetworkInner<N: NetworkPrimitives = EthNetworkPrimitives> {
514    /// Number of active peer sessions the node's currently handling.
515    num_active_peers: Arc<AtomicUsize>,
516    /// Sender half of the message channel to the [`crate::NetworkManager`].
517    to_manager_tx: UnboundedSender<NetworkHandleMessage<N>>,
518    /// The local address that accepts incoming connections.
519    listener_address: Arc<Mutex<SocketAddr>>,
520    /// The secret key used for authenticating sessions.
521    secret_key: SecretKey,
522    /// The identifier used by this node.
523    local_peer_id: PeerId,
524    /// Access to all the nodes.
525    peers: PeersHandle,
526    /// The mode of the network
527    network_mode: NetworkMode,
528    /// Represents if the network is currently syncing.
529    is_syncing: Arc<AtomicBool>,
530    /// Used to differentiate between an initial pipeline sync or a live sync
531    initial_sync_done: Arc<AtomicBool>,
532    /// The chain id
533    chain_id: Arc<AtomicU64>,
534    /// Shared blob cell custody bitmap.
535    cell_custody: CellCustody,
536    /// Whether to disable transaction gossip
537    tx_gossip_disabled: bool,
538    /// The instance of the discv4 service
539    discv4: Option<Discv4>,
540    /// The instance of the discv5 service
541    discv5: Option<Discv5>,
542    /// Sender for high level network events.
543    event_sender: EventSender<NetworkEvent<PeerRequest<N>>>,
544    /// The NAT resolver
545    nat: Option<NatResolver>,
546}
547
548/// Provides access to modify the network's additional protocol handlers.
549pub trait NetworkProtocols: Send + Sync {
550    /// Adds an additional protocol handler to the `RLPx` sub-protocol list.
551    fn add_rlpx_sub_protocol(&self, protocol: RlpxSubProtocol);
552}
553
554/// Internal messages that can be passed to the  [`NetworkManager`](crate::NetworkManager).
555#[derive(Debug)]
556pub(crate) enum NetworkHandleMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
557    /// Marks a peer as trusted.
558    AddTrustedPeerId(PeerId),
559    /// Adds a trusted peer that may use a hostname, registering it for periodic DNS re-resolution.
560    AddTrustedPeerNode(TrustedPeer),
561    /// Adds an address for a peer, including its ID, kind, and socket address.
562    AddPeerAddress(PeerId, Option<PeerKind>, PeerAddr),
563    /// Removes a peer from the peerset corresponding to the given kind.
564    RemovePeer(PeerId, PeerKind),
565    /// Disconnects a connection to a peer if it exists, optionally providing a disconnect reason.
566    DisconnectPeer(PeerId, Option<DisconnectReason>),
567    /// Bans a peer and disconnects an active non-trusted session if one exists.
568    BanPeer(PeerId),
569    /// Unbans a peer.
570    UnbanPeer(PeerId),
571    /// Broadcasts an event to announce a new block to all nodes.
572    AnnounceBlock(N::NewBlockPayload, B256),
573    /// Sends a list of transactions to the given peer.
574    SendTransaction {
575        /// The ID of the peer to which the transactions are sent.
576        peer_id: PeerId,
577        /// The shared transactions to send.
578        msg: SharedTransactions<N::BroadcastedTransaction>,
579    },
580    /// Sends cached full pool transactions to the given peer.
581    SendBroadcastPoolTransactions {
582        /// The ID of the peer to which the transactions are sent.
583        peer_id: PeerId,
584        /// The cached pool transactions to send.
585        msg: BroadcastPoolTransactions,
586    },
587    /// Sends a list of transaction hashes to the given peer.
588    SendPooledTransactionHashes {
589        /// The ID of the peer to which the transaction hashes are sent.
590        peer_id: PeerId,
591        /// The new pooled transaction hashes to send.
592        msg: NewPooledTransactionHashes,
593    },
594    /// Sends an `eth` protocol request to the peer.
595    EthRequest {
596        /// The peer to send the request to.
597        peer_id: PeerId,
598        /// The request to send to the peer's sessions.
599        request: PeerRequest<N>,
600    },
601    /// Sends an `eth` protocol message to the peer.
602    EthMessage {
603        /// The peer to send the message to.
604        peer_id: PeerId,
605        /// The `eth` protocol message to send to the peer's session.
606        message: PeerMessage<N>,
607    },
608    /// Applies a reputation change to the given peer.
609    ReputationChange(PeerId, ReputationChangeKind),
610    /// Returns the client that can be used to interact with the network.
611    FetchClient(oneshot::Sender<FetchClient<N>>),
612    /// Applies a status update.
613    StatusUpdate {
614        /// The head status to apply.
615        head: Head,
616    },
617    /// Retrieves the current status via a oneshot sender.
618    GetStatus(oneshot::Sender<NetworkStatus>),
619    /// Gets `PeerInfo` for the specified peer IDs.
620    GetPeerInfosByIds(Vec<PeerId>, oneshot::Sender<Vec<PeerInfo>>),
621    /// Gets `PeerInfo` from all the peers via a oneshot sender.
622    GetPeerInfos(oneshot::Sender<Vec<PeerInfo>>),
623    /// Gets `PeerInfo` for a specific peer via a oneshot sender.
624    GetPeerInfoById(PeerId, oneshot::Sender<Option<PeerInfo>>),
625    /// Gets `PeerInfo` for a specific peer kind via a oneshot sender.
626    GetPeerInfosByPeerKind(PeerKind, oneshot::Sender<Vec<PeerInfo>>),
627    /// Gets the reputation for a specific peer via a oneshot sender.
628    GetReputationById(PeerId, oneshot::Sender<Option<Reputation>>),
629    /// Retrieves the `TransactionsHandle` via a oneshot sender.
630    GetTransactionsHandle(oneshot::Sender<Option<TransactionsHandle<N>>>),
631    /// Initiates a graceful shutdown of the network via a oneshot sender.
632    Shutdown(oneshot::Sender<()>),
633    /// Sets the network state between hibernation and active.
634    SetNetworkState(NetworkConnectionState),
635    /// Adds a new listener for `DiscoveryEvent`.
636    DiscoveryListener(UnboundedSender<DiscoveryEvent>),
637    /// Adds an additional `RlpxSubProtocol`.
638    AddRlpxSubProtocol(RlpxSubProtocol),
639    /// Connect to the given peer.
640    ConnectPeer(PeerId, PeerKind, PeerAddr),
641    /// Message to update the node's advertised block range information.
642    InternalBlockRangeUpdate(BlockRangeUpdate),
643}
reth_network/network.rs

reth_network/
network.rs
