Update rpc

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

Author: Gsantomaggio

rust/src/bin/receive_logs_topic.rs

@@ -1,3 +1,4 @@
+use futures::StreamExt;
 use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties, ExchangeKind};
 
 #[tokio::main]
@@ -35,7 +36,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .await?;
 
-    futures::future::join_all(binding_keys.into_iter().map(|binding_key| {
+    futures::future::join_all(binding_keys.iter().map(|binding_key| {
         channel.queue_bind(
             queue.name().as_str(),
             "topic_logs",
@@ -46,7 +47,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",
@@ -60,14 +61,17 @@ 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(())
 }

rust/src/bin/rpc_client.rs

@@ -5,7 +5,6 @@ use lapin::{
 };
 use std::convert::TryInto;
 use std::fmt::Display;
-use tokio::stream::StreamExt;
 use uuid::Uuid;
 
 #[derive(Debug)]
@@ -87,15 +86,16 @@ impl FibonacciRpcClient {
             .await?;
 
         while let Some(delivery) = self.consumer.next().await {
-            let (_, reply) = delivery?;
-            if reply.properties.correlation_id().as_ref() == Some(&self.correlation_id) {
-                return Ok(u64::from_le_bytes(
-                    reply
-                        .data
-                        .as_slice()
-                        .try_into()
-                        .map_err(|_| Error::CannotDecodeReply)?,
-                ));
+            if let Ok(delivery) = delivery {
+                if delivery.properties.correlation_id().as_ref() == Some(&self.correlation_id) {
+                    return Ok(u64::from_le_bytes(
+                        delivery
+                            .data
+                            .as_slice()
+                            .try_into()
+                            .map_err(|_| Error::CannotDecodeReply)?,
+                    ));
+                }
             }
         }
 

rust/src/bin/rpc_server.rs

@@ -1,6 +1,7 @@
-use lapin::{options::*, types::FieldTable, BasicProperties, Connection, ConnectionProperties};
 use std::convert::TryInto;
 use std::fmt::Display;
+use futures::StreamExt;
+use lapin::{BasicProperties, Connection, ConnectionProperties, options::*, types::FieldTable};
 
 #[derive(Debug)]
 enum Error {
@@ -45,7 +46,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     channel.basic_qos(1, BasicQosOptions::default()).await?;
 
-    let consumer = channel
+    let mut consumer = channel
         .basic_consume(
             "rpc_queue",
             "rpc_server",
@@ -56,44 +57,48 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     println!(" [x] Awaiting RPC requests");
 
-    for delivery in consumer {
-        let (channel, delivery) = delivery?;
-        let n = u64::from_le_bytes(
-            delivery
-                .data
-                .as_slice()
-                .try_into()
-                .map_err(|_| Error::CannotDecodeArg)?,
-        );
-        println!(" [.] fib({})", n);
-        let response = fib(n);
-
-        let routing_key = delivery
-            .properties
-            .reply_to()
-            .as_ref()
-            .ok_or(Error::MissingReplyTo)?
-            .as_str();
-
-        let correlation_id = delivery
-            .properties
-            .correlation_id()
-            .clone()
-            .ok_or(Error::MissingCorrelationId)?;
-
-        channel
-            .basic_publish(
-                "",
-                routing_key,
-                BasicPublishOptions::default(),
-                response.to_le_bytes().to_vec(),
-                BasicProperties::default().with_correlation_id(correlation_id),
-            )
-            .await?;
-
-        channel
-            .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
-            .await?;
+
+    while let Some(delivery) = consumer.next().await {
+        if let Ok(delivery) = delivery {
+            println!(" [x] Received {:?}", std::str::from_utf8(&delivery.data)?);
+            let n = u64::from_le_bytes(
+                delivery
+                    .data
+                    .as_slice()
+                    .try_into()
+                    .map_err(|_| Error::CannotDecodeArg)?,
+            );
+            println!(" [.] fib({})", n);
+            let response = fib(n);
+            let payload = response.to_be_bytes();
+
+            let routing_key = delivery
+                .properties
+                .reply_to()
+                .as_ref()
+                .ok_or(Error::MissingReplyTo)?
+                .as_str();
+
+            let correlation_id = delivery
+                .properties
+                .correlation_id()
+                .clone()
+                .ok_or(Error::MissingCorrelationId)?;
+
+            channel
+                .basic_publish(
+                    "",
+                    routing_key,
+                    BasicPublishOptions::default(),
+                    &payload,
+                    BasicProperties::default().with_correlation_id(correlation_id),
+                )
+                .await?;
+
+            channel
+                .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
+                .await?;
+        }
     }
 
     Ok(())