Implement ExchangeServerHandler

node/Cargo.toml

@@ -7,6 +7,7 @@ license = "Apache-2.0"
 [dependencies]
 celestia-proto = { workspace = true }
 celestia-types = { workspace = true }
+tendermint = { workspace = true }
 tendermint-proto = { workspace = true }
 
 async-trait = "0.1.73"

node/src/exchange.rs

@@ -36,9 +36,11 @@ const RESPONSE_SIZE_MAXIMUM: usize = 10 * 1024 * 1024;
 /// Maximum length of the protobuf length delimiter in bytes
 const PROTOBUF_MAX_LENGTH_DELIMITER_LEN: usize = 10;
 
+type RequestType = HeaderRequest;
+type ResponseType = Vec<HeaderResponse>;
 type ReqRespBehaviour = request_response::Behaviour<HeaderCodec>;
-type ReqRespEvent = request_response::Event<HeaderRequest, Vec<HeaderResponse>>;
-type ReqRespMessage = request_response::Message<HeaderRequest, Vec<HeaderResponse>>;
+type ReqRespEvent = request_response::Event<RequestType, ResponseType>;
+type ReqRespMessage = request_response::Message<RequestType, ResponseType>;
 
 pub(crate) struct ExchangeBehaviour<S>
 where
@@ -226,13 +228,20 @@ where
         cx: &mut Context<'_>,
         params: &mut impl PollParameters,
     ) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
-        while let Poll::Ready(ev) = self.req_resp.poll(cx, params) {
-            if let Some(ev) = self.on_to_swarm(ev) {
-                return Poll::Ready(ev);
+        loop {
+            while let Poll::Ready(ev) = self.req_resp.poll(cx, params) {
+                if let Some(ev) = self.on_to_swarm(ev) {
+                    return Poll::Ready(ev);
+                }
+
+                continue;
+            }
+            if self.server_handler.poll(cx, &mut self.req_resp).is_ready() {
+                continue;
             }
-        }
 
-        Poll::Pending
+            return Poll::Pending;
+        }
     }
 }
 

node/src/exchange/client.rs

@@ -22,7 +22,7 @@ use crate::utils::{OneshotResultSender, OneshotResultSenderExt};
 
 pub(super) struct ExchangeClientHandler<S = ReqRespBehaviour>
 where
-    S: Sender,
+    S: RequestSender,
 {
     reqs: HashMap<S::RequestId, State>,
     peer_tracker: Arc<PeerTracker>,
@@ -33,13 +33,13 @@ struct State {
     respond_to: OneshotResultSender<Vec<ExtendedHeader>, P2pError>,
 }
 
-pub(super) trait Sender {
+pub(super) trait RequestSender {
     type RequestId: Hash + Eq + Debug;
 
     fn send_request(&mut self, peer: &PeerId, request: HeaderRequest) -> Self::RequestId;
 }
 
-impl Sender for ReqRespBehaviour {
+impl RequestSender for ReqRespBehaviour {
     type RequestId = RequestId;
 
     fn send_request(&mut self, peer: &PeerId, request: HeaderRequest) -> RequestId {
@@ -49,7 +49,7 @@ impl Sender for ReqRespBehaviour {
 
 impl<S> ExchangeClientHandler<S>
 where
-    S: Sender,
+    S: RequestSender,
 {
     pub(super) fn new(peer_tracker: Arc<PeerTracker>) -> Self {
         ExchangeClientHandler {
@@ -880,7 +880,7 @@ mod tests {
         peer: PeerId,
     }
 
-    impl Sender for MockReq {
+    impl RequestSender for MockReq {
         type RequestId = MockReqId;
 
         fn send_request(&mut self, peer: &PeerId, _request: HeaderRequest) -> Self::RequestId {