ab_networking/protocols/request_response/handlers/
generic_request_handler.rs

1//! Generic request-response handler, typically is used with a type implementing [`GenericRequest`]
2//! to significantly reduce boilerplate when implementing [`RequestHandler`].
3
4use crate::protocols::request_response::request_response_factory::{
5    IncomingRequest, OutgoingResponse, ProtocolConfig, RequestHandler,
6};
7use async_trait::async_trait;
8use futures::channel::mpsc;
9use futures::prelude::*;
10use libp2p::PeerId;
11use parity_scale_codec::{Decode, Encode};
12use std::fmt;
13use std::pin::Pin;
14use std::sync::Arc;
15use tracing::{debug, trace};
16
17/// Could be changed after the production feedback.
18const REQUESTS_BUFFER_SIZE: usize = 50;
19
20/// Generic request with associated response
21pub trait GenericRequest: Encode + Decode + Send + Sync + 'static {
22    /// Defines request-response protocol name.
23    const PROTOCOL_NAME: &'static str;
24    /// Specifies log-parameters for tracing.
25    const LOG_TARGET: &'static str;
26    /// Response type that corresponds to this request
27    type Response: Encode + Decode + Send + Sync + 'static;
28}
29
30type RequestHandlerFn<Request> = Arc<
31    dyn (Fn(
32            PeerId,
33            Request,
34        )
35            -> Pin<Box<dyn Future<Output = Option<<Request as GenericRequest>::Response>> + Send>>)
36        + Send
37        + Sync
38        + 'static,
39>;
40
41/// Defines generic request-response protocol handler.
42pub struct GenericRequestHandler<Request>
43where
44    Request: GenericRequest,
45{
46    request_receiver: mpsc::Receiver<IncomingRequest>,
47    request_handler: RequestHandlerFn<Request>,
48    protocol_config: ProtocolConfig,
49}
50
51impl<Request> fmt::Debug for GenericRequestHandler<Request>
52where
53    Request: GenericRequest,
54{
55    #[inline]
56    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
57        f.debug_struct("GenericRequestHandler")
58            .field("protocol_name", &Request::PROTOCOL_NAME)
59            .finish_non_exhaustive()
60    }
61}
62
63impl<Request> GenericRequestHandler<Request>
64where
65    Request: GenericRequest,
66{
67    /// Creates new [`GenericRequestHandler`] by given handler.
68    pub fn create<RH, Fut>(request_handler: RH) -> Box<dyn RequestHandler>
69    where
70        RH: (Fn(PeerId, Request) -> Fut) + Send + Sync + 'static,
71        Fut: Future<Output = Option<Request::Response>> + Send + 'static,
72    {
73        let (request_sender, request_receiver) = mpsc::channel(REQUESTS_BUFFER_SIZE);
74
75        let mut protocol_config = ProtocolConfig::new(Request::PROTOCOL_NAME);
76        protocol_config.inbound_queue = Some(request_sender);
77
78        Box::new(Self {
79            request_receiver,
80            request_handler: Arc::new(move |peer_id, request| {
81                Box::pin(request_handler(peer_id, request))
82            }),
83            protocol_config,
84        })
85    }
86
87    /// Invokes external protocol handler.
88    async fn handle_request(
89        &self,
90        peer: PeerId,
91        payload: Vec<u8>,
92    ) -> Result<Vec<u8>, RequestHandlerError> {
93        trace!(%peer, protocol=Request::LOG_TARGET, "Handling request...");
94        let request = Request::decode(&mut payload.as_slice())
95            .map_err(|_| RequestHandlerError::InvalidRequestFormat)?;
96        let response = (self.request_handler)(peer, request).await;
97
98        Ok(response.ok_or(RequestHandlerError::NoResponse)?.encode())
99    }
100}
101
102#[async_trait]
103impl<Request> RequestHandler for GenericRequestHandler<Request>
104where
105    Request: GenericRequest,
106{
107    /// Run [`RequestHandler`].
108    async fn run(&mut self) {
109        while let Some(request) = self.request_receiver.next().await {
110            let IncomingRequest {
111                peer,
112                payload,
113                pending_response,
114            } = request;
115
116            match self.handle_request(peer, payload).await {
117                Ok(response_data) => {
118                    let response = OutgoingResponse {
119                        result: Ok(response_data),
120                        sent_feedback: None,
121                    };
122
123                    match pending_response.send(response) {
124                        Ok(()) => trace!(target = Request::LOG_TARGET, %peer, "Handled request",),
125                        Err(_) => debug!(
126                            target = Request::LOG_TARGET,
127                            protocol = Request::PROTOCOL_NAME,
128                            %peer,
129                            "Failed to handle request: {}",
130                            RequestHandlerError::SendResponse
131                        ),
132                    };
133                }
134                Err(e) => {
135                    debug!(
136                        target = Request::LOG_TARGET,
137                        protocol = Request::PROTOCOL_NAME,
138                        %e,
139                        "Failed to handle request.",
140                    );
141
142                    let response = OutgoingResponse {
143                        result: Err(()),
144                        sent_feedback: None,
145                    };
146
147                    if pending_response.send(response).is_err() {
148                        debug!(
149                            target = Request::LOG_TARGET,
150                            protocol = Request::PROTOCOL_NAME,
151                            %peer,
152                            "Failed to handle request: {}", RequestHandlerError::SendResponse
153                        );
154                    };
155                }
156            }
157        }
158    }
159
160    fn protocol_config(&self) -> ProtocolConfig {
161        self.protocol_config.clone()
162    }
163
164    fn protocol_name(&self) -> &'static str {
165        Request::PROTOCOL_NAME
166    }
167
168    fn clone_box(&self) -> Box<dyn RequestHandler> {
169        let (request_sender, request_receiver) = mpsc::channel(REQUESTS_BUFFER_SIZE);
170
171        let mut protocol_config = ProtocolConfig::new(Request::PROTOCOL_NAME);
172        protocol_config.inbound_queue = Some(request_sender);
173
174        Box::new(Self {
175            request_receiver,
176            request_handler: Arc::clone(&self.request_handler),
177            protocol_config,
178        })
179    }
180}
181
182#[derive(Debug, thiserror::Error)]
183enum RequestHandlerError {
184    #[error("Failed to send response.")]
185    SendResponse,
186
187    #[error("Incorrect request format.")]
188    InvalidRequestFormat,
189
190    #[error("No response.")]
191    NoResponse,
192}
ab_networking/protocols/request_response/handlers/generic_request_handler.rs

ab_networking/protocols/request_response/handlers/
generic_request_handler.rs
