Format API rpc/message.rs

Signed-off-by: Lee Smet <[email protected]>

Author: LeeSmet

mycelium-api/src/rpc/message.rs

@@ -4,13 +4,13 @@ use jsonrpc_core::{Error, ErrorCode, Result as RpcResult};
 use std::time::Duration;
 use tracing::debug;
 
-use mycelium::metrics::Metrics;
 use mycelium::message::{MessageId, MessageInfo};
+use mycelium::metrics::Metrics;
 
-use crate::HttpServerState;
 use crate::message::{MessageReceiveInfo, MessageSendInfo, PushMessageResponse};
 use crate::rpc::models::error_codes;
 use crate::rpc::traits::MessageApi;
+use crate::HttpServerState;
 
 /// Implementation of Message-related JSON-RPC methods
 pub struct MessageRpc<M>
@@ -28,10 +28,11 @@ where
     pub fn new(state: HttpServerState<M>) -> Self {
         Self { state }
     }
-    
+
     /// Convert a base64 string to bytes
     fn decode_base64(&self, s: &str) -> Result<Vec<u8>, Error> {
-        base64::engine::general_purpose::STANDARD.decode(s.as_bytes())
+        base64::engine::general_purpose::STANDARD
+            .decode(s.as_bytes())
             .map_err(|e| Error {
                 code: ErrorCode::InvalidParams,
                 message: format!("Invalid base64 encoding: {}", e),
@@ -44,19 +45,24 @@ impl<M> MessageApi for MessageRpc<M>
 where
     M: Metrics + Clone + Send + Sync + 'static,
 {
-    fn pop_message(&self, peek: Option<bool>, timeout: Option<u64>, topic: Option<String>) -> RpcResult<MessageReceiveInfo> {
+    fn pop_message(
+        &self,
+        peek: Option<bool>,
+        timeout: Option<u64>,
+        topic: Option<String>,
+    ) -> RpcResult<MessageReceiveInfo> {
         debug!(
             "Attempt to get message via RPC, peek {}, timeout {} seconds",
             peek.unwrap_or(false),
             timeout.unwrap_or(0)
         );
-        
+
         let topic_bytes = if let Some(topic_str) = topic {
             Some(self.decode_base64(&topic_str)?)
         } else {
             None
         };
-        
+
         // A timeout of 0 seconds essentially means get a message if there is one, and return
         // immediately if there isn't.
         let result = tokio::task::block_in_place(|| {
@@ -72,7 +78,7 @@ where
                 .await
             })
         });
-        
+
         match result {
             Ok(Ok(m)) => Ok(MessageReceiveInfo {
                 id: m.id,
@@ -94,22 +100,26 @@ where
             }),
         }
     }
-    
-    fn push_message(&self, message: MessageSendInfo, reply_timeout: Option<u64>) -> RpcResult<PushMessageResponse> {
+
+    fn push_message(
+        &self,
+        message: MessageSendInfo,
+        reply_timeout: Option<u64>,
+    ) -> RpcResult<PushMessageResponse> {
         let dst = match message.dst {
             crate::message::MessageDestination::Ip(ip) => ip,
             crate::message::MessageDestination::Pk(pk) => pk.address().into(),
         };
-        
+
         debug!(
             message.dst=%dst,
             message.len=message.payload.len(),
             "Pushing new message via RPC",
         );
-        
+
         // Default message try duration
         const DEFAULT_MESSAGE_TRY_DURATION: Duration = Duration::from_secs(60 * 5);
-        
+
         let result = tokio::task::block_in_place(|| {
             tokio::runtime::Handle::current().block_on(async {
                 self.state.node.lock().await.push_message(
@@ -121,7 +131,7 @@ where
                 )
             })
         });
-        
+
         let (id, sub) = match result {
             Ok((id, sub)) => (id, sub),
             Err(_) => {
@@ -132,14 +142,16 @@ where
                 });
             }
         };
-        
+
         if reply_timeout.is_none() {
             // If we don't wait for the reply just return here.
-            return Ok(PushMessageResponse::Id(crate::message::MessageIdReply { id }));
+            return Ok(PushMessageResponse::Id(crate::message::MessageIdReply {
+                id,
+            }));
         }
-        
+
         let mut sub = sub.unwrap();
-        
+
         // Wait for reply with timeout
         let reply_result = tokio::task::block_in_place(|| {
             tokio::runtime::Handle::current().block_on(async {
@@ -183,13 +195,13 @@ where
                 }
             })
         });
-        
+
         match reply_result {
             Ok(response) => Ok(response),
             Err(e) => Err(e),
         }
     }
-    
+
     fn push_message_reply(&self, id: String, message: MessageSendInfo) -> RpcResult<bool> {
         let message_id = match MessageId::from_hex(&id) {
             Ok(id) => id,
@@ -201,22 +213,22 @@ where
                 });
             }
         };
-        
+
         let dst = match message.dst {
             crate::message::MessageDestination::Ip(ip) => ip,
             crate::message::MessageDestination::Pk(pk) => pk.address().into(),
         };
-        
+
         debug!(
             message.id=id,
             message.dst=%dst,
             message.len=message.payload.len(),
             "Pushing new reply to message via RPC",
         );
-        
+
         // Default message try duration
         const DEFAULT_MESSAGE_TRY_DURATION: Duration = Duration::from_secs(60 * 5);
-        
+
         tokio::task::block_in_place(|| {
             tokio::runtime::Handle::current().block_on(async {
                 self.state.node.lock().await.reply_message(
@@ -227,10 +239,10 @@ where
                 );
             })
         });
-        
+
         Ok(true)
     }
-    
+
     fn get_message_info(&self, id: String) -> RpcResult<MessageInfo> {
         let message_id = match MessageId::from_hex(&id) {
             Ok(id) => id,
@@ -242,15 +254,14 @@ where
                 });
             }
         };
-        
+
         debug!(message.id=%id, "Fetching message status via RPC");
-        
+
         let result = tokio::task::block_in_place(|| {
-            tokio::runtime::Handle::current().block_on(async {
-                self.state.node.lock().await.message_status(message_id)
-            })
+            tokio::runtime::Handle::current()
+                .block_on(async { self.state.node.lock().await.message_status(message_id) })
         });
-        
+
         match result {
             Some(info) => Ok(info),
             None => Err(Error {
@@ -260,4 +271,5 @@ where
             }),
         }
     }
-}
+}
+