[TestLoop] (4/n) Add multi-instance testing support to TestLoop frame…

…work. (near#8587)

Also includes some refactoring from earlier. DelaySender is now a struct wrapping an `Arc<dyn CanSendWithDelay>`, rather than a trait. This follows the Sender design (for non tests). Code is split into separate files. This also makes LoopEventHandler easier to write without having to specify some complex trait bounds (see below).

Two more tests cases are added, one for timed tests (the logic of which is already supported from the previous PR), and one for multi-instance tests added here.

Multi-instance tests are simply tests whose Data type is a Vec<InnerData>, and whose Event type is (usize, InnerEvent). The whole framework works the same way (it does not inherently has the notion of multi-instance tests), but we provide some helper methods so that multi-instance tests can reuse single instance test utils.

Altogether, we provide the following helpers to transform LoopEventHandlers:
 * `.widen()`: Converts a `LoopEventHandler<Data, Event>` to one that can handle some superset Data and a superset Event. This allows reuse of the same event handler test util for different test setups that have different kinds of events and components.
 * `.for_index(usize)`: Converts a `LoopEventHandler<Data, Event>` to `LoopEventHandler<Vec<Data>, (usize, Event)>` to handle a specific instance in a multi-instance test.

and the following helpers to transform DelaySender:
* `.narrow()`: Converts a `DelaySender<A>` into a `DelaySender<B>` where B is a `Into<A>`. This is used for internal implementation of `.widen()` above.
* `.for_index(usize)`: Converts a `DelaySender<(usize, Event)>` into a `DelaySender<Event>` that automatically attaches a specific index. This allows the test loop's `.sender()` to be converted to a single-instance sender to be passed into a component's constructor, so that when that component sends a message, it automatically ends up into the test loop queue with the correct instance index attached.

core/async/Cargo.toml

@@ -18,4 +18,4 @@ once_cell.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 
-near-o11y = { path = "../o11y" }
+near-o11y = { path = "../o11y" }      

core/async/src/examples/mod.rs

@@ -1,2 +1,5 @@
+mod multi_instance_test;
 mod sum_numbers;
 mod sum_numbers_test;
+mod timed_component;
+mod timed_component_test;

core/async/src/examples/multi_instance_test.rs

@@ -0,0 +1,113 @@
+use std::time::Duration;
+
+use derive_enum_from_into::{EnumFrom, EnumTryInto};
+
+use crate::{
+    examples::sum_numbers_test::ForwardSumRequest,
+    messaging::{CanSend, IntoSender},
+    test_loop::{
+        delay_sender::DelaySender,
+        event_handler::{capture_events, LoopEventHandler},
+        TestLoopBuilder,
+    },
+};
+
+use super::sum_numbers::{ReportSumMsg, SumNumbersComponent, SumRequest};
+
+#[derive(derive_more::AsMut, derive_more::AsRef)]
+struct TestData {
+    summer: SumNumbersComponent,
+    sums: Vec<ReportSumMsg>,
+}
+
+#[derive(Debug, EnumTryInto, EnumFrom)]
+enum TestEvent {
+    RemoteRequest(i64),
+    LocalRequest(SumRequest),
+    Sum(ReportSumMsg),
+}
+
+/// Let's pretend that when we send a remote request, the number gets sent to
+/// every other instance in the setup as a local request.
+pub struct ForwardRemoteRequestToOtherInstances {
+    sender: Option<DelaySender<(usize, TestEvent)>>,
+}
+
+impl ForwardRemoteRequestToOtherInstances {
+    pub fn new() -> Self {
+        Self { sender: None }
+    }
+}
+
+impl LoopEventHandler<Vec<TestData>, (usize, TestEvent)> for ForwardRemoteRequestToOtherInstances {
+    fn init(&mut self, sender: DelaySender<(usize, TestEvent)>) {
+        self.sender = Some(sender);
+    }
+
+    fn handle(
+        &mut self,
+        event: (usize, TestEvent),
+        data: &mut Vec<TestData>,
+    ) -> Result<(), (usize, TestEvent)> {
+        if let TestEvent::RemoteRequest(number) = event.1 {
+            for i in 0..data.len() {
+                if i != event.0 {
+                    self.sender
+                        .as_ref()
+                        .unwrap()
+                        .send((i, TestEvent::LocalRequest(SumRequest::Number(number))))
+                }
+            }
+            Ok(())
+        } else {
+            Err(event)
+        }
+    }
+}
+
+#[test]
+fn test_multi_instance() {
+    let builder = TestLoopBuilder::<(usize, TestEvent)>::new();
+    // Build the SumNumberComponents so that it sends messages back to the test loop.
+    let mut data = vec![];
+    for i in 0..5 {
+        data.push(TestData {
+            // Multi-instance sender can be converted to a single-instance sender
+            // so we can pass it into a component's constructor.
+            summer: SumNumbersComponent::new(builder.sender().for_index(i).into_sender()),
+            sums: vec![],
+        });
+    }
+    let sender = builder.sender();
+    let mut test = builder.build(data);
+    test.register_handler(ForwardRemoteRequestToOtherInstances::new());
+    for i in 0..5 {
+        // Single-instance handlers can be reused for multi-instance tests.
+        test.register_handler(ForwardSumRequest.widen().for_index(i));
+        test.register_handler(capture_events::<ReportSumMsg>().widen().for_index(i));
+    }
+
+    // Send a RemoteRequest from each instance.
+    sender.send((0, TestEvent::RemoteRequest(1)));
+    sender.send((1, TestEvent::RemoteRequest(2)));
+    sender.send((2, TestEvent::RemoteRequest(3)));
+    sender.send((3, TestEvent::RemoteRequest(4)));
+    sender.send((4, TestEvent::RemoteRequest(5)));
+
+    // Then send a GetSum request for each instance; we use a delay so that we can ensure
+    // these messages arrive later. (In a real test we wouldn't do this - the component would
+    // automatically emit some events and we would assert on these events. But for this
+    // contrived test we'll do it manually as a demonstration.)
+    for i in 0..5 {
+        sender.send_with_delay(
+            (i, TestEvent::LocalRequest(SumRequest::GetSum)),
+            Duration::from_millis(1),
+        );
+    }
+    test.run(Duration::from_millis(2));
+    assert_eq!(test.data[0].sums, vec![ReportSumMsg(14)]);
+    assert_eq!(test.data[1].sums, vec![ReportSumMsg(13)]);
+    assert_eq!(test.data[2].sums, vec![ReportSumMsg(12)]);
+    assert_eq!(test.data[3].sums, vec![ReportSumMsg(11)]);
+    assert_eq!(test.data[4].sums, vec![ReportSumMsg(10)]);
+}

core/async/src/examples/sum_numbers_test.rs

@@ -4,7 +4,10 @@ use derive_enum_from_into::{EnumFrom, EnumTryInto};
 
 use crate::{
     messaging::{CanSend, IntoSender},
-    test_loop::{CaptureEvents, LoopEventHandler, TestLoopBuilder, TryIntoOrSelf},
+    test_loop::{
+        event_handler::{capture_events, LoopEventHandler},
+        TestLoopBuilder,
+    },
 };
 
 use super::sum_numbers::{ReportSumMsg, SumNumbersComponent, SumRequest};
@@ -26,17 +29,14 @@ enum TestEvent {
 // be reused for any test that needs to send messages to this component.
 pub struct ForwardSumRequest;
 
-impl<Data: AsMut<SumNumbersComponent>, Event: TryIntoOrSelf<SumRequest>>
-    LoopEventHandler<Data, Event> for ForwardSumRequest
-{
-    fn handle(&mut self, event: Event, data: &mut Data) -> Option<Event> {
-        match event.try_into_or_self() {
-            Ok(request) => {
-                data.as_mut().handle(request);
-                None
-            }
-            Err(event) => Some(event),
-        }
+impl LoopEventHandler<SumNumbersComponent, SumRequest> for ForwardSumRequest {
+    fn handle(
+        &mut self,
+        event: SumRequest,
+        data: &mut SumNumbersComponent,
+    ) -> Result<(), SumRequest> {
+        data.handle(event);
+        Ok(())
     }
 }
 
@@ -48,8 +48,8 @@ fn test_simple() {
         TestData { summer: SumNumbersComponent::new(builder.sender().into_sender()), sums: vec![] };
     let sender = builder.sender();
     let mut test = builder.build(data);
-    test.register_handler(ForwardSumRequest);
-    test.register_handler(CaptureEvents::<ReportSumMsg>::new());
+    test.register_handler(ForwardSumRequest.widen());
+    test.register_handler(capture_events::<ReportSumMsg>().widen());
 
     sender.send(TestEvent::Request(SumRequest::Number(1)));
     sender.send(TestEvent::Request(SumRequest::Number(2)));

core/async/src/examples/timed_component.rs

@@ -0,0 +1,28 @@
+use crate::messaging::Sender;
+
+pub(crate) struct TimedComponent {
+    buffered_messages: Vec<String>,
+    message_sender: Sender<Vec<String>>,
+}
+
+/// Mimics a component that has a specific function that is supposed to be
+/// triggered by a timer.
+impl TimedComponent {
+    pub fn new(message_sender: Sender<Vec<String>>) -> Self {
+        Self { buffered_messages: vec![], message_sender }
+    }
+
+    pub fn send_message(&mut self, msg: String) {
+        self.buffered_messages.push(msg);
+    }
+
+    /// This is supposed to be triggered by a timer so it flushes the
+    /// messages every tick.
+    pub fn flush(&mut self) {
+        if self.buffered_messages.is_empty() {
+            return;
+        }
+        self.message_sender.send(self.buffered_messages.clone());
+        self.buffered_messages.clear();
+    }
+}

core/async/src/examples/timed_component_test.rs

@@ -0,0 +1,67 @@
+use std::time::Duration;
+
+use derive_enum_from_into::{EnumFrom, EnumTryInto};
+
+use crate::{
+    messaging::IntoSender,
+    test_loop::event_handler::{capture_events, periodic_interval, LoopEventHandler},
+};
+
+use super::timed_component::TimedComponent;
+
+#[derive(Debug, Clone, PartialEq)]
+struct Flush;
+
+#[derive(Debug, EnumTryInto, EnumFrom)]
+enum TestEvent {
+    SendMessage(String),
+    Flush(Flush),
+    MessageSent(Vec<String>),
+}
+
+#[derive(derive_more::AsMut, derive_more::AsRef)]
+struct TestData {
+    component: TimedComponent,
+    messages_sent: Vec<Vec<String>>,
+}
+
+struct ForwardSendMessage;
+
+impl LoopEventHandler<TimedComponent, String> for ForwardSendMessage {
+    fn handle(&mut self, event: String, data: &mut TimedComponent) -> Result<(), String> {
+        data.send_message(event);
+        Ok(())
+    }
+}
+
+#[test]
+fn test_timed_component() {
+    let builder = crate::test_loop::TestLoopBuilder::<TestEvent>::new();
+    let data = TestData {
+        component: TimedComponent::new(builder.sender().into_sender()),
+        messages_sent: vec![],
+    };
+    let sender = builder.sender();
+    let mut test = builder.build(data);
+    test.register_handler(ForwardSendMessage.widen());
+    test.register_handler(
+        periodic_interval(Duration::from_millis(100), Flush, |data: &mut TimedComponent| {
+            data.flush()
+        })
+        .widen(),
+    );
+    test.register_handler(capture_events::<Vec<String>>().widen());
+
+    sender.send_with_delay("Hello".to_string().into(), Duration::from_millis(10));
+    sender.send_with_delay("World".to_string().into(), Duration::from_millis(20));
+    // The timer fires at 100ms here and flushes "Hello" and "World".
+    sender.send_with_delay("!".to_string().into(), Duration::from_millis(110));
+    // The timer fires again at 200ms here and flushes "!"".
+    // Further timer events do not send messages.
+
+    test.run(Duration::from_secs(1));
+    assert_eq!(
+        test.data.messages_sent,
+        vec![vec!["Hello".to_string(), "World".to_string()], vec!["!".to_string()]]
+    );
+}