shepherd-rs 0.2.0

//! # Consumer Trait
//!
//! This trait defines the behavior of consumers in the shepherd framework.
//!
//! ## Overview
//! - **Consumer**: Processes consumption attempts and manages results.
//! - **Channels**: Uses channels for communication between components.
//!
//! ## Example
//! ```rust
//! struct MyConsumer;
//!
//! impl Consumer for MyConsumer {
//!     // Implementation details...
//! }
//! ```

use std::error::Error;
use std::sync::Arc;

use async_trait::async_trait;
use tokio::sync::Mutex;
use tokio::sync::mpsc::{Receiver, Sender};

use crate::config::Config;
use crate::consumer::attempt::ConsumeAttempt;

/// A consumer that processes consumption attempts.
///
/// The consumer is envisioned as a *Sequential* entity, meaning:
/// 1. That it always receives [`ConsumeAttempt`] in a sequential order.
/// 2. That it always responds back the [`ConsumeAttemptResult`] in the same
///    order.
///
/// *Sequential* means that two sequences `[CA1, CA2]` and `[CA2, CA1]` are not
/// both valid for different underlying `TransformRequest`s. The same argument
/// applies to the results.
///
/// The last seen [`ConsumeAttemptResult`] at the database end dictates the
/// current dynamic state / configuration of the consumer.
#[async_trait]
pub trait Consumer: Send {
    type ConsumeAttempt: ConsumeAttempt;
    type ConsumeError: Send + Error;
    type Config: Config;

    async fn new(
        init_config: Arc<Mutex<Self::Config>>,
        recv_channel: Receiver<Self::ConsumeAttempt>,
        send_channel: Sender<ConsumeAttemptResult<Self::ConsumeAttempt>>,
    ) -> Result<Self, Self::ConsumeError>
    where
        Self: Sized;

    /// The running loop of the consumer.
    /// This loop should run indefinitely, processing incoming consume attempt
    /// requests in a sequential manner. See [`Consumer`] for more details.
    async fn consumer_loop(&mut self) -> Result<(), Self::ConsumeError>;
}

#[derive(Debug, Clone)]
pub enum ConsumeAttemptResult<T: ConsumeAttempt> {
    Success(T::Identifier, T::ReturnCtx),
    Failure(T::Identifier, T::ReturnCtx),
}