cuprate_p2p_core/client/
connector.rs

Help
1//! Connector
2//!
3//! This module handles connecting to peers and giving the sink/stream to the handshaker which will then
4//! perform a handshake and create a [`Client`].
5//!
6//! This is where outbound connections are created.
7use std::{
8    future::Future,
9    pin::Pin,
10    task::{Context, Poll},
11};
12
13use futures::{FutureExt, Stream};
14use tokio::sync::OwnedSemaphorePermit;
15use tower::{Service, ServiceExt};
16
17use crate::{
18    client::{handshaker::HandShaker, Client, DoHandshakeRequest, HandshakeError, InternalPeerID},
19    AddressBook, BroadcastMessage, ConnectionDirection, CoreSyncSvc, NetworkZone,
20    ProtocolRequestHandlerMaker, Transport,
21};
22
23/// A request to connect to a peer.
24pub struct ConnectRequest<Z: NetworkZone> {
25    /// The peer's address.
26    pub addr: Z::Addr,
27    /// A permit which will be held be the connection allowing you to set limits on the number of
28    /// connections.
29    ///
30    /// This doesn't have to be set.
31    pub permit: Option<OwnedSemaphorePermit>,
32}
33
34/// The connector service, this service connects to peer and returns the [`Client`].
35pub struct Connector<Z: NetworkZone, T: Transport<Z>, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr> {
36    handshaker: HandShaker<Z, T, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>,
37}
38
39impl<Z: NetworkZone, T: Transport<Z>, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
40    Connector<Z, T, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
41{
42    /// Create a new connector from a handshaker.
43    pub const fn new(
44        handshaker: HandShaker<Z, T, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>,
45    ) -> Self {
46        Self { handshaker }
47    }
48}
49
50impl<Z: NetworkZone, T, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr, BrdcstStrm>
51    Service<ConnectRequest<Z>> for Connector<Z, T, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
52where
53    T: Transport<Z>,
54    AdrBook: AddressBook<Z> + Clone,
55    CSync: CoreSyncSvc + Clone,
56    ProtoHdlrMkr: ProtocolRequestHandlerMaker<Z> + Clone,
57    BrdcstStrm: Stream<Item = BroadcastMessage> + Send + 'static,
58    BrdcstStrmMkr: Fn(InternalPeerID<Z::Addr>) -> BrdcstStrm + Clone + Send + 'static,
59{
60    type Response = Client<Z>;
61    type Error = HandshakeError;
62    type Future =
63        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
64
65    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
66        Poll::Ready(Ok(()))
67    }
68
69    fn call(&mut self, req: ConnectRequest<Z>) -> Self::Future {
70        tracing::debug!("Connecting to peer: {}", req.addr);
71        let mut handshaker = self.handshaker.clone();
72
73        async move {
74            let (peer_stream, peer_sink) =
75                T::connect_to_peer(req.addr, handshaker.transport_config()).await?;
76            let req = DoHandshakeRequest {
77                addr: InternalPeerID::KnownAddr(req.addr),
78                permit: req.permit,
79                peer_stream,
80                peer_sink,
81                direction: ConnectionDirection::Outbound,
82            };
83            handshaker.ready().await?.call(req).await
84        }
85        .boxed()
86    }
87}
cuprate_p2p_core/client/connector.rs

cuprate_p2p_core/client/
connector.rs
