fix(dmq): consumer can not serve messages multiple times

Author: jpraynaud

internal/mithril-dmq/src/consumer/client/pallas.rs

@@ -1,8 +1,8 @@
 use std::{fmt::Debug, marker::PhantomData, path::PathBuf};
 
 use anyhow::{Context, anyhow};
-use pallas_network::facades::DmqClient;
-use slog::{Logger, debug, error};
+use pallas_network::{facades::DmqClient, miniprotocols::localmsgnotification::State};
+use slog::{Logger, debug};
 use tokio::sync::{Mutex, MutexGuard};
 
 use mithril_common::{
@@ -94,21 +94,20 @@ impl<M: TryFromBytes + Debug> DmqConsumerClientPallas<M> {
         debug!(self.logger, "Waiting for messages from DMQ...");
         let mut client_guard = self.get_client().await?;
         let client = client_guard.as_mut().ok_or(anyhow!("DMQ client does not exist"))?;
-        client
-            .msg_notification()
-            .send_request_messages_blocking()
-            .await
-            .with_context(|| "Failed to request notifications from DMQ server: {}")?;
+        if *client.msg_notification().state() == State::Idle {
+            client
+                .msg_notification()
+                .send_request_messages_blocking()
+                .await
+                .with_context(|| "Failed to request notifications from DMQ server: {}")?;
+        }
 
         let reply = client
             .msg_notification()
             .recv_next_reply()
             .await
             .with_context(|| "Failed to receive notifications from DMQ server")?;
         debug!(self.logger, "Received single signatures from DMQ"; "messages" => ?reply);
-        if let Err(e) = client.msg_notification().send_done().await {
-            error!(self.logger, "Failed to send Done"; "error" => ?e);
-        }
 
         reply
             .0
@@ -227,10 +226,6 @@ mod tests {
                     // server replies with messages if any
                     server_msg.send_reply_messages_blocking(reply_messages).await.unwrap();
                     assert_eq!(*server_msg.state(), localmsgnotification::State::Idle);
-
-                    // server receives done from client
-                    server_msg.recv_done().await.unwrap();
-                    assert_eq!(*server_msg.state(), localmsgnotification::State::Done);
                 } else {
                     // server waits if no message available
                     future::pending().await

internal/mithril-dmq/src/consumer/server/pallas.rs

@@ -151,9 +151,7 @@ impl DmqConsumerServerPallas {
                     }
                 }
             },
-            None => {
-                return Err(anyhow!("DMQ message receiver is not registered"));
-            }
+            None => Err(anyhow!("DMQ message receiver is not registered")),
         }
     }
 
@@ -229,8 +227,7 @@ impl DmqConsumerServer for DmqConsumerServerPallas {
                     .await?;
                 debug!(
                     self.logger,
-                    "Blocking notification replied to the DMQ notification client: {:?}",
-                    reply_messages
+                    "Messages replied to the DMQ notification client: {:?}", reply_messages
                 );
             }
             Request::NonBlocking => {
@@ -244,9 +241,12 @@ impl DmqConsumerServer for DmqConsumerServerPallas {
                 let has_more = !self.messages_buffer.is_empty().await;
                 server
                     .msg_notification()
-                    .send_reply_messages_non_blocking(reply_messages, has_more)
+                    .send_reply_messages_non_blocking(reply_messages.clone(), has_more)
                     .await?;
-                server.msg_notification().recv_done().await?;
+                debug!(
+                    self.logger,
+                    "Messages replied to the DMQ notification client: {:?}", reply_messages
+                );
             }
         };
 

internal/mithril-dmq/tests/consumer_client_server.rs

@@ -47,7 +47,7 @@ async fn dmq_consumer_client_server() {
         async move {
             let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
                 socket_path.to_path_buf(),
-                cardano_network.to_owned(),
+                cardano_network,
                 stop_rx,
                 slog_scope::logger(),
             ));
@@ -61,13 +61,14 @@ async fn dmq_consumer_client_server() {
         async move {
             let consumer_client = DmqConsumerClientPallas::<DmqMessageTestPayload>::new(
                 socket_path,
-                cardano_network.clone(),
+                cardano_network,
                 slog_scope::logger(),
             );
             let mut messages = vec![];
-            signature_dmq_tx.send(create_fake_msg(b"msg_1").await.into()).unwrap();
+            signature_dmq_tx.send(create_fake_msg(b"msg_1").await).unwrap();
+            signature_dmq_tx.send(create_fake_msg(b"msg_2").await).unwrap();
             messages.extend_from_slice(&consumer_client.consume_messages().await.unwrap());
-            signature_dmq_tx.send(create_fake_msg(b"msg_2").await.into()).unwrap();
+            signature_dmq_tx.send(create_fake_msg(b"msg_3").await).unwrap();
             messages.extend_from_slice(&consumer_client.consume_messages().await.unwrap());
             stop_tx.send(()).unwrap();
 
@@ -79,7 +80,8 @@ async fn dmq_consumer_client_server() {
     assert_eq!(
         vec![
             DmqMessageTestPayload::new(b"msg_1"),
-            DmqMessageTestPayload::new(b"msg_2")
+            DmqMessageTestPayload::new(b"msg_2"),
+            DmqMessageTestPayload::new(b"msg_3")
         ],
         messages
     );