Add mocking check support

grpc/broadcaster/Cargo.toml

@@ -17,5 +17,8 @@ tokio-stream = { version = "0.1", features = ["sync"] }
 tonic = { version = "0.9" }
 prost = { version = "0.11" }
 
+[dev-dependencies]
+mockall = "0.11"
+
 [build-dependencies]
 tonic-build = { version = "0.9" }

grpc/broadcaster/src/main.rs

@@ -1,6 +1,8 @@
 // SPDX-FileCopyrightText: 2023 Matteo Settenvini <matteo.settenvini@montecristosoftware.eu>
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use tonic::transport::server::TcpIncoming;
+
 mod pb {
     tonic::include_proto!("package");
 }
@@ -14,7 +16,6 @@ use {
     std::pin::Pin,
     std::sync::{Arc, Mutex, Weak},
     tokio::sync::broadcast,
-    tokio::task::JoinHandle,
     tokio_stream::wrappers::BroadcastStream,
     tonic::{Request, Response, Status, Streaming},
 };
@@ -24,18 +25,22 @@ const RENDEZVOUS: &'static str = "[::1]:10000";
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
-    let server_handle = run_server().await;
-    Ok(server_handle.await.unwrap()?)
+    let url = RENDEZVOUS.parse()?;
+    let incoming = tonic::transport::server::TcpIncoming::new(url, true, None)
+        .expect("Cannot bind server socket");
+    run_server(incoming).await?;
+    Ok(())
 }
 
-async fn run_server() -> JoinHandle<Result<(), tonic::transport::Error>> {
+fn run_server(
+    incoming: TcpIncoming,
+) -> impl std::future::Future<Output = Result<(), tonic::transport::Error>> {
     use tonic::transport::Server;
 
-    let addr = RENDEZVOUS.parse().unwrap();
     let service = MessageService::default();
     let svc_adapter = ServiceServer::new(service);
     let builder = Server::builder().add_service(svc_adapter);
-    tokio::spawn(builder.serve(addr))
+    builder.serve_with_incoming(incoming)
 }
 
 #[derive(Default, Debug)]
@@ -94,24 +99,32 @@ mod test {
         super::pb::service_client::ServiceClient,
         super::run_server,
         super::{Message, RENDEZVOUS},
+        mockall::automock,
         tonic::Request,
     };
 
+    #[automock]
+    trait MessageChecker {
+        fn check_contents(&self, msg: &Message);
+    }
+
     #[tokio::test]
     async fn bidi_streaming() -> Result<(), Box<dyn std::error::Error>> {
-        let _server_handle = run_server().await;
+        let url = RENDEZVOUS.parse()?;
+        let incoming = tonic::transport::server::TcpIncoming::new(url, true, None)
+            .expect("Cannot bind server socket");
+        tokio::spawn(run_server(incoming));
 
-        // FIXME: avoid sleep waiting for server to start
-        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+        const N_MESSAGES: usize = 20;
+        const CLIENT_IDS: &[&str] = &["client AAA", "client BBB", "client CCC"];
 
-        async fn client(id: &str) {
-            let mut client = ServiceClient::connect(format!("http://{}", RENDEZVOUS))
-                .await
-                .unwrap();
+        async fn client(id: &str, mock: &MockMessageChecker) {
+            let addr = format!("http://{}", RENDEZVOUS);
+            let mut client = ServiceClient::connect(addr).await.unwrap();
 
             let client_id = String::from(id);
             let outbound = async_stream::stream! {
-                for i in 1..21 {
+                for i in 1..(N_MESSAGES+1) {
                     tokio::task::yield_now().await;
                     let message = Message {
                         contents: format!("{}: {}", client_id, i),
@@ -123,16 +136,31 @@ mod test {
 
             let response = client.broadcaster(Request::new(outbound)).await.unwrap();
             let mut inbound = response.into_inner();
-
             while let Some(msg) = inbound.message().await.unwrap() {
                 println!("{} received {:?}", id, msg);
+                mock.check_contents(&msg);
             }
         }
 
-        let c1 = client("client AAA");
-        let c2 = client("client BBB");
-        let c3 = client("client CCC");
-        let _ = tokio::join!(c1, c2, c3);
+        let mut mock = MockMessageChecker::new();
+
+        let mut clients = vec![];
+        for client_id in CLIENT_IDS.iter() {
+            for i in 1..(N_MESSAGES + 1) {
+                let expected = format!("{}: {}", client_id, i);
+                mock.expect_check_contents()
+                    .times(CLIENT_IDS.len())
+                    .withf(move |msg| msg.contents == expected)
+                    .return_const(());
+            }
+        }
+
+        for client_id in CLIENT_IDS {
+            let c = client(client_id, &mock);
+            clients.push(c);
+        }
+
+        let _ = futures::future::join_all(clients).await;
         println!("Exiting.");
         Ok(())
     }