refactor: builder pattern

Author: Devdutt Shenoi

src/audit.rs

@@ -173,56 +173,64 @@ impl Default for AuditLogBuilder {
 }
 
 impl AuditLogBuilder {
-    pub fn set_stream_name(&mut self, stream: impl Into<String>) {
+    pub fn set_stream_name(mut self, stream: impl Into<String>) -> Self {
         if AUDIT_LOGGER.is_none() {
-            return;
+            return self;
         }
         self.stream = stream.into();
+
+        self
     }
 
     pub fn set_actor(
-        &mut self,
+        mut self,
         host: impl Into<String>,
         username: impl Into<String>,
         user_agent: impl Into<String>,
         auth_method: impl Into<String>,
-    ) {
+    ) -> Self {
         if AUDIT_LOGGER.is_none() {
-            return;
+            return self;
         }
         self.actor = Some(ActorLog {
             remote_host: host.into(),
             user_agent: user_agent.into(),
             username: username.into(),
             authorization_method: auth_method.into(),
         });
+
+        self
     }
 
     pub fn set_request(
-        &mut self,
+        mut self,
         method: impl Into<String>,
         path: impl Into<String>,
         protocol: impl Into<String>,
         headers: impl IntoIterator<Item = (String, String)>,
-    ) {
+    ) -> Self {
         if AUDIT_LOGGER.is_none() {
-            return;
+            return self;
         }
         self.request = Some(RequestLog {
             method: method.into(),
             path: path.into(),
             protocol: protocol.into(),
             headers: headers.into_iter().collect(),
         });
+
+        self
     }
 
-    pub fn set_response(&mut self, status_code: u16, err: impl Display) {
+    pub fn set_response(mut self, status_code: u16, err: impl Display) -> Self {
         if AUDIT_LOGGER.is_none() {
-            return;
+            return self;
         }
         let error = err.to_string();
         let error = error.is_empty().then_some(error);
         self.response = Some(ResponseLog { status_code, error });
+
+        self
     }
 
     // NOTE: Ensure that the logger has been constructed by Default

src/handlers/http/audit.rs

@@ -44,12 +44,12 @@ pub async fn audit_log_middleware(
     {
         let attribute_value: &str = kinesis_common_attributes.to_str().unwrap();
         let message: Message = serde_json::from_str(attribute_value).unwrap();
-        log_builder.set_stream_name(message.common_attributes.x_p_stream);
+        log_builder = log_builder.set_stream_name(message.common_attributes.x_p_stream);
     } else if let Some(stream) = req.match_info().get("logstream") {
-        log_builder.set_stream_name(stream.to_owned());
+        log_builder = log_builder.set_stream_name(stream.to_owned());
     } else if let Some(value) = req.headers().get(STREAM_NAME_HEADER_KEY) {
         if let Ok(stream) = value.to_str() {
-            log_builder.set_stream_name(stream.to_owned());
+            log_builder = log_builder.set_stream_name(stream.to_owned());
         }
     }
     let mut username = "Unknown".to_owned();
@@ -84,24 +84,24 @@ pub async fn audit_log_middleware(
                     .ok()
             }
         });
-    log_builder.set_request(
-        req.method().as_str(),
-        req.path(),
-        req.connection_info().scheme(),
-        headers,
-    );
-
-    log_builder.set_actor(
-        req.connection_info()
-            .realip_remote_addr()
-            .unwrap_or_default(),
-        req.headers()
-            .get("User-Agent")
-            .and_then(|a| a.to_str().ok())
-            .unwrap_or_default(),
-        username,
-        authorization_method,
-    );
+    log_builder = log_builder
+        .set_request(
+            req.method().as_str(),
+            req.path(),
+            req.connection_info().scheme(),
+            headers,
+        )
+        .set_actor(
+            req.connection_info()
+                .realip_remote_addr()
+                .unwrap_or_default(),
+            req.headers()
+                .get("User-Agent")
+                .and_then(|a| a.to_str().ok())
+                .unwrap_or_default(),
+            username,
+            authorization_method,
+        );
 
     let res = next.call(req).await;
 
@@ -111,10 +111,10 @@ pub async fn audit_log_middleware(
             let status = res.status();
             // Use error information from reponse object if an error
             if let Some(err) = res.response().error() {
-                log_builder.set_response(status.as_u16(), err);
+                log_builder = log_builder.set_response(status.as_u16(), err);
             }
         }
-        Err(err) => log_builder.set_response(500, err),
+        Err(err) => log_builder = log_builder.set_response(500, err),
     }
 
     log_builder.send().await;

src/kafka.rs

@@ -276,23 +276,21 @@ pub async fn setup_integration() {
 
     while let Ok(curr) = stream.next().await.unwrap() {
         // Constructs a log for each kafka request
-        let mut log_builder = AuditLogBuilder::default();
-        log_builder.set_actor(
-            CONFIG.parseable.kafka_host.as_deref().unwrap_or(""),
-            "Kafka Client",
-            "",
-            "",
-        );
-
-        log_builder.set_request("", "", "Kafka", []);
-        log_builder.set_stream_name(curr.topic());
+        let log_builder = AuditLogBuilder::default()
+            .set_actor(
+                CONFIG.parseable.kafka_host.as_deref().unwrap_or(""),
+                "Kafka Client",
+                "",
+                "",
+            )
+            .set_request("", "", "Kafka", [])
+            .set_stream_name(curr.topic());
 
         let Err(err) = ingest_message(curr).await else {
-            log_builder.set_response(200, "");
+            log_builder.set_response(200, "").send().await;
             continue;
         };
-        log_builder.set_response(500, &err);
         error!("Unable to ingest incoming kafka message- {err}");
-        log_builder.send().await;
+        log_builder.set_response(500, &err).send().await;
     }
 }