Update receivers

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

rust/src/bin/emit_log_direct.rs

@@ -7,8 +7,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args: Vec<_> = std::env::args().skip(1).collect();
     let severity = args.first().map(String::as_str).unwrap_or("info");
     let message = match args.len() {
-        x if x < 2 => b"Hello, world!".to_vec(),
-        _ => args[1..].join(" ").into_bytes(),
+        x if x < 2 => "Hello, world!".to_string(),
+        _ => args[1..].join(" ").to_string(),
     };
 
     let addr = "amqp://127.0.0.1:5672";
@@ -29,15 +29,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             "direct_logs",
             severity,
             BasicPublishOptions::default(),
-            &*message.clone(),
+            message.as_bytes(),
             BasicProperties::default(),
         )
         .await?;
 
     println!(
         " [x] Sent {}:{:?}",
         severity,
-        std::str::from_utf8(&message)?
+        std::str::from_utf8(message.as_bytes())?
     );
 
     conn.close(0, "").await?;

rust/src/bin/receive_logs.rs

@@ -1,3 +1,4 @@
+use futures::StreamExt;
 use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties, ExchangeKind};
 
 #[tokio::main]
@@ -36,7 +37,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .await?;
 
-    let consumer = channel
+    let mut consumer = channel
         .basic_consume(
             queue.name().as_str(),
             "consumer",
@@ -50,10 +51,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     println!(" [*] Waiting for logs. To exit press CTRL+C");
 
-    for delivery in consumer {
-        let (_, delivery) = delivery?;
-        println!(" [x] {:?}", std::str::from_utf8(&delivery.data)?);
+
+    while let Some(delivery) = consumer.next().await {
+        if let Ok(delivery) = delivery {
+            println!(" [x] Received {:?}", std::str::from_utf8(&delivery.data)?);
+        }
     }
 
+
     Ok(())
 }

rust/src/bin/receive_logs_direct.rs

@@ -1,4 +1,7 @@
-use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties, ExchangeKind};
+use std::borrow::Borrow;
+
+use futures::StreamExt;
+use lapin::{Connection, ConnectionProperties, ExchangeKind, options::*, types::FieldTable};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -35,6 +38,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .await?;
 
+
     futures::future::join_all(severities.into_iter().map(|severity| {
         channel.queue_bind(
             queue.name().as_str(),
@@ -43,10 +47,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             QueueBindOptions::default(),
             FieldTable::default(),
         )
-    }))
-    .await;
+    })).await;
 
-    let consumer = channel
+    let mut consumer = channel
         .basic_consume(
             queue.name().as_str(),
             "consumer",
@@ -60,14 +63,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     println!(" [*] Waiting for logs. To exit press CTRL+C");
 
-    for delivery in consumer {
-        let (_, delivery) = delivery?;
-        println!(
-            " [x] {}:{:?}",
-            delivery.routing_key,
-            std::str::from_utf8(&delivery.data)?
-        );
+    while let Some(delivery) = consumer.next().await {
+        if let Ok(delivery) = delivery {
+            println!(
+                " [x] {}:{:?}",
+                delivery.routing_key,
+                std::str::from_utf8(&delivery.data)?
+            );
+        }
     }
-
     Ok(())
 }