chore: Fix lint warnings across the board

book/src/SUMMARY.md

@@ -36,4 +36,4 @@
     - [Intersect Options](./advanced/intersect_options.md)
 - [Guides](./guides/README.md)
     - [Cardano => Kafka](./guides/cardano_2_kafka.md)
-    - [Custom networks](./guides/connecting_to_custom_networks.md)
+    - [Custom networks](./guides/connecting_to_custom_networks.md)

book/src/guides/connecting_to_custom_networks.md

@@ -27,17 +27,17 @@ adahandle_policy = String
 
 Some details on the cofigurataion:
 
-| Name                 | DataType | Description              |
-| :---                 | :---     | :---                     |
-| byron_epoch_length   | u32      | ....                     |
-| byron_slot_length    | u32      | ....                     |
-| byron_known_slot     | u32      | ....                     |
-| byron_known_hash     | String   | ....                     |
-| byron_known_time     | u64      | ....                     |
-| shelley_epoch_length | u32      | ....                     |
-| shelley_slot_length  | u32      | ....                     |
-| shelley_known_slot   | u32      | ....                     |
-| shelley_known_hash   | String   | ....                     |
-| shelley_known_time   | u64      | ....                     |
-| address_hrp          | String   | ....                     |
-| adahandle_policy     | String   | Minting policy of AdaHandle on the network. |
+| Name                 | DataType | Description                                 |
+| :------------------- | :------- | :------------------------------------------ |
+| byron_epoch_length   | u32      | ....                                        |
+| byron_slot_length    | u32      | ....                                        |
+| byron_known_slot     | u32      | ....                                        |
+| byron_known_hash     | String   | ....                                        |
+| byron_known_time     | u64      | ....                                        |
+| shelley_epoch_length | u32      | ....                                        |
+| shelley_slot_length  | u32      | ....                                        |
+| shelley_known_slot   | u32      | ....                                        |
+| shelley_known_hash   | String   | ....                                        |
+| shelley_known_time   | u64      | ....                                        |
+| address_hrp          | String   | ....                                        |
+| adahandle_policy     | String   | Minting policy of AdaHandle on the network. |

book/src/sinks/redis_streams.md

@@ -32,4 +32,4 @@ stream_strategy = "ByEventType"
 - `type`: the literal value `Redis`.
 - `redis_server`: the redis server in the format `redis://[<username>][:<password>]@<hostname>[:port][/<db>]`
 - `stream_name` : the name of the redis stream for StreamStrategy `None`, default is "oura" if not specified
-- `stream_strategy` : `None` or `ByEventType`
+- `stream_strategy` : `None` or `ByEventType`

src/bin/oura/daemon.rs

@@ -124,13 +124,11 @@ enum Sink {
     #[cfg(feature = "aws")]
     AwsS3(AwsS3Config),
 
-
     #[cfg(feature = "redissink")]
     Redis(RedisConfig),
-  
+
     #[cfg(feature = "gcp")]
     GcpPubSub(GcpPubSubConfig),
-
 }
 
 fn bootstrap_sink(config: Sink, input: StageReceiver, utils: Arc<Utils>) -> BootstrapResult {
@@ -165,7 +163,6 @@ fn bootstrap_sink(config: Sink, input: StageReceiver, utils: Arc<Utils>) -> Boot
 
         #[cfg(feature = "gcp")]
         Sink::GcpPubSub(c) => WithUtils::new(c, utils).bootstrap(input),
-
     }
 }
 

src/sinks/mod.rs

@@ -31,4 +31,4 @@ pub mod aws_s3;
 pub mod redis;
 
 #[cfg(feature = "gcp")]
-pub mod gcp_pubsub;
+pub mod gcp_pubsub;

src/sinks/redis/mod.rs

@@ -1,3 +1,3 @@
 mod run;
 mod setup;
-pub use setup::*;
+pub use setup::*;

src/sinks/redis/run.rs

@@ -1,27 +1,24 @@
 #![allow(unused_variables)]
-use std::sync::Arc;
+use super::StreamStrategy;
+use crate::{model::Event, pipelining::StageReceiver, utils::Utils, Error};
 use serde::Serialize;
-use serde_json::{json};
-use crate::{pipelining::StageReceiver, utils::Utils, Error, model::Event};
-use super::{StreamStrategy};
+use serde_json::json;
+use std::sync::Arc;
 
 #[derive(Serialize)]
 pub struct RedisRecord {
-    pub event: Event, 
+    pub event: Event,
     pub key: String,
 }
 
 impl From<Event> for RedisRecord {
     fn from(event: Event) -> Self {
         let key = key(&event);
-        RedisRecord {
-            event,
-            key,
-        }
+        RedisRecord { event, key }
     }
 }
 
-fn key(event :  &Event) -> String {
+fn key(event: &Event) -> String {
     if let Some(fingerprint) = &event.fingerprint {
         fingerprint.clone()
     } else {
@@ -30,26 +27,34 @@ fn key(event :  &Event) -> String {
 }
 
 pub fn producer_loop(
-    input           : StageReceiver,
-    utils           : Arc<Utils>,
-    conn            : &mut redis::Connection,
-    stream_strategy : StreamStrategy,
-    redis_stream    : String,
+    input: StageReceiver,
+    utils: Arc<Utils>,
+    conn: &mut redis::Connection,
+    stream_strategy: StreamStrategy,
+    redis_stream: String,
 ) -> Result<(), Error> {
     for event in input.iter() {
         utils.track_sink_progress(&event);
         let payload = RedisRecord::from(event);
-        let stream : String; 
-        match stream_strategy {
-            StreamStrategy::ByEventType => {
-                stream = payload.event.data.clone().to_string().to_lowercase();
-            }
-            _ => {
-                stream = redis_stream.clone();
-            }
-        }
-        log::debug!("Stream: {:?}, Key: {:?}, Event: {:?}", stream, payload.key, payload.event);
-        let _ : () = redis::cmd("XADD").arg(stream).arg("*").arg(&[(payload.key,json!(payload.event).to_string())]).query(conn)?;
+
+        let stream = match stream_strategy {
+            StreamStrategy::ByEventType => payload.event.data.clone().to_string().to_lowercase(),
+            _ => redis_stream.clone(),
+        };
+
+        log::debug!(
+            "Stream: {:?}, Key: {:?}, Event: {:?}",
+            stream,
+            payload.key,
+            payload.event
+        );
+
+        let _: () = redis::cmd("XADD")
+            .arg(stream)
+            .arg("*")
+            .arg(&[(payload.key, json!(payload.event).to_string())])
+            .query(conn)?;
     }
+
     Ok(())
-}
+}

src/sinks/redis/setup.rs

@@ -1,4 +1,4 @@
-use redis::{Client};
+use redis::Client;
 use serde::Deserialize;
 
 use crate::{
@@ -11,32 +11,39 @@ use super::run::*;
 #[derive(Debug, Clone, Deserialize)]
 pub enum StreamStrategy {
     ByEventType,
-    None
-}   
+    None,
+}
 
 #[derive(Debug, Deserialize)]
 pub struct Config {
-    pub redis_server      : String,
-    pub stream_strategy   : Option<StreamStrategy>,
-    pub stream_name       : Option<String>,
+    pub redis_server: String,
+    pub stream_strategy: Option<StreamStrategy>,
+    pub stream_name: Option<String>,
 }
 
 impl SinkProvider for WithUtils<Config> {
     fn bootstrap(&self, input: StageReceiver) -> BootstrapResult {
         let client = Client::open(self.inner.redis_server.clone())?;
         let mut connection = client.get_connection()?;
         log::debug!("Connected to Redis Database!");
+
         let stream_strategy = match self.inner.stream_strategy.clone() {
-            Some(strategy)  => { 
-                strategy
-            },
-            _               => StreamStrategy::None
+            Some(strategy) => strategy,
+            _ => StreamStrategy::None,
         };
-        let redis_stream = self.inner.stream_name.clone().unwrap_or("oura".to_string());
+
+        let redis_stream = self
+            .inner
+            .stream_name
+            .clone()
+            .unwrap_or_else(|| "oura".to_string());
+
         let utils = self.utils.clone();
         let handle = std::thread::spawn(move || {
-            producer_loop(input, utils, &mut connection, stream_strategy, redis_stream).expect("redis sink loop failed");
+            producer_loop(input, utils, &mut connection, stream_strategy, redis_stream)
+                .expect("redis sink loop failed");
         });
+
         Ok(handle)
     }
 }