Refactor: Replace dyn Trait with Enum for Encryptor and Persister Typ…

…es (#1442)

Addresses part of ##667

This PR refactors the codebase to replace the use of dyn Trait for
Encryptor and Persister with enum types (EncryptorKind and
PersisterKind)

Co-authored-by: Hubert Gruszecki <[email protected]>

cli/src/main.rs

@@ -65,7 +65,7 @@ use iggy::cli::{
 use iggy::cli_command::{CliCommand, PRINT_TARGET};
 use iggy::client_provider::{self, ClientProviderConfig};
 use iggy::clients::client::IggyClient;
-use iggy::utils::crypto::{Aes256GcmEncryptor, Encryptor};
+use iggy::utils::crypto::{Aes256GcmEncryptor, EncryptorKind};
 use iggy::utils::personal_access_token_expiry::PersonalAccessTokenExpiry;
 use std::sync::Arc;
 use tracing::{event, Level};
@@ -339,11 +339,11 @@ async fn main() -> Result<(), IggyCmdError> {
     // Create credentials based on command line arguments and command
     let mut credentials = IggyCredentials::new(&cli_options, &iggy_args, command.login_required())?;
 
-    let encryptor: Option<Arc<dyn Encryptor>> = match iggy_args.encryption_key.is_empty() {
+    let encryptor = match iggy_args.encryption_key.is_empty() {
         true => None,
-        false => Some(Arc::new(
+        false => Some(Arc::new(EncryptorKind::Aes256Gcm(
             Aes256GcmEncryptor::from_base64_key(&iggy_args.encryption_key).unwrap(),
-        )),
+        ))),
     };
     let client_provider_config = Arc::new(ClientProviderConfig::from_args_set_autologin(
         iggy_args.clone(),

integration/tests/state/mod.rs

@@ -1,6 +1,6 @@
-use iggy::utils::crypto::{Aes256GcmEncryptor, Encryptor};
+use iggy::utils::crypto::{Aes256GcmEncryptor, EncryptorKind};
 use server::state::file::FileState;
-use server::streaming::persistence::persister::FilePersister;
+use server::streaming::persistence::persister::{FilePersister, PersisterKind};
 use server::versioning::SemanticVersion;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -31,11 +31,12 @@ impl StateSetup {
         create_dir(&directory_path).await.unwrap();
 
         let version = SemanticVersion::from_str("1.2.3").unwrap();
-        let persister = FilePersister {};
-        let encryptor: Option<Arc<dyn Encryptor>> = match encryption_key {
-            Some(key) => Some(Arc::new(Aes256GcmEncryptor::new(key).unwrap())),
-            None => None,
-        };
+        let persister = PersisterKind::File(FilePersister {});
+        let encryptor = encryption_key.map(|key| {
+            Arc::new(EncryptorKind::Aes256Gcm(
+                Aes256GcmEncryptor::new(key).unwrap(),
+            ))
+        });
         let state = FileState::new(&log_path, &version, Arc::new(persister), encryptor);
 
         Self {

integration/tests/streaming/common/test_setup.rs

@@ -1,5 +1,5 @@
 use server::configs::system::SystemConfig;
-use server::streaming::persistence::persister::FilePersister;
+use server::streaming::persistence::persister::{FilePersister, PersisterKind};
 use server::streaming::storage::SystemStorage;
 use std::sync::Arc;
 use tokio::fs;
@@ -20,7 +20,7 @@ impl TestSetup {
 
         let config = Arc::new(config);
         fs::create_dir(config.get_system_path()).await.unwrap();
-        let persister = FilePersister {};
+        let persister = PersisterKind::File(FilePersister {});
         let storage = Arc::new(SystemStorage::new(config.clone(), Arc::new(persister)));
         TestSetup { config, storage }
     }

sdk/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "iggy"
-version = "0.6.71"
+version = "0.6.80"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2021"
 license = "Apache-2.0"

sdk/src/clients/builder.rs

@@ -8,7 +8,7 @@ use crate::quic::client::QuicClient;
 use crate::quic::config::QuicClientConfigBuilder;
 use crate::tcp::client::TcpClient;
 use crate::tcp::config::TcpClientConfigBuilder;
-use crate::utils::crypto::Encryptor;
+use crate::utils::crypto::EncryptorKind;
 use crate::utils::duration::IggyDuration;
 use std::sync::Arc;
 use tracing::error;
@@ -18,7 +18,7 @@ use tracing::error;
 pub struct IggyClientBuilder {
     client: Option<Box<dyn Client>>,
     partitioner: Option<Arc<dyn Partitioner>>,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
 }
 
 impl IggyClientBuilder {
@@ -49,7 +49,7 @@ impl IggyClientBuilder {
     }
 
     /// Use the custom encryptor implementation.
-    pub fn with_encryptor(mut self, encryptor: Arc<dyn Encryptor>) -> Self {
+    pub fn with_encryptor(mut self, encryptor: Arc<EncryptorKind>) -> Self {
         self.encryptor = Some(encryptor);
         self
     }

sdk/src/clients/client.rs

@@ -31,7 +31,7 @@ use crate::partitioner::Partitioner;
 use crate::snapshot::{SnapshotCompression, SystemSnapshotType};
 use crate::tcp::client::TcpClient;
 use crate::utils::byte_size::IggyByteSize;
-use crate::utils::crypto::Encryptor;
+use crate::utils::crypto::EncryptorKind;
 use crate::utils::duration::IggyDuration;
 use crate::utils::expiry::IggyExpiry;
 use crate::utils::personal_access_token_expiry::PersonalAccessTokenExpiry;
@@ -55,7 +55,7 @@ use tracing::{debug, error, info};
 pub struct IggyClient {
     client: IggySharedMut<Box<dyn Client>>,
     partitioner: Option<Arc<dyn Partitioner>>,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
 }
 
 impl Default for IggyClient {
@@ -96,7 +96,7 @@ impl IggyClient {
     pub fn create(
         client: Box<dyn Client>,
         partitioner: Option<Arc<dyn Partitioner>>,
-        encryptor: Option<Arc<dyn Encryptor>>,
+        encryptor: Option<Arc<EncryptorKind>>,
     ) -> Self {
         if partitioner.is_some() {
             info!("Partitioner is enabled.");

sdk/src/clients/consumer.rs

@@ -7,7 +7,7 @@ use crate::locking::{IggySharedMut, IggySharedMutFn};
 use crate::messages::poll_messages::{PollingKind, PollingStrategy};
 use crate::models::messages::{PolledMessage, PolledMessages};
 use crate::utils::byte_size::IggyByteSize;
-use crate::utils::crypto::Encryptor;
+use crate::utils::crypto::EncryptorKind;
 use crate::utils::duration::IggyDuration;
 use crate::utils::timestamp::IggyTimestamp;
 use bytes::Bytes;
@@ -81,7 +81,7 @@ pub struct IggyConsumer {
     current_offsets: Arc<DashMap<u32, AtomicU64>>,
     poll_future: Option<PollMessagesFuture>,
     buffered_messages: VecDeque<PolledMessage>,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
     store_offset_sender: flume::Sender<(u32, u64)>,
     store_offset_after_each_message: bool,
     store_offset_after_all_messages: bool,
@@ -107,7 +107,7 @@ impl IggyConsumer {
         auto_commit: AutoCommit,
         auto_join_consumer_group: bool,
         create_consumer_group_if_not_exists: bool,
-        encryptor: Option<Arc<dyn Encryptor>>,
+        encryptor: Option<Arc<EncryptorKind>>,
         retry_interval: IggyDuration,
         allow_replay: bool,
     ) -> Self {
@@ -866,7 +866,7 @@ pub struct IggyConsumerBuilder {
     auto_commit: AutoCommit,
     auto_join_consumer_group: bool,
     create_consumer_group_if_not_exists: bool,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
     retry_interval: IggyDuration,
     allow_replay: bool,
 }
@@ -880,7 +880,7 @@ impl IggyConsumerBuilder {
         stream_id: Identifier,
         topic_id: Identifier,
         partition_id: Option<u32>,
-        encryptor: Option<Arc<dyn Encryptor>>,
+        encryptor: Option<Arc<EncryptorKind>>,
         polling_interval: Option<IggyDuration>,
     ) -> Self {
         Self {
@@ -990,7 +990,7 @@ impl IggyConsumerBuilder {
     }
 
     /// Sets the encryptor for decrypting the messages' payloads.
-    pub fn encryptor(self, encryptor: Arc<dyn Encryptor>) -> Self {
+    pub fn encryptor(self, encryptor: Arc<EncryptorKind>) -> Self {
         Self {
             encryptor: Some(encryptor),
             ..self

sdk/src/clients/producer.rs

@@ -6,7 +6,7 @@ use crate::identifier::{IdKind, Identifier};
 use crate::locking::{IggySharedMut, IggySharedMutFn};
 use crate::messages::send_messages::{Message, Partitioning};
 use crate::partitioner::Partitioner;
-use crate::utils::crypto::Encryptor;
+use crate::utils::crypto::EncryptorKind;
 use crate::utils::duration::IggyDuration;
 use crate::utils::expiry::IggyExpiry;
 use crate::utils::timestamp::IggyTimestamp;
@@ -35,7 +35,7 @@ pub struct IggyProducer {
     topic_name: String,
     batch_size: Option<usize>,
     partitioning: Option<Arc<Partitioning>>,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
     partitioner: Option<Arc<dyn Partitioner>>,
     send_interval_micros: u64,
     create_stream_if_not_exists: bool,
@@ -60,7 +60,7 @@ impl IggyProducer {
         topic_name: String,
         batch_size: Option<usize>,
         partitioning: Option<Partitioning>,
-        encryptor: Option<Arc<dyn Encryptor>>,
+        encryptor: Option<Arc<EncryptorKind>>,
         partitioner: Option<Arc<dyn Partitioner>>,
         interval: Option<IggyDuration>,
         create_stream_if_not_exists: bool,
@@ -423,7 +423,7 @@ pub struct IggyProducerBuilder {
     topic_name: String,
     batch_size: Option<usize>,
     partitioning: Option<Partitioning>,
-    encryptor: Option<Arc<dyn Encryptor>>,
+    encryptor: Option<Arc<EncryptorKind>>,
     partitioner: Option<Arc<dyn Partitioner>>,
     send_interval: Option<IggyDuration>,
     create_stream_if_not_exists: bool,
@@ -443,7 +443,7 @@ impl IggyProducerBuilder {
         stream_name: String,
         topic: Identifier,
         topic_name: String,
-        encryptor: Option<Arc<dyn Encryptor>>,
+        encryptor: Option<Arc<EncryptorKind>>,
         partitioner: Option<Arc<dyn Partitioner>>,
     ) -> Self {
         Self {
@@ -514,7 +514,7 @@ impl IggyProducerBuilder {
     }
 
     /// Sets the encryptor for encrypting the messages' payloads.
-    pub fn encryptor(self, encryptor: Arc<dyn Encryptor>) -> Self {
+    pub fn encryptor(self, encryptor: Arc<EncryptorKind>) -> Self {
         Self {
             encryptor: Some(encryptor),
             ..self

sdk/src/utils/crypto.rs

@@ -5,7 +5,25 @@ use aes_gcm::aead::{Aead, OsRng};
 use aes_gcm::{AeadCore, Aes256Gcm, KeyInit};
 use std::fmt::Debug;
 
-pub trait Encryptor: Send + Sync + Debug {
+#[derive(Debug)]
+pub enum EncryptorKind {
+    Aes256Gcm(Aes256GcmEncryptor),
+}
+
+impl EncryptorKind {
+    pub fn encrypt(&self, data: &[u8]) -> Result<Vec<u8>, IggyError> {
+        match self {
+            EncryptorKind::Aes256Gcm(e) => e.encrypt(data),
+        }
+    }
+    pub fn decrypt(&self, data: &[u8]) -> Result<Vec<u8>, IggyError> {
+        match self {
+            EncryptorKind::Aes256Gcm(e) => e.decrypt(data),
+        }
+    }
+}
+
+pub trait Encryptor {
     fn encrypt(&self, data: &[u8]) -> Result<Vec<u8>, IggyError>;
     fn decrypt(&self, data: &[u8]) -> Result<Vec<u8>, IggyError>;
 }
@@ -14,9 +32,6 @@ pub struct Aes256GcmEncryptor {
     cipher: Aes256Gcm,
 }
 
-unsafe impl Send for Aes256GcmEncryptor {}
-unsafe impl Sync for Aes256GcmEncryptor {}
-
 impl Debug for Aes256GcmEncryptor {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Encryptor").finish()

server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.112"
+version = "0.4.120"
 edition = "2021"
 build = "src/build.rs"
 license = "Apache-2.0"
@@ -90,6 +90,9 @@ uuid = { version = "1.11.0", features = ["v7", "fast-rng", "zerocopy"] }
 xxhash-rust = { version = "0.8.15", features = ["xxh32"] }
 zip = "2.2.2"
 
+[dev-dependencies]
+mockall = "0.13.1"
+
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 mimalloc = { version = "0.1", optional = true }