espresso_node/request_response/
mod.rs

1use std::future::Future;
2
3use data_source::DataSource;
4use derive_more::derive::Deref;
5use espresso_types::{PubKey, SeqTypes, traits::SequencerPersistence};
6use hotshot::{traits::NodeImplementation, types::BLSPrivKey};
7use hotshot_new_protocol::storage::NewProtocolStorage;
8use hotshot_types::traits::network::ConnectedNetwork;
9use network::Sender;
10use recipient_source::RecipientSource;
11use request::{Request, Response};
12use request_response::{
13    RequestError, RequestResponse, RequestResponseConfig, RequestType, network::Bytes,
14};
15use tokio::sync::mpsc::Receiver;
16
17pub mod catchup;
18pub mod data_source;
19pub mod network;
20pub mod recipient_source;
21pub mod request;
22
23/// A concrete type wrapper around `RequestResponse`. We need this so that we can implement
24/// local traits like `StateCatchup`. It also helps with readability.
25#[derive(Clone, Deref)]
26pub struct RequestResponseProtocol<
27    I: NodeImplementation<SeqTypes>,
28    N: ConnectedNetwork<PubKey>,
29    P: SequencerPersistence,
30> where
31    I::Storage: NewProtocolStorage<SeqTypes>,
32{
33    #[deref]
34    #[allow(clippy::type_complexity)]
35    /// The actual inner request response protocol
36    inner: RequestResponse<
37        Sender,
38        Receiver<Bytes>,
39        Request,
40        RecipientSource<I>,
41        DataSource<I, N, P>,
42        PubKey,
43    >,
44
45    /// The configuration we used for the above inner protocol. This is nice to have for
46    /// estimating when we should make another request
47    config: RequestResponseConfig,
48
49    /// The public key of this node
50    public_key: PubKey,
51    /// The private key of this node
52    private_key: BLSPrivKey,
53}
54
55impl<I: NodeImplementation<SeqTypes>, N: ConnectedNetwork<PubKey>, P: SequencerPersistence>
56    RequestResponseProtocol<I, N, P>
57where
58    I::Storage: NewProtocolStorage<SeqTypes>,
59{
60    /// Create a new RequestResponseProtocol from the inner
61    pub fn new(
62        // The configuration for the protocol
63        config: RequestResponseConfig,
64        // The network sender that [`RequestResponseProtocol`] will use to send messages
65        sender: Sender,
66        // The network receiver that [`RequestResponseProtocol`] will use to receive messages
67        receiver: Receiver<Bytes>,
68        // The recipient source that [`RequestResponseProtocol`] will use to get the recipients
69        // that a specific message should expect responses from
70        recipient_source: RecipientSource<I>,
71        // The [response] data source that [`RequestResponseProtocol`] will use to derive the
72        // response data for a specific request
73        data_source: DataSource<I, N, P>,
74        // The public key of this node
75        public_key: PubKey,
76        // The private key of this node
77        private_key: BLSPrivKey,
78    ) -> Self {
79        Self {
80            inner: RequestResponse::new(
81                config.clone(),
82                sender,
83                receiver,
84                recipient_source,
85                data_source,
86            ),
87            config,
88            public_key,
89            private_key,
90        }
91    }
92}
93
94impl<I: NodeImplementation<SeqTypes>, N: ConnectedNetwork<PubKey>, P: SequencerPersistence>
95    RequestResponseProtocol<I, N, P>
96where
97    I::Storage: NewProtocolStorage<SeqTypes>,
98{
99    pub async fn request_indefinitely<F, Fut, O>(
100        &self,
101        // The request to make
102        request: Request,
103        // The type of request
104        request_type: RequestType,
105        // The response validation function
106        response_validation_fn: F,
107    ) -> std::result::Result<O, RequestError>
108    where
109        F: Fn(&Request, Response) -> Fut + Send + Sync + 'static + Clone,
110        Fut: Future<Output = anyhow::Result<O>> + Send + Sync + 'static,
111        O: Send + Sync + 'static + Clone,
112    {
113        // Request from the inner protocol
114        self.inner
115            .request_indefinitely(
116                &self.public_key,
117                &self.private_key,
118                request_type,
119                self.config.incoming_request_ttl,
120                request,
121                response_validation_fn,
122            )
123            .await
124    }
125}
espresso_node/request_response/mod.rs

espresso_node/request_response/
mod.rs
