Cuprate/cuprate
Cuprate/cuprate
1
0
Fork 0
mirror of https://github.com/Cuprate/cuprate.git synced 2024-11-16 15:58:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix message requesting in Connection

Browse source
This commit is contained in:
Boog900 2024-01-13 13:22:34 +00:00
parent 478a8c1545
commit a0b9280801
No known key found for this signature in database
GPG key ID: 5401367FB7302004
3 changed files with 25 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
net/levin/src/lib.rs
View file

@ -64,7 +64,7 @@ pub enum BucketError {
#[error("Levin header had incorrect signature")]
InvalidHeaderSignature,
/// Error decoding the body
#[error("Error decoding bucket body")]
#[error("Error decoding bucket body: {0}")]
BodyDecodingError(Box<dyn std::error::Error + Send + Sync>),
/// Unknown command ID
#[error("Unknown command ID")]

28
p2p/monero-p2p/src/client/connection.rs
View file

@ -90,6 +90,8 @@ where
}
async fn handle_client_request(&mut self, req: ConnectionTaskRequest) -> Result<(), PeerError> {
tracing::debug!("handling client request, id: {:?}", req.request.id());
if req.request.needs_response() {
self.state = State::WaitingForResponse {
request_id: req.request.id(),
@ -103,6 +105,8 @@ where
}
async fn handle_peer_request(&mut self, req: PeerRequest) -> Result<(), PeerError> {
tracing::debug!("Received peer request: {:?}", req.id());
let ready_svc = self.peer_request_handler.ready().await?;
let res = ready_svc.call(req).await?;
if matches!(res, PeerResponse::NA) {
@ -113,6 +117,8 @@ where
}
async fn handle_potential_response(&mut self, mes: Message) -> Result<(), PeerError> {
tracing::debug!("Received peer message, command: {:?}", mes.command());
if mes.is_request() {
return self.handle_peer_request(mes.try_into().unwrap()).await;
}
@ -145,6 +151,7 @@ where
where
Str: FusedStream<Item = Result<Message, monero_wire::BucketError>> + Unpin,
{
tracing::debug!("waiting for peer/client request.");
tokio::select! {
biased;
bradcast_req = self.broadcast_rx.next() => {
@ -152,15 +159,17 @@ where
}
client_req = self.client_rx.next() => {
if let Some(client_req) = client_req {
self.handle_client_request(client_req).await?
self.handle_client_request(client_req).await
} else {
Err(PeerError::ClientChannelClosed)
}
Err(PeerError::ClientChannelClosed)
},
peer_message = stream.next() => {
if let Some(peer_message) = peer_message {
self.handle_peer_request(peer_message?.try_into().map_err(|_| PeerError::PeerSentInvalidMessage)?).await?
self.handle_peer_request(peer_message?.try_into().map_err(|_| PeerError::PeerSentInvalidMessage)?).await
}else {
Err(PeerError::ClientChannelClosed)
}
Err(PeerError::ConnectionClosed)
},
}
}
@ -169,6 +178,7 @@ where
where
Str: FusedStream<Item = Result<Message, monero_wire::BucketError>> + Unpin,
{
tracing::debug!("waiting for peer response..");
tokio::select! {
biased;
bradcast_req = self.broadcast_rx.next() => {
@ -176,9 +186,10 @@ where
}
peer_message = stream.next() => {
if let Some(peer_message) = peer_message {
self.handle_peer_request(peer_message?.try_into().map_err(|_| PeerError::PeerSentInvalidMessage)?).await?
self.handle_potential_response(peer_message?).await
}else {
Err(PeerError::ClientChannelClosed)
}
Err(PeerError::ConnectionClosed)
},
}
}
@ -187,6 +198,10 @@ where
where
Str: FusedStream<Item = Result<Message, monero_wire::BucketError>> + Unpin,
{
tracing::debug!(
"Handling eager messages len: {}",
eager_protocol_messages.len()
);
for message in eager_protocol_messages {
let message = Message::Protocol(message).try_into();
@ -202,6 +217,7 @@ where
loop {
if self.connection_guard.should_shutdown() {
tracing::debug!("connection guard has shutdown, shutting down connection.");
return self.shutdown(PeerError::ConnectionClosed);
}

4
p2p/monero-p2p/tests/utils.rs
View file

@ -86,10 +86,10 @@ impl Service<PeerRequest> for DummyPeerRequestHandlerSvc {
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
todo!()
Poll::Ready(Ok(()))
}
fn call(&mut self, _: PeerRequest) -> Self::Future {
todo!()
async move { Ok(PeerResponse::NA) }.boxed()
}
}