# msgq : Robust Redis Stream Message Queue

[中文](zh.md)

Robust Redis Stream based message queue with auto-claim and retry handling.

## Features

- **Redis Stream Based**: Utilizes `XREADGROUP` for efficient message consumption.
- **Consumer Groups**: Supports scalable message processing with consumer groups.
- **Auto Group Creation**: Automatically creates consumer groups if they don't exist.
- **Reliability**: Built-in `BLOCK` and `CLAIM` mechanisms for handling idle messages.
- **Concurrency**: Processes messages concurrently using `tokio::spawn`.
- **Error Handling**: Automatic retry mechanism (up to 3 times) with error callbacks.

## Usage

```rust
use std::sync::Arc;
use aok::{OK, Void};
use log::info;
use msgq::{Kv, ReadGroup};

async fn on_error(mail: Arc<Kv>, error: String) -> Void {
  info!("on_error {error} {:?}", mail);
  OK
}

async fn print_mail(mail: Arc<Kv>) -> Void {
  info!("print_mail {:?}", mail);
  OK
}

#[tokio::main]
async fn main() -> Void {
  // Initialize Redis connection (example)
  // xboot::init().await?;

  // Create a ReadGroup
  // stream: "smtp", group: "send", consumer: "test"
  // block: 1s, claim_idle: 3s, count: 2
  let read_group = ReadGroup::new("smtp", "send", "test", 1, 3, 2);

  // Run the consumer loop
  read_group.run(print_mail, on_error).await?;
  OK
}
```

## Design

The `ReadGroup::run` method executes a continuous loop that:

1.  Constructs an `XREADGROUP` command with `BLOCK` and `CLAIM` options.
2.  Fetches messages from the specified Redis Stream.
3.  If the consumer group is missing, it calls `auto_new` to create it.
4.  Spawns asynchronous tasks to process each message using the provided callback.
5.  Monitors task execution:
    -   If a task fails, it logs the error.
    -   If retries exceed 3, the message is moved to the error handler.
    -   Failed messages are acknowledged and deleted (`XACK` + `XDEL`) via `rm_err_id_li`.

## Tech Stack

-   **Rust**: Core language.
-   **Tokio**: Asynchronous runtime.
-   **Fred**: Redis client.
-   **xkv**: Redis connection management.

## Directory Structure

-   `src/lib.rs`: Library exports.
-   `src/read_group.rs`: Core consumer logic.
-   `src/auto_new.rs`: Consumer group management.
-   `src/rm_err_id_li.rs`: Error cleanup logic.
-   `tests/`: Integration tests and examples.

## API Reference

### `ReadGroup`

Main struct for configuring and running the consumer.

-   `new(stream, group, consumer, block_sec, claim_idle_sec, count)`: Creates a new instance.
-   `run(parse, on_error)`: Starts the processing loop.

### `StreamItem`

Represents a message from the stream.

-   `id`: Message ID.
-   `kv`: Key-Value pairs (payload).

## Trivia

Redis Streams, introduced in Redis 5.0, brought a log data structure to Redis, inspired by Kafka. It allows for consumer groups, enabling distributed message processing where multiple consumers can read from the same stream without processing the same message twice. `XREADGROUP` is the primary command for this pattern, facilitating robust and scalable message queue architectures directly within Redis.