GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
Jon Häggblad
2022-11-22 11:49:35 +01:00
parent 31778c6e51
commit 1c379fa0db
1 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service-providers/network-requester/src/core.rs
+18 -8
View File
@@ -117,6 +117,20 @@ impl ServiceProvider {
}
}
fn handle_lane_queue_length_response(
lane_queue_lengths: &LaneQueueLengths,
lane: u64,
queue_length: usize,
) {
log::info!("received LaneQueueLength lane: {lane}, queue_length: {queue_length}");
if let Ok(mut lane_queue_lengths) = lane_queue_lengths.lock() {
let lane = TransmissionLane::ConnectionId(lane);
lane_queue_lengths.map.insert(lane, queue_length);
} else {
log::warn!("Unable to lock lane queue lengths, skipping updating received lane length")
}
}
async fn read_websocket_message(
websocket_reader: &mut SplitStream<TSWebsocketStream>,
lane_queue_lengths: LaneQueueLengths,
@@ -141,15 +155,11 @@ impl ServiceProvider {
let received = match deserialized_message {
ServerResponse::Received(received) => received,
ServerResponse::LaneQueueLength(lane, queue_length) => {
log::info!(
"received LaneQueueLength lane: {lane}, queue_length: {queue_length}"
Self::handle_lane_queue_length_response(
&lane_queue_lengths,
lane,
queue_length,
);
if let Ok(mut lane_queue_lengths) = lane_queue_lengths.lock() {
let lane = TransmissionLane::ConnectionId(lane);
lane_queue_lengths.map.insert(lane, queue_length);
} else {
log::warn!("Unable to lock lane queue lengths, skipping updating received lane length")
}
continue;
}
ServerResponse::Error(err) => {