pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-05-31 19:11:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove queueing from message-lane (#352)

Browse Source 
* remove queueing from message-lane

* also remove queueing from RPCs

* another trace

* new clippy
This commit is contained in:
Svyatoslav Nikolsky
2020-09-15 15:14:52 +03:00
committed by Bastian Köcher
parent c2791c2772
commit d918bcb6f8
8 changed files with 29 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bridges/modules/message-lane/src/inbound_lane.rs
+13 -152
View File
@@ -16,7 +16,7 @@
//! Everything about incoming messages receival.
use bp_message_lane::{InboundLaneData, LaneId, Message, MessageKey, MessageNonce, MessageResult, OnMessageReceived};
use bp_message_lane::{InboundLaneData, LaneId, Message, MessageKey, MessageNonce, OnMessageReceived};
/// Inbound lane storage.
pub trait InboundLaneStorage {
@@ -29,12 +29,6 @@ pub trait InboundLaneStorage {
fn data(&self) -> InboundLaneData;
/// Update lane data in the storage.
fn set_data(&mut self, data: InboundLaneData);
/// Returns saved inbound message payload.
fn message(&self, nonce: &MessageNonce) -> Option<Self::Payload>;
/// Save inbound message in the storage.
fn save_message(&mut self, nonce: MessageNonce, payload: Self::Payload);
/// Remove inbound message from the storage.
fn remove_message(&mut self, nonce: &MessageNonce);
}
/// Inbound messages lane.
@@ -49,11 +43,10 @@ impl<S: InboundLaneStorage> InboundLane<S> {
}
/// Receive new message.
pub fn receive_message(
pub fn receive_message<P: OnMessageReceived<S::Payload>>(
&mut self,
nonce: MessageNonce,
payload: S::Payload,
processor: &mut impl OnMessageReceived<S::Payload>,
) -> bool {
let mut data = self.storage.data();
let is_correct_message = nonce == data.latest_received_nonce + 1;
@@ -61,70 +54,19 @@ impl<S: InboundLaneStorage> InboundLane<S> {
return false;
}
let is_process_required = is_correct_message && data.oldest_unprocessed_nonce == nonce;
data.latest_received_nonce = nonce;
self.storage.set_data(data);
let payload_to_save = match is_process_required {
true => {
let message = Message {
key: MessageKey {
lane_id: self.storage.id(),
nonce,
},
payload,
};
match processor.on_message_received(message) {
MessageResult::Processed => None,
MessageResult::NotProcessed(message) => Some(message.payload),
}
}
false => Some(payload),
};
if let Some(payload_to_save) = payload_to_save {
self.storage.save_message(nonce, payload_to_save);
}
P::on_message_received(Message {
key: MessageKey {
lane_id: self.storage.id(),
nonce,
},
payload,
});
true
}
/// Process stored lane messages.
///
/// Stops processing either when all messages are processed, or when processor returns
/// MessageResult::NotProcessed.
pub fn process_messages(&mut self, processor: &mut impl OnMessageReceived<S::Payload>) {
let mut anything_processed = false;
let mut data = self.storage.data();
while data.oldest_unprocessed_nonce <= data.latest_received_nonce {
let nonce = data.oldest_unprocessed_nonce;
let payload = self
.storage
.message(&nonce)
.expect("message is referenced by lane; referenced message is not pruned; qed");
let message = Message {
key: MessageKey {
lane_id: self.storage.id(),
nonce,
},
payload,
};
let process_result = processor.on_message_received(message);
if let MessageResult::NotProcessed(_) = process_result {
break;
}
self.storage.remove_message(&nonce);
anything_processed = true;
data.oldest_unprocessed_nonce += 1;
}
if anything_processed {
self.storage.set_data(data);
}
}
}
#[cfg(test)]
@@ -132,105 +74,24 @@ mod tests {
use super::*;
use crate::{
inbound_lane,
mock::{
run_test, TestMessageProcessor, TestPayload, TestRuntime, PAYLOAD_TO_QUEUE, REGULAR_PAYLOAD, TEST_LANE_ID,
},
mock::{run_test, TestRuntime, REGULAR_PAYLOAD, TEST_LANE_ID},
};
#[test]
fn fails_to_receive_message_with_incorrect_nonce() {
run_test(|| {
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert!(!lane.receive_message(10, REGULAR_PAYLOAD, &mut TestMessageProcessor));
assert!(lane.storage.message(&10).is_none());
assert!(!lane.receive_message::<()>(10, REGULAR_PAYLOAD));
assert_eq!(lane.storage.data().latest_received_nonce, 0);
});
}
#[test]
fn correct_message_is_queued_if_some_other_messages_are_queued() {
fn correct_message_is_processed_instantly() {
run_test(|| {
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert!(lane.receive_message(1, PAYLOAD_TO_QUEUE, &mut TestMessageProcessor));
assert!(lane.storage.message(&1).is_some());
assert!(lane.receive_message(2, REGULAR_PAYLOAD, &mut TestMessageProcessor));
assert!(lane.storage.message(&2).is_some());
assert_eq!(lane.storage.data().latest_received_nonce, 2);
});
}
#[test]
fn correct_message_is_queued_if_processor_wants_to_queue() {
run_test(|| {
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert!(lane.receive_message(1, PAYLOAD_TO_QUEUE, &mut TestMessageProcessor));
assert!(lane.storage.message(&1).is_some());
assert!(lane.receive_message::<()>(1, REGULAR_PAYLOAD));
assert_eq!(lane.storage.data().latest_received_nonce, 1);
});
}
#[test]
fn correct_message_is_not_queued_if_processed_instantly() {
run_test(|| {
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert!(lane.receive_message(1, REGULAR_PAYLOAD, &mut TestMessageProcessor));
assert!(lane.storage.message(&1).is_none());
assert_eq!(lane.storage.data().latest_received_nonce, 1);
});
}
#[test]
fn process_message_does_nothing_when_lane_is_empty() {
run_test(|| {
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert_eq!(lane.storage.data().oldest_unprocessed_nonce, 1);
lane.process_messages(&mut TestMessageProcessor);
assert_eq!(lane.storage.data().oldest_unprocessed_nonce, 1);
});
}
#[test]
fn process_message_works() {
run_test(|| {
pub struct QueueByNonce(MessageNonce);
impl OnMessageReceived<TestPayload> for QueueByNonce {
fn on_message_received(&mut self, message: Message<TestPayload>) -> MessageResult<TestPayload> {
if message.key.nonce == self.0 {
MessageResult::NotProcessed(message)
} else {
MessageResult::Processed
}
}
}
let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
assert!(lane.receive_message(1, PAYLOAD_TO_QUEUE, &mut TestMessageProcessor));
assert!(lane.receive_message(2, PAYLOAD_TO_QUEUE, &mut TestMessageProcessor));
assert!(lane.receive_message(3, PAYLOAD_TO_QUEUE, &mut TestMessageProcessor));
assert!(lane.receive_message(4, REGULAR_PAYLOAD, &mut TestMessageProcessor));
assert!(lane.storage.message(&1).is_some());
assert!(lane.storage.message(&2).is_some());
assert!(lane.storage.message(&3).is_some());
assert!(lane.storage.message(&4).is_some());
assert_eq!(lane.storage.data().oldest_unprocessed_nonce, 1);
lane.process_messages(&mut QueueByNonce(3));
assert!(lane.storage.message(&1).is_none());
assert!(lane.storage.message(&2).is_none());
assert!(lane.storage.message(&3).is_some());
assert!(lane.storage.message(&4).is_some());
assert_eq!(lane.storage.data().oldest_unprocessed_nonce, 3);
lane.process_messages(&mut QueueByNonce(10));
assert!(lane.storage.message(&1).is_none());
assert!(lane.storage.message(&2).is_none());
assert!(lane.storage.message(&3).is_none());
assert!(lane.storage.message(&4).is_none());
assert_eq!(lane.storage.data().oldest_unprocessed_nonce, 5);
});
}
}