refactor: clean up code formatting and improve readability in WebSocket client and codec

Author: lxsaah

aimdb-websocket-connector/src/client/builder.rs

@@ -171,11 +171,8 @@ where
             // Mirrors `AimxClientConnector`: `run_client` owns demux/reconnect/
             // keepalive over the WS `Dialer` + per-connection `WsCodec`;
             // `pump_client` wires `link_to`/`link_from` routes to the handle.
-            let (handle, engine_fut) = run_client(
-                WsDialer::new(self.url.clone()),
-                WsCodec::new(),
-                config,
-            );
+            let (handle, engine_fut) =
+                run_client(WsDialer::new(self.url.clone()), WsCodec::new(), config);
             let mut futures = pump_client(db, "ws-client", &handle);
             futures.push(engine_fut);
             Ok(futures)

aimdb-websocket-connector/src/codec.rs

@@ -126,7 +126,8 @@ impl aimdb_core::EnvelopeCodec for WsCodec {
     // ---- server direction: read a ClientMessage, write a ServerMessage ------
 
     fn decode(&self, frame: &[u8]) -> Result<Inbound, CodecError> {
-        let msg: ClientMessage = serde_json::from_slice(frame).map_err(|_| CodecError::Malformed)?;
+        let msg: ClientMessage =
+            serde_json::from_slice(frame).map_err(|_| CodecError::Malformed)?;
         let mut st = self.state.lock().unwrap();
         match msg {
             // The transport splits multi-topic frames, so exactly one topic here.
@@ -190,7 +191,12 @@ impl aimdb_core::EnvelopeCodec for WsCodec {
                     .unwrap()
                     .topic_of(sub)
                     .ok_or(CodecError::Malformed)?;
-                write_server(out, &ServerMessage::Subscribed { topics: vec![topic] })
+                write_server(
+                    out,
+                    &ServerMessage::Subscribed {
+                        topics: vec![topic],
+                    },
+                )
             }
             Outbound::Pong => write_server(out, &ServerMessage::Pong),
             // `Reply::Ok` payloads are already a complete `ServerMessage` JSON
@@ -388,10 +394,20 @@ mod tests {
         let id = sub(&codec, "a/b");
         let mut out = Vec::new();
         codec
-            .encode(Outbound::Subscribed { sub: &id.to_string() }, &mut out)
+            .encode(
+                Outbound::Subscribed {
+                    sub: &id.to_string(),
+                },
+                &mut out,
+            )
             .unwrap();
         let v: ServerMessage = serde_json::from_slice(&out).unwrap();
-        assert_eq!(v, ServerMessage::Subscribed { topics: vec!["a/b".into()] });
+        assert_eq!(
+            v,
+            ServerMessage::Subscribed {
+                topics: vec!["a/b".into()]
+            }
+        );
     }
 
     #[test]

aimdb-websocket-connector/src/dispatch.rs

@@ -119,11 +119,7 @@ impl Session for WsSession {
                     .handle_query(&pattern, from, to, limit)
                     .await
                 {
-                    Ok((records, total)) => ServerMessage::QueryResult {
-                        id,
-                        records,
-                        total,
-                    },
+                    Ok((records, total)) => ServerMessage::QueryResult { id, records, total },
                     Err(message) => ServerMessage::Error {
                         code: ErrorCode::ServerError,
                         topic: None,
@@ -192,8 +188,8 @@ mod tests {
     use std::sync::Mutex;
     use std::time::Duration;
 
-    use aimdb_core::session::{run_session, SessionConfig};
     use aimdb_core::router::RouterBuilder;
+    use aimdb_core::session::{run_session, SessionConfig};
     use aimdb_core::{Connection, SessionLimits, TransportResult};
     use tokio::sync::mpsc;
 
@@ -266,8 +262,10 @@ mod tests {
     // snapshot, then a bus broadcast fans out as a Data frame.
     #[tokio::test]
     async fn subscribe_ack_snapshot_and_fanout() {
-        let dispatch =
-            dispatch_with(Arc::new(OneSnap("sensors/temp".into(), b"\"last\"".to_vec())));
+        let dispatch = dispatch_with(Arc::new(OneSnap(
+            "sensors/temp".into(),
+            b"\"last\"".to_vec(),
+        )));
         let mgr = dispatch.client_mgr.clone();
 
         let (tx, rx) = mpsc::unbounded_channel::<Vec<u8>>();
@@ -303,8 +301,12 @@ mod tests {
 
         // Ack + snapshot should have been emitted, in order.
         let msgs = parse(&out);
-        assert!(matches!(&msgs[0], ServerMessage::Subscribed { topics } if topics == &vec!["sensors/temp".to_string()]));
-        assert!(matches!(&msgs[1], ServerMessage::Snapshot { topic, .. } if topic == "sensors/temp"));
+        assert!(
+            matches!(&msgs[0], ServerMessage::Subscribed { topics } if topics == &vec!["sensors/temp".to_string()])
+        );
+        assert!(
+            matches!(&msgs[1], ServerMessage::Snapshot { topic, .. } if topic == "sensors/temp")
+        );
 
         // A bus broadcast now fans out to this subscription as a Data frame.
         mgr.broadcast("sensors/temp", b"\"22.5\"").await;
@@ -313,7 +315,9 @@ mod tests {
         let data = msgs
             .iter()
             .find_map(|m| match m {
-                ServerMessage::Data { topic, payload, .. } => Some((topic.clone(), payload.clone())),
+                ServerMessage::Data { topic, payload, .. } => {
+                    Some((topic.clone(), payload.clone()))
+                }
                 _ => None,
             })
             .expect("a Data frame");

aimdb-websocket-connector/src/transport.rs

@@ -142,7 +142,9 @@ impl WsDialer {
 
 #[cfg(feature = "client")]
 impl aimdb_core::Dialer for WsDialer {
-    fn connect(&self) -> aimdb_core::BoxFut<'_, aimdb_core::TransportResult<Box<dyn aimdb_core::Connection>>> {
+    fn connect(
+        &self,
+    ) -> aimdb_core::BoxFut<'_, aimdb_core::TransportResult<Box<dyn aimdb_core::Connection>>> {
         Box::pin(async move {
             let (ws, _resp) = tokio_tungstenite::connect_async(&self.url)
                 .await