Add support for .env (#1368)

.gitignore

@@ -8,3 +8,4 @@ bench_*
 /rust-clippy-results.sarif
 /Cross.toml
 /performance_results*
+.env

bench/src/benchmarks/benchmark.rs

@@ -55,7 +55,6 @@ pub trait Benchmarkable {
     fn display_settings(&self);
 
     /// Below methods have common implementation for all benchmarks.
-
     /// Initializes the streams and topics for the benchmark.
     /// This method is called before the benchmark is executed.
     async fn init_streams(&self) -> Result<(), IggyError> {

integration/tests/examples/test_basic.rs

@@ -7,7 +7,7 @@ struct TestBasic<'a> {
     expected_consumer_output: Vec<&'a str>,
 }
 
-impl<'a> IggyExampleTestCase for TestBasic<'a> {
+impl IggyExampleTestCase for TestBasic<'_> {
     fn verify_log_output(&self, producer_stdout: &str, consumer_stdout: &str) {
         verify_stdout_contains_expected_logs(
             producer_stdout,

integration/tests/examples/test_getting_started.rs

@@ -16,7 +16,7 @@ struct TestGettingStarted<'a> {
     expected_consumer_output: Vec<&'a str>,
 }
 
-impl<'a> IggyExampleTestCase for TestGettingStarted<'a> {
+impl IggyExampleTestCase for TestGettingStarted<'_> {
     fn verify_log_output(&self, producer_stdout: &str, consumer_stdout: &str) {
         verify_stdout_contains_expected_logs(
             producer_stdout,

integration/tests/examples/test_message_envelope.rs

@@ -7,7 +7,7 @@ struct TestMessageEnvelope<'a> {
     expected_consumer_output: Vec<&'a str>,
 }
 
-impl<'a> IggyExampleTestCase for TestMessageEnvelope<'a> {
+impl IggyExampleTestCase for TestMessageEnvelope<'_> {
     fn verify_log_output(&self, producer_stdout: &str, consumer_stdout: &str) {
         verify_stdout_contains_expected_logs(
             producer_stdout,

integration/tests/examples/test_message_headers.rs

@@ -7,7 +7,7 @@ struct TestMessageHeaders<'a> {
     expected_consumer_output: Vec<&'a str>,
 }
 
-impl<'a> IggyExampleTestCase for TestMessageHeaders<'a> {
+impl IggyExampleTestCase for TestMessageHeaders<'_> {
     fn verify_log_output(&self, producer_stdout: &str, consumer_stdout: &str) {
         verify_stdout_contains_expected_logs(
             producer_stdout,

sdk/src/compression/compression_algorithm.rs

@@ -81,7 +81,7 @@ impl From<CompressionAlgorithm> for String {
 }
 struct CompressionKindVisitor;
 
-impl<'de> Visitor<'de> for CompressionKindVisitor {
+impl Visitor<'_> for CompressionKindVisitor {
     type Value = CompressionAlgorithm;
 
     fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {

sdk/src/locking/fast_async_lock.rs

@@ -9,8 +9,14 @@ impl<T> IggySharedMutFn<T> for IggyFastAsyncRwLock<T>
 where
     T: Send + Sync,
 {
-    type ReadGuard<'a> = RwLockReadGuard<'a, T> where T: 'a;
-    type WriteGuard<'a> = RwLockWriteGuard<'a, T> where T: 'a;
+    type ReadGuard<'a>
+        = RwLockReadGuard<'a, T>
+    where
+        T: 'a;
+    type WriteGuard<'a>
+        = RwLockWriteGuard<'a, T>
+    where
+        T: 'a;
     fn new(data: T) -> Self {
         IggyFastAsyncRwLock(Arc::new(FastAsyncRwLock::new(data)))
     }

sdk/src/locking/tokio_lock.rs

@@ -10,8 +10,14 @@ impl<T> IggySharedMutFn<T> for IggyTokioRwLock<T>
 where
     T: Send + Sync,
 {
-    type ReadGuard<'a> = RwLockReadGuard<'a, T> where T: 'a;
-    type WriteGuard<'a> = RwLockWriteGuard<'a, T> where T: 'a;
+    type ReadGuard<'a>
+        = RwLockReadGuard<'a, T>
+    where
+        T: 'a;
+    type WriteGuard<'a>
+        = RwLockWriteGuard<'a, T>
+    where
+        T: 'a;
 
     fn new(data: T) -> Self {
         IggyTokioRwLock(Arc::new(TokioRwLock::new(data)))

sdk/src/utils/duration.rs

@@ -168,7 +168,7 @@ impl<'de> Deserialize<'de> for IggyDuration {
     }
 }
 
-impl<'de> Visitor<'de> for IggyDurationVisitor {
+impl Visitor<'_> for IggyDurationVisitor {
     type Value = IggyDuration;
 
     fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result {

sdk/src/utils/expiry.rs

@@ -178,7 +178,7 @@ impl<'de> Deserialize<'de> for IggyExpiry {
 
 struct IggyExpiryVisitor;
 
-impl<'de> Visitor<'de> for IggyExpiryVisitor {
+impl Visitor<'_> for IggyExpiryVisitor {
     type Value = IggyExpiry;
 
     fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {

sdk/src/utils/timestamp.rs

@@ -105,7 +105,7 @@ impl<'de> Deserialize<'de> for IggyTimestamp {
 }
 struct IggyTimestampVisitor;
 
-impl<'de> Visitor<'de> for IggyTimestampVisitor {
+impl Visitor<'_> for IggyTimestampVisitor {
     type Value = IggyTimestamp;
 
     fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {

sdk/src/utils/topic_size.rs

@@ -118,7 +118,7 @@ impl Serialize for MaxTopicSize {
 
 struct MaxTopicSizeVisitor;
 
-impl<'de> Visitor<'de> for MaxTopicSizeVisitor {
+impl Visitor<'_> for MaxTopicSizeVisitor {
     type Value = MaxTopicSize;
 
     fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {

server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.82"
+version = "0.4.83"
 edition = "2021"
 build = "src/build.rs"
 
@@ -25,6 +25,7 @@ clap = { version = "4.5.17", features = ["derive"] }
 console-subscriber = { version = "0.4.0", optional = true }
 dashmap = "6.0.1"
 derive_more = "1.0.0"
+dotenvy = { version = "0.15.7" }
 figlet-rs = "0.1.5"
 figment = { version = "0.10.18", features = ["toml", "env"] }
 flume = "0.11.0"

server/src/configs/resource_quota.rs

@@ -54,7 +54,7 @@ impl FromStr for MemoryResourceQuota {
 
 struct ResourceQuotaVisitor;
 
-impl<'de> Visitor<'de> for ResourceQuotaVisitor {
+impl Visitor<'_> for ResourceQuotaVisitor {
     type Value = MemoryResourceQuota;
 
     fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {

server/src/configs/validators.rs

@@ -188,7 +188,7 @@ impl Validatable<ServerError> for ArchiverConfig {
             return Ok(());
         }
 
-        return match self.kind {
+        match self.kind {
             ArchiverKind::Disk => {
                 if self.disk.is_none() {
                     return Err(ServerError::InvalidConfiguration(
@@ -245,7 +245,7 @@ impl Validatable<ServerError> for ArchiverConfig {
                 }
                 Ok(())
             }
-        };
+        }
     }
 }
 

server/src/main.rs

@@ -1,5 +1,6 @@
 use anyhow::Result;
 use clap::Parser;
+use dotenvy::dotenv;
 use figlet_rs::FIGfont;
 use server::args::Args;
 use server::channels::commands::archive_state::ArchiveStateExecutor;
@@ -31,6 +32,13 @@ async fn main() -> Result<(), ServerError> {
     let figure = standard_font.convert("Iggy Server");
     println!("{}", figure.unwrap());
 
+    if let Ok(path) = dotenv() {
+        println!(
+            "Loaded environment variables from .env file at path: {}",
+            path.display()
+        );
+    }
+
     let args = Args::parse();
     let config_provider = config_provider::resolve(&args.config_provider)?;
     let config = ServerConfig::load(config_provider.as_ref()).await?;

server/src/streaming/batching/iterator.rs

@@ -24,7 +24,7 @@ impl<'a> RetainedMessageBatchIterator<'a> {
 
 // TODO(numinex): Consider using FallibleIterator instead of this
 // https://crates.io/crates/fallible-iterator
-impl<'a> Iterator for RetainedMessageBatchIterator<'a> {
+impl Iterator for RetainedMessageBatchIterator<'_> {
     type Item = RetainedMessage;
     fn next(&mut self) -> Option<Self::Item> {
         if self.current_position < self.batch.length.as_bytes_u64() {

server/src/streaming/cache/mod.rs

@@ -1,2 +1,3 @@
 pub mod buffer;
+#[allow(static_mut_refs)]
 pub mod memory_tracker;