cuprate_p2p_core/client/
connector.rs

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,
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, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr> {
36    handshaker: HandShaker<Z, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>,
37}
38
39impl<Z: NetworkZone, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
40    Connector<Z, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
41{
42    /// Create a new connector from a handshaker.
43    pub const fn new(
44        handshaker: HandShaker<Z, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>,
45    ) -> Self {
46        Self { handshaker }
47    }
48}
49
50impl<Z: NetworkZone, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr, BrdcstStrm>
51    Service<ConnectRequest<Z>> for Connector<Z, AdrBook, CSync, ProtoHdlrMkr, BrdcstStrmMkr>
52where
53    AdrBook: AddressBook<Z> + Clone,
54    CSync: CoreSyncSvc + Clone,
55    ProtoHdlrMkr: ProtocolRequestHandlerMaker<Z> + Clone,
56    BrdcstStrm: Stream<Item = BroadcastMessage> + Send + 'static,
57    BrdcstStrmMkr: Fn(InternalPeerID<Z::Addr>) -> BrdcstStrm + Clone + Send + 'static,
58{
59    type Response = Client<Z>;
60    type Error = HandshakeError;
61    type Future =
62        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
63
64    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
65        Poll::Ready(Ok(()))
66    }
67
68    fn call(&mut self, req: ConnectRequest<Z>) -> Self::Future {
69        tracing::debug!("Connecting to peer: {}", req.addr);
70        let mut handshaker = self.handshaker.clone();
71
72        async move {
73            let (peer_stream, peer_sink) = Z::connect_to_peer(req.addr).await?;
74            let req = DoHandshakeRequest {
75                addr: InternalPeerID::KnownAddr(req.addr),
76                permit: req.permit,
77                peer_stream,
78                peer_sink,
79                direction: ConnectionDirection::Outbound,
80            };
81            handshaker.ready().await?.call(req).await
82        }
83        .boxed()
84    }
85}
cuprate_p2p_core/client/connector.rs

cuprate_p2p_core/client/
connector.rs
