Merge pull request #29 from OpenMatchmaking/feature-ack-message

Acking published messages
OpenMatchmaking · Jan 24, 2019 · 90bc5da · 90bc5da
2 parents 13b725a + 9628335
commit 90bc5da
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 8 deletions.
diff --git a/pathfinder/src/engine/futures.rs b/pathfinder/src/engine/futures.rs
@@ -14,7 +14,7 @@ use lapin_futures_rustls::lapin::channel::{
     QueueDeclareOptions, QueueDeleteOptions, QueueUnbindOptions,
 };
 use lapin_futures_rustls::lapin::types::{AMQPValue, FieldTable};
-use log::error;
+use log::{error, info, warn};
 
 use crate::error::PathfinderError;
 use crate::rabbitmq::{RabbitMQContext};
@@ -99,7 +99,14 @@ pub fn rpc_request_future(
                     publish_message_options,
                     basic_properties
                 )
-                .map(move |_confirmation| (publish_channel, consume_channel, queue, options))
+                .map(move |confirmation| {
+                    match confirmation {
+                        Some(_) => info!("Publish message got confirmation."),
+                        None => warn!("Request wasn't delivered."),
+                    };
+
+                    (publish_channel, consume_channel, queue, options)
+                })
         })
         // 4. Consume a response message from the queue, that was declared on the 1st step
         .and_then(move |(publish_channel, consume_channel, queue, options)| {
@@ -119,7 +126,7 @@ pub fn rpc_request_future(
                 })
         })
         // 5. Prepare a response for a client, serialize and sent via WebSocket transmitter
-        .and_then(move |(_publish_channel, consume_channel, queue, message, options)| {
+        .and_then(move |(publish_channel, consume_channel, queue, message, options)| {
             let raw_data = from_utf8(&message.data).unwrap();
             let json = Arc::new(Box::new(json_parse(raw_data).unwrap()));
             let serializer = Serializer::new();
@@ -129,7 +136,7 @@ pub fn rpc_request_future(
 
             consume_channel
                 .basic_ack(message.delivery_tag, false)
-                .map(move |_confirmation| (_publish_channel, consume_channel, queue, options))
+                .map(move |_confirmation| (publish_channel, consume_channel, queue, options))
         })
         // 6. Unbind the response queue from the exchange point
         .and_then(move |(publish_channel, consume_channel, _queue, options)| {

diff --git a/pathfinder/src/engine/middleware/jwt.rs b/pathfinder/src/engine/middleware/jwt.rs
@@ -14,7 +14,7 @@ use lapin_futures_rustls::lapin::channel::{
     QueueDeclareOptions, QueueDeleteOptions, QueueUnbindOptions,
 };
 use lapin_futures_rustls::lapin::types::{AMQPValue, FieldTable};
-use log::error;
+use log::{error, info, warn};
 use uuid::Uuid;
 
 use crate::error::PathfinderError;
@@ -121,7 +121,14 @@ impl JwtTokenMiddleware {
                     publish_message_options,
                     basic_properties
                 )
-                .map(move |_confirmation| (publish_channel, consume_channel, queue, options))
+                .map(move |confirmation| {
+                    match confirmation {
+                        Some(_) => info!("Publish for verifying JWT got confirmation."),
+                        None => warn!("Request for verifying JWT wasn't delivered."),
+                    };
+
+                    (publish_channel, consume_channel, queue, options)
+                })
         })
         // 4. Consume a response message from the queue, that was declared on the 2nd step
         .and_then(move |(publish_channel, consume_channel, queue, options)| {
@@ -284,7 +291,14 @@ impl JwtTokenMiddleware {
                     publish_message_options,
                     basic_properties
                 )
-                .map(move |_confirmation| (publish_channel, consume_channel, queue, options))
+                .map(move |confirmation| {
+                    match confirmation {
+                        Some(_) => info!("Publish for getting headers got confirmation."),
+                        None => warn!("Request for getting headers wasn't delivered."),
+                    };
+
+                    (publish_channel, consume_channel, queue, options)
+                })
         })
         // 4. Consume a response message from the queue, that was declared on the 2nd step
         .and_then(move |(publish_channel, consume_channel, queue, options)| {

diff --git a/pathfinder/src/rabbitmq/utils.rs b/pathfinder/src/rabbitmq/utils.rs
@@ -33,7 +33,7 @@ pub fn get_uri(cli: &CliOptions) -> AMQPUri {
         false => "amqp",
     };
     format!(
-        "{}://{}:{}@{}:{}/{}",
+        "{}://{}:{}@{}:{}/{}?heartbeat=10",
         schema.to_string(),
         cli.rabbitmq_username.clone(),
         cli.rabbitmq_password.clone(),