cargo clippy & fmt

rmqtt · Jul 30, 2023 · 66bf377 · 66bf377
1 parent 8536257
commit 66bf377
Show file tree

Hide file tree

Showing 5 changed files with 54 additions and 34 deletions.
diff --git a/src/lib.rs b/src/lib.rs
@@ -1,8 +1,8 @@
 use std::time::Duration;
 
 pub use crate::error::{Error, Result};
-pub use crate::raft::{Mailbox, Raft, Store};
 pub use crate::message::Status;
+pub use crate::raft::{Mailbox, Raft, Store};
 pub use tikv_raft::ReadOnlyOption;
 
 mod error;

diff --git a/src/raft.rs b/src/raft.rs
@@ -17,8 +17,8 @@ use crate::error::{Error, Result};
 use crate::message::{Message, RaftResponse, Status};
 use crate::raft_node::{Peer, RaftNode};
 use crate::raft_server::RaftServer;
-use crate::raft_service::{ConfChange as RiteraftConfChange, Empty, ResultCode};
 use crate::raft_service::connect;
+use crate::raft_service::{ConfChange as RiteraftConfChange, Empty, ResultCode};
 use crate::Config;
 
 type DashMap<K, V> = dashmap::DashMap<K, V, ahash::RandomState>;
@@ -254,8 +254,16 @@ pub struct Raft<S: Store + 'static> {
 
 impl<S: Store + Send + Sync + 'static> Raft<S> {
     /// creates a new node with the given address and store.
-    pub fn new<A: ToSocketAddrs>(addr: A, store: S, logger: slog::Logger, cfg: Config) -> Result<Self> {
-        let addr = addr.to_socket_addrs()?.next().ok_or(Error::from("None"))?;
+    pub fn new<A: ToSocketAddrs>(
+        addr: A,
+        store: S,
+        logger: slog::Logger,
+        cfg: Config,
+    ) -> Result<Self> {
+        let addr = addr
+            .to_socket_addrs()?
+            .next()
+            .ok_or_else(|| Error::from("None"))?;
         let (tx, rx) = mpsc::channel(100_000);
         let cfg = Arc::new(cfg);
         Ok(Self {
@@ -432,7 +440,9 @@ impl<S: Store + Send + Sync + 'static> Raft<S> {
         let mut change = ConfChange::default();
         change.set_node_id(node_id);
         change.set_change_type(ConfChangeType::AddNode);
-        change.set_context(prost::bytes::Bytes::from(serialize(&self.addr.to_string())?));
+        change.set_context(prost::bytes::Bytes::from(serialize(
+            &self.addr.to_string(),
+        )?));
         // change.set_context(serialize(&self.addr)?);
 
         let change = RiteraftConfChange {
@@ -447,7 +457,10 @@ impl<S: Store + Send + Sync + 'static> Raft<S> {
             peer_addrs,
         } = deserialize(&raft_response.inner)?
         {
-            info!("change_config response.assigned_id: {:?}, peer_addrs: {:?}", assigned_id, peer_addrs);
+            info!(
+                "change_config response.assigned_id: {:?}, peer_addrs: {:?}",
+                assigned_id, peer_addrs
+            );
             for (id, addr) in peer_addrs {
                 if id != assigned_id {
                     node.add_peer(&addr, id);

diff --git a/src/raft_server.rs b/src/raft_server.rs
@@ -33,10 +33,7 @@ impl RaftServer {
         let addr = self.addr;
         info!("listening gRPC requests on: {}", addr);
         let svc = RaftServiceServer::new(self);
-        Server::builder()
-            .add_service(svc)
-            .serve(addr)
-            .await?;
+        Server::builder().add_service(svc).serve(addr).await?;
         info!("server has quit");
         Ok(())
     }
@@ -143,26 +140,23 @@ impl RaftService for RaftServer {
         let message = Message::Propose { proposal, chan: tx };
 
         let reply = match sender.try_send(message) {
-            Ok(()) => {
-                let reply = match timeout(self.timeout, rx).await {
-                    Ok(Ok(raft_response)) => match serialize(&raft_response) {
-                        Ok(resp) => Ok(Response::new(raft_service::RaftResponse { inner: resp })),
-                        Err(e) => {
-                            warn!("serialize error, {}", e);
-                            Err(Status::unavailable("serialize error"))
-                        }
-                    },
-                    Ok(Err(e)) => {
-                        warn!("recv error for reply, {}", e);
-                        Err(Status::unavailable("recv error for reply"))
-                    }
+            Ok(()) => match timeout(self.timeout, rx).await {
+                Ok(Ok(raft_response)) => match serialize(&raft_response) {
+                    Ok(resp) => Ok(Response::new(raft_service::RaftResponse { inner: resp })),
                     Err(e) => {
-                        warn!("timeout waiting for reply, {}", e);
-                        Err(Status::unavailable("timeout waiting for reply"))
+                        warn!("serialize error, {}", e);
+                        Err(Status::unavailable("serialize error"))
                     }
-                };
-                reply
-            }
+                },
+                Ok(Err(e)) => {
+                    warn!("recv error for reply, {}", e);
+                    Err(Status::unavailable("recv error for reply"))
+                }
+                Err(e) => {
+                    warn!("timeout waiting for reply, {}", e);
+                    Err(Status::unavailable("timeout waiting for reply"))
+                }
+            },
             Err(e) => {
                 warn!("error for try send message, {}", e);
                 Err(Status::unavailable("error for try send message"))

diff --git a/src/raft_service.rs b/src/raft_service.rs
@@ -11,18 +11,31 @@ tonic::include_proto!("raftservice");
 pub(crate) type RaftServiceClientType = RaftServiceClient<Channel>;
 
 #[inline]
-pub(crate) fn endpoint(saddr: &str, concurrency_limit: usize, timeout: Duration) -> Result<Endpoint> {
+pub(crate) fn endpoint(
+    saddr: &str,
+    concurrency_limit: usize,
+    timeout: Duration,
+) -> Result<Endpoint> {
     let endpoint = Channel::from_shared(format!("http://{}", saddr))
         .map(|endpoint| {
             endpoint
                 .concurrency_limit(concurrency_limit)
                 .connect_timeout(timeout)
                 .timeout(timeout)
-        }).map_err(anyhow::Error::new)?;
+        })
+        .map_err(anyhow::Error::new)?;
     Ok(endpoint)
 }
 
 #[inline]
-pub(crate) async fn connect(saddr: &str, concurrency_limit: usize, timeout: Duration) -> Result<RaftServiceClientType> {
-    Ok(RaftServiceClientType::new(endpoint(saddr, concurrency_limit, timeout)?.connect().await?))
-}
+pub(crate) async fn connect(
+    saddr: &str,
+    concurrency_limit: usize,
+    timeout: Duration,
+) -> Result<RaftServiceClientType> {
+    Ok(RaftServiceClientType::new(
+        endpoint(saddr, concurrency_limit, timeout)?
+            .connect()
+            .await?,
+    ))
+}
diff --git a/src/storage.rs b/src/storage.rs
@@ -95,7 +95,7 @@ impl Storage for MemStorage {
         low: u64,
         high: u64,
         max_size: impl Into<Option<u64>>,
-        context: GetEntriesContext
+        context: GetEntriesContext,
     ) -> tikv_raft::Result<Vec<Entry>> {
         let entries = self.core.entries(low, high, max_size, context)?;
         Ok(entries)