test(dmq): make publisher integration test check client/server deconnection

Author: jpraynaud

internal/mithril-dmq/tests/publisher_client_server.rs

@@ -19,10 +19,8 @@ async fn create_fake_msg(bytes: &[u8]) -> DmqMessage {
     let dmq_builder = DmqMessageBuilder::new(
         {
             let (kes_signature, operational_certificate) = KesSignerFake::dummy_signature();
-            let kes_signer = KesSignerFake::new(vec![
-                Ok((kes_signature, operational_certificate.clone())),
-                Ok((kes_signature, operational_certificate.clone())), // TODO: remove this line once the hack of KES signature is removed in DMQ message builder
-            ]);
+            let kes_signer =
+                KesSignerFake::new(vec![Ok((kes_signature, operational_certificate.clone()))]);
 
             Arc::new(kes_signer)
         },
@@ -41,6 +39,7 @@ async fn dmq_publisher_client_server() {
             .join("node.socket");
     let (stop_tx, stop_rx) = watch::channel(());
 
+    // Start the server
     let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
     let server = tokio::spawn({
         let socket_path = socket_path.clone();
@@ -59,6 +58,7 @@ async fn dmq_publisher_client_server() {
         }
     });
 
+    // Start a first client, publish messages and wait for its deconnection
     let client = tokio::spawn({
         let socket_path = socket_path.clone();
         async move {
@@ -68,8 +68,6 @@ async fn dmq_publisher_client_server() {
                     let kes_signer = KesSignerFake::new(vec![
                         Ok((kes_signature, operational_certificate.clone())),
                         Ok((kes_signature, operational_certificate.clone())),
-                        Ok((kes_signature, operational_certificate.clone())), // TODO: remove this line once the hack of KES signature is removed in DMQ message builder
-                        Ok((kes_signature, operational_certificate.clone())), // TODO: remove this line once the hack of KES signature is removed in DMQ message builder
                     ]);
 
                     Arc::new(kes_signer)
@@ -94,13 +92,49 @@ async fn dmq_publisher_client_server() {
                 .publish_message(DmqMessageTestPayload::new(b"msg_2"))
                 .await
                 .unwrap();
+        }
+    });
+    client.await.unwrap();
+
+    // Sleep to avoid refused connection from the server
+    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+
+    // Start a second client and publish messages
+    let client = tokio::spawn({
+        let socket_path = socket_path.clone();
+        async move {
+            let dmq_builder = DmqMessageBuilder::new(
+                {
+                    let (kes_signature, operational_certificate) = KesSignerFake::dummy_signature();
+                    let kes_signer = KesSignerFake::new(vec![
+                        Ok((kes_signature, operational_certificate.clone())),
+                        Ok((kes_signature, operational_certificate.clone())),
+                    ]);
+
+                    Arc::new(kes_signer)
+                },
+                Arc::new(FakeChainObserver::default()),
+            )
+            .set_ttl(100);
+            let publisher_client = DmqPublisherClientPallas::<DmqMessageTestPayload>::new(
+                socket_path,
+                cardano_network,
+                dmq_builder,
+                slog_scope::logger(),
+            );
+
+            publisher_client
+                .publish_message(DmqMessageTestPayload::new(b"msg_3"))
+                .await
+                .unwrap();
 
             stop_tx
                 .send(())
                 .expect("Failed to send stop signal to DMQ publisher server");
         }
     });
 
+    // Record messages received by the server
     let recorder = tokio::spawn(async move {
         let messages: Vec<DmqMessage> = {
             let mut messages = vec![];
@@ -115,9 +149,14 @@ async fn dmq_publisher_client_server() {
         messages
     });
 
+    // Check that all messages have been correctly received
     let (_, _, messages) = tokio::try_join!(server, client, recorder).unwrap();
     assert_eq!(
-        vec![create_fake_msg(b"msg_1").await, create_fake_msg(b"msg_2").await],
+        vec![
+            create_fake_msg(b"msg_1").await,
+            create_fake_msg(b"msg_2").await,
+            create_fake_msg(b"msg_3").await,
+        ],
         messages
     );
 }