diff --git a/clients/native/examples/websocket_binarysend.rs b/clients/native/examples/websocket_binarysend.rs index 160a9bf009..f103815af2 100644 --- a/clients/native/examples/websocket_binarysend.rs +++ b/clients/native/examples/websocket_binarysend.rs @@ -43,7 +43,7 @@ async fn send_file_with_reply() { recipient, message: read_data, with_reply_surb: true, - connection_id: 0, + connection_id: Some(0), }; println!("sending content of 'dummy_file' over the mix network..."); @@ -92,7 +92,7 @@ async fn send_file_without_reply() { recipient, message: read_data, with_reply_surb: false, - connection_id: 0, + connection_id: Some(0), }; println!("sending content of 'dummy_file' over the mix network..."); diff --git a/clients/native/src/websocket/handler.rs b/clients/native/src/websocket/handler.rs index 0daa171532..11d0dda55f 100644 --- a/clients/native/src/websocket/handler.rs +++ b/clients/native/src/websocket/handler.rs @@ -92,26 +92,42 @@ impl Handler { recipient: &Recipient, message: Vec, with_reply_surb: bool, - connection_id: u64, + connection_id: Option, ) -> Option { + // We map the absence of a connection id as going into the general lane. + let lane = connection_id.map_or(TransmissionLane::General, |id| { + TransmissionLane::ConnectionId(id) + }); + // the ack control is now responsible for chunking, etc. - let lane = TransmissionLane::ConnectionId(connection_id); let input_msg = InputMessage::new_fresh(*recipient, message, with_reply_surb, lane); self.msg_input .send(input_msg) .await .expect("InputMessageReceiver has stopped receiving!"); + // Only reply back with a `LaneQueueLength` if the sender providided a connection id + let connection_id = match lane { + TransmissionLane::General + | TransmissionLane::Reply + | TransmissionLane::Retransmission + | TransmissionLane::Control => return None, + TransmissionLane::ConnectionId(id) => id, + }; + // on receiving a send, we reply back the current lane queue length for that connection id. // Note that this does _NOT_ take into account the packets that have been received but not // yet reach `OutQueueControl`, so it might be a tad low. - if let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() { - let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0); - return Some(ServerResponse::LaneQueueLength(connection_id, queue_length)); - } + let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() else { + log::warn!( + "Failed to get the lane queue length lock, \ + not responding back with the current queue length" + ); + return None; + }; - log::warn!("Failed to get the lane queue length lock, not responding back with the current queue length"); - None + let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0); + Some(ServerResponse::LaneQueueLength(connection_id, queue_length)) } async fn handle_reply( @@ -120,7 +136,13 @@ impl Handler { message: Vec, ) -> Option { if message.len() > ReplySurb::max_msg_len(Default::default()) { - return Some(ServerResponse::new_error(format!("too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes", message.len(), ReplySurb::max_msg_len(Default::default())))); + return Some( + ServerResponse::new_error( + format!( + "too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes", + message.len(), ReplySurb::max_msg_len(Default::default())) + ) + ); } let input_msg = InputMessage::new_reply(reply_surb, message); diff --git a/clients/native/websocket-requests/src/requests.rs b/clients/native/websocket-requests/src/requests.rs index 9b5af6e70d..fafefb5827 100644 --- a/clients/native/websocket-requests/src/requests.rs +++ b/clients/native/websocket-requests/src/requests.rs @@ -34,7 +34,7 @@ pub enum ClientRequest { message: Vec, // Perhaps we could change it to a number to indicate how many reply_SURBs we want to include? with_reply_surb: bool, - connection_id: u64, + connection_id: Option, }, Reply { message: Vec, @@ -53,10 +53,10 @@ impl ClientRequest { recipient: Recipient, data: Vec, with_reply_surb: bool, - connection_id: u64, + connection_id: Option, ) -> Vec { let data_len_bytes = (data.len() as u64).to_be_bytes(); - let conn_id_bytes = connection_id.to_be_bytes(); + let conn_id_bytes = connection_id.unwrap_or(0).to_be_bytes(); std::iter::once(SEND_REQUEST_TAG) .chain(std::iter::once(with_reply_surb as u8)) .chain(recipient.to_bytes().iter().cloned()) // will not be length prefixed because the length is constant @@ -106,6 +106,11 @@ impl ClientRequest { connection_id_bytes .copy_from_slice(&b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::()]); let connection_id = u64::from_be_bytes(connection_id_bytes); + let connection_id = if connection_id == 0 { + None + } else { + Some(connection_id) + }; let data_len_bytes = &b[2 + Recipient::LEN + size_of::()..2 + Recipient::LEN + 2 * size_of::()]; @@ -350,7 +355,7 @@ mod tests { recipient, message: b"foomp".to_vec(), with_reply_surb: false, - connection_id: 42, + connection_id: Some(42), }; let bytes = send_request_no_surb.serialize(); @@ -365,7 +370,7 @@ mod tests { assert_eq!(recipient.to_string(), recipient_string); assert_eq!(message, b"foomp".to_vec()); assert!(!with_reply_surb); - assert_eq!(connection_id, 42) + assert_eq!(connection_id, Some(42)) } _ => unreachable!(), } @@ -374,7 +379,7 @@ mod tests { recipient, message: b"foomp".to_vec(), with_reply_surb: true, - connection_id: 213, + connection_id: None, }; let bytes = send_request_surb.serialize(); @@ -389,7 +394,7 @@ mod tests { assert_eq!(recipient.to_string(), recipient_string); assert_eq!(message, b"foomp".to_vec()); assert!(with_reply_surb); - assert_eq!(connection_id, 213) + assert_eq!(connection_id, None) } _ => unreachable!(), } diff --git a/clients/native/websocket-requests/src/text.rs b/clients/native/websocket-requests/src/text.rs index cca6569f4f..1083ecd9a8 100644 --- a/clients/native/websocket-requests/src/text.rs +++ b/clients/native/websocket-requests/src/text.rs @@ -20,7 +20,7 @@ pub(super) enum ClientRequestText { message: String, recipient: String, with_reply_surb: bool, - connection_id: u64, + connection_id: Option, }, SelfAddress, #[serde(rename_all = "camelCase")] diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 80528d1bed..870376d896 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -102,7 +102,7 @@ impl ServiceProvider { recipient: return_address, message: msg.into_bytes(), with_reply_surb: false, - connection_id: conn_id, + connection_id: Some(conn_id), }; let message = Message::Binary(response_message.serialize());