v0.4.0

.github/workflows/CI.yml

@@ -22,9 +22,8 @@ jobs:
       - name: Git checkout
         uses: actions/checkout@v3
 
-      - name: Build proyect
+      - name: Build project
         run: cargo build --verbose
 
-      - name: Run proyect tests
+      - name: Run project tests
         run: cargo test --verbose
-

Cargo.toml

@@ -1,35 +1,5 @@
-[package]
-name = "redsumer-rs"
-description = "Lightweight implementation of Redis Streams for Rust"
-version = "0.3.3"
-edition = "2021"
-license-file = "LICENSE"
-readme = "README.md"
-keywords = [
-    "redis",
-    "redis_streams",
-]
-repository = "https://github.com/enerBit/redsumer-rs"
-categories = [
-    "database-implementations",
-]
-authors = [
-    "enerBit",
-    "Juan Manuel Tamayo <[email protected]>",
-]
-
-[dependencies]
-redis = { version = "0.24.0", features = ["tokio-comp", "streams"] }
-tokio = { version = "1.35.1", features = ["full"] }
-log = { version = "0.4.20" }
-async-trait = { version = "0.1.77" }
-uuid = { version = "1.7.0", features = ["v4"] }
-time = { version = "0.3.31", features = ["formatting", "parsing"] }
-bytes = { version = "1.5.0", features = [] }
-serde = { version = "1.0.195", features = ["derive"] }
-serde_json = { version = "1.0.111", features = [] }
-
-
-[dev-dependencies]
-structmap = { version = "0.1.6" }
-structmap-derive = { version = "0.1.6" }
+[workspace]
+resolver="2"
+members = [
+    "redsumer-rs",
+]

README.md

@@ -1,20 +1,85 @@
 # redsumer-rs
 
 A **lightweight implementation** of **Redis Streams** for Rust, that allows you managing streaming messages in an easy way.
-With redsumer-rs you can:
-- **Produce** stream messages in specific *database* and *stream name*.
-- **Consume** stream messages from specific *stream name* and *consumer group*, setting config parameters like *min idle time*, *fisrt id*, *max number of messages to consume* and more. Also you can set message as consumed for specific *consumer group*.
-To use redsumer-rs from github repository, set the dependency in Cargo.toml file as follows:
-
-```ini
-[dependencies]
-redsumer-rs = {git = "https://github.com/enerBit/redsumer-rs"}
+With `redsumer-rs` you can:
+- **Produce** new messages in a specific *stream*.
+- **Consume** messages from specific *stream*, setting config parameters that allow you a flexible implementation.
+
+For more dependency options from git, check section [3.0 Cargo Reference](https://doc.rust-lang.org/cargo/reference/index.html) from `The Cargo Book`.
+
+## Basic Usage
+
+### Producer:
+
+Create a new producer instance:
+
+```rust,no_run
+    use redsumer::*;
+
+    let credentials: Option<ClientCredentials> = None;
+    let host: &str = "localhost";
+    let port: &str = "6379";
+    let db: &str = "0";
+    let stream_name: &str = "my-stream";
+
+    let producer_result: RedsumerResult<RedsumerProducer> =
+        RedsumerProducer::new(
+            credentials,
+            host,
+            port,
+            db,
+            stream_name,
+        );
+
+    let producer: RedsumerProducer = producer_result.unwrap();
 ```
 
-If you need to use an specific version, set the dependency in Cargo.toml file as follows:
- ```ini
-[dependencies]
-redsumer-rs = {git = "https://github.com/enerBit/redsumer-rs", version = "0.2.2"}
+### Consumer:
+
+Create a new consumer instance:
+
+```rust,no_run
+    use redsumer::*;
+
+    let credentials: Option<ClientCredentials> = None;
+    let host: &str = "localhost";
+    let port: &str = "6379";
+    let db: &str = "0";
+    let stream_name: &str = "my-stream";
+    let group_name: &str = "group-name";
+    let consumer_name: &str = "consumer";
+    let since_id: &str = "0-0";
+    let min_idle_time_milliseconds: usize = 1000;
+    let new_messages_count: usize = 3;
+    let pending_messages_count: usize = 2;
+    let claimed_messages_count: usize = 1;
+    let block: u8 = 5;
+
+    let consumer_result: RedsumerResult<RedsumerConsumer> = RedsumerConsumer::new(
+        credentials,
+        host,
+        port,
+        db,
+        stream_name,
+        group_name,
+        consumer_name,
+        since_id,
+        min_idle_time_milliseconds,
+        new_messages_count,
+        pending_messages_count,
+        claimed_messages_count,
+        block,
+    );
+
+    let mut consumer: RedsumerConsumer = consumer_result.unwrap();
 ```
 
-For more dependency options from git, check section [3.0 Cargo Reference](https://doc.rust-lang.org/cargo/reference/index.html) from `The Cargo Book`.
+## Contributing
+
+We welcome contributions to `redsumer-rs`. Here are some ways you can contribute:
+
+- **Bug Reports**: If you find a bug, please create an issue detailing the problem, the steps to reproduce it, and the expected behavior.
+- **Feature Requests**: If you have an idea for a new feature or an enhancement to an existing one, please create an issue describing your idea.
+- **Pull Requests**: If you've fixed a bug or implemented a new feature, we'd love to see your work! Please submit a pull request. Make sure your code follows the existing style and all tests pass.
+
+Thank you for your interest in improving `redsumer-rs`!

redsumer-rs/CHANGELOG.md

@@ -0,0 +1,35 @@
+# Changelog 📘💜
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## ✨ v0.4.0 [2024-04-23]
+
+### Added:
+
+- ⚡ Implementation of new types: RedsumerResult, RedsumerError and Id (Breaking change).
+- ⚡ Debug and Clone implementation in RedsumerProducer and RedsumerConsumer.
+- ⚡ The consumer configuration parameters were implemented directly in RedsumerConsumer (Breaking change).
+
+### Fixed:
+
+- 🛠 General refactoring of the package in order to improve performance.
+
+### Changed:
+
+- 🚀 New project structure as workspace.
+- 🚀 Update dependencies and documentation.
+- 🚀 Library modules reorganization (Breaking change).
+- 🚀 FromRedisValueImplHandler was changed to FromRedisValueHandler (Breaking change).
+- 🚀 The produce_from_map method was replaced by the produce method in RedsumerProducer (Breaking change).
+- 🚀 The validate_pending_message_ownership method was replaced by is_still_mine in RedsumerConsumer (Breaking change).
+- 🚀 The acknowledge method was replaced by ack in RedsumerConsumer (Breaking change).
+- 🚀 The consume method was refactored in RedsumerConsumer in order to implement a new consumption methodology that allows scalability in distributed systems. To understand this new implementation in detail, take a look at the project https://github.com/elpablete/refactored-computing-machine.
+
+### Removed:
+
+- ❌ The stream_information.rs module was removed from the project: StreamInfo and StreamConsumersInfo implementations were removed (Breaking change).
+- ❌ RedsumerConsumerOptions was removed (Breaking change).
+- ❌ The produce_from_items method was removed from RedsumerProducer (Breaking change).

redsumer-rs/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "redsumer"
+description = "Lightweight implementation of Redis Streams for Rust"
+version = "0.4.0"
+edition = "2021"
+license-file = "LICENSE"
+readme = "README.md"
+keywords = [
+    "redis",
+    "redis_streams",
+]
+homepage = "https://github.com/enerBit/redsumer-rs/redsumer-rs"
+repository = "https://github.com/enerBit/redsumer-rs/redsumer-rs"
+documentation = "https://docs.rs/redsumer"
+categories = [
+    "database-implementations",
+]
+authors = [
+    "enerBit",
+    "Juan Manuel Tamayo <[email protected]>",
+]
+
+[dependencies]
+redis = { version = "0.25.3", features = ["tokio-comp", "streams"] }
+tokio = { version = "1.37.0", features = ["full"] }
+uuid = { version = "1.8.0" }
+time = { version = "0.3.36", features = ["parsing"] }
+bytes = { version = "1.6.0" }
+serde = { version = "1.0.198", features = ["derive"] }
+serde_json = { version = "1.0.116" }
+log = { version = "0.4.21" }
+
+[dev-dependencies]
+structmap = { version = "0.1.6" }
+structmap-derive = { version = "0.1.6" }

redsumer-rs/LICENSE

@@ -0,0 +1 @@
+../LICENSE

redsumer-rs/README.md

@@ -0,0 +1 @@
+../README.md

redsumer-rs/src/lib.rs

@@ -0,0 +1,91 @@
+//! A **lightweight implementation** of **Redis Streams** for Rust, that allows you managing streaming messages in an easy way.
+//! With `redsumer-rs` you can:
+//! - **Produce** new messages in a specific *stream*.
+//! - **Consume** messages from specific *stream*, setting config parameters that allow you a flexible implementation.
+//!
+//! For more dependency options from git, check section [3.0 Cargo Reference](https://doc.rust-lang.org/cargo/reference/index.html) from `The Cargo Book`.
+//!
+//! ## Basic Usage
+//!
+//! Create a new producer instance:
+//!
+//! ```rust,no_run
+//! use redsumer::*;
+//!     
+//! let credentials: Option<ClientCredentials> = None;
+//! let host: &str = "localhost";
+//! let port: &str = "6379";
+//! let db: &str = "0";
+//! let stream_name: &str = "my-stream";
+//!
+//! let producer_result: RedsumerResult<RedsumerProducer> =
+//!     RedsumerProducer::new(
+//!     credentials,
+//!     host,
+//!     port,
+//!     db,
+//!     stream_name,
+//! );
+//!
+//! let producer: RedsumerProducer = producer_result.unwrap();
+//! ```
+//!
+//! Create a new consumer instance:
+//!
+//! ```rust,no_run
+//! use redsumer::*;
+//!
+//! let credentials: Option<ClientCredentials> = None;
+//! let host: &str = "localhost";
+//! let port: &str = "6379";
+//! let db: &str = "0";
+//! let stream_name: &str = "my-stream";
+//! let group_name: &str = "group-name";
+//! let consumer_name: &str = "consumer";
+//! let since_id: &str = "0-0";
+//! let min_idle_time_milliseconds: usize = 1000;
+//! let new_messages_count: usize = 3;
+//! let pending_messages_count: usize = 2;
+//! let claimed_messages_count: usize = 1;
+//! let block: u8 = 5;
+//!
+//! let consumer_result: RedsumerResult<RedsumerConsumer> = RedsumerConsumer::new(
+//!     credentials,
+//!     host,
+//!     port,
+//!     db,
+//!     stream_name,
+//!     group_name,
+//!     consumer_name,
+//!     since_id,
+//!     min_idle_time_milliseconds,
+//!     new_messages_count,
+//!     pending_messages_count,
+//!     claimed_messages_count,
+//!     block,
+//! );
+//!
+//! let mut consumer: RedsumerConsumer = consumer_result.unwrap();
+//! ```
+//!
+//! ## Contributing
+//!
+//! We welcome contributions to `redsumer-rs`. Here are some ways you can contribute:
+//!
+//! - **Bug Reports**: If you find a bug, please create an issue detailing the problem, the steps to reproduce it, and the expected behavior.
+//! - **Feature Requests**: If you have an idea for a new feature or an enhancement to an existing one, please create an issue describing your idea.
+//! - **Pull Requests**: If you've fixed a bug or implemented a new feature, we'd love to see your work! Please submit a pull request. Make sure your code follows the existing style and all tests pass.
+//!
+//! Thank you for your interest in improving `redsumer-rs`!
+mod redsumer;
+
+pub use redsumer::client::ClientCredentials;
+pub use redsumer::consumer::*;
+pub use redsumer::producer::*;
+pub use redsumer::types::*;
+
+pub mod redis {
+    //! Utilities from [redis] crate.
+    pub use redis::streams::StreamId;
+    pub use redis::{from_redis_value, ErrorKind, FromRedisValue, RedisError, ToRedisArgs, Value};
+}

redsumer-rs/src/redsumer/client.rs

@@ -0,0 +1,65 @@
+use redis::Client;
+
+use super::types::RedsumerResult;
+
+/// To hold credentials to authenticate in *Redis*.
+pub struct ClientCredentials<'k> {
+    user: &'k str,
+    password: &'k str,
+}
+
+impl<'k> ClientCredentials<'k> {
+    /// Get *user*
+    fn get_user(&self) -> &str {
+        self.user
+    }
+
+    /// Get *password*
+    fn get_password(&self) -> &str {
+        self.password
+    }
+
+    /// Build a new instance of [`ClientCredentials`].
+    /// # Arguments:
+    /// - **user**: Redis user.
+    /// - **password**: Redis password.
+    ///
+    /// ```rust,no_run
+    /// use redsumer::ClientCredentials;
+    /// let credentials = ClientCredentials::new("user", "password");
+    /// ```
+    pub fn new(user: &'k str, password: &'k str) -> ClientCredentials<'k> {
+        ClientCredentials { user, password }
+    }
+}
+
+/// Get a new [`Client`] instance to connect to *Redis* using a connection URL in format:
+/// `redis://[<username>][:<password>@]<hostname>:<port>/<db>`
+///
+/// # Arguments:
+/// - **credentials**: Option to authenticate in *Redis*.
+/// - **host**: Redis host.
+/// - **port**: Redis port.
+/// - **db**: Redis database.
+pub fn get_redis_client(
+    credentials: Option<ClientCredentials>,
+    host: &str,
+    port: &str,
+    db: &str,
+) -> RedsumerResult<Client> {
+    let url: String = match credentials {
+        Some(credentials) => {
+            format!(
+                "redis://{}:{}@{}:{}/{}",
+                credentials.get_user(),
+                credentials.get_password(),
+                host,
+                port,
+                db,
+            )
+        }
+        None => format!("redis://{}:{}/{}", host, port, db,),
+    };
+
+    Ok(Client::open(url)?)
+}