noema 0.4.0

//! MediatR-style in-process mediator: `Input` / `InputHandler` / `send()` with optional pipeline.

#[macro_use]
mod macros;

pub mod global;
pub mod pipeline;

pub use global::{GLOBAL, enable_global_pipeline, mediator_config};
pub use noema_macros::input;
pub use pipeline::{Next, PipelineBehavior, run_pipeline};

use std::sync::Arc;

use crate::core::Container;

/// Boxed error type for mediator operations.
pub type BoxDynError = Box<dyn std::error::Error + Send + Sync>;

/// Result type for mediator operations.
pub type NoemaResult<T> = Result<T, BoxDynError>;

/// Request type with associated output.
pub trait Input {
    fn name(&self) -> String {
        std::any::type_name::<Self>().to_string()
    }
    fn description(&self) -> String {
        format!("Input of type {}", self.name())
    }
    type Output;
}

/// Handler for a single request type.
#[async_trait::async_trait]
pub trait InputHandler<T: Input + Send + Sync>: Send + Sync {
    async fn handle(&self, input: Arc<T>) -> NoemaResult<T::Output>;
}

/// Mediator entry point for request type `I` (implemented on `Container` via `request!`).
#[async_trait::async_trait]
pub trait Mediator<I: Input + Send + Sync + 'static> {
    async fn send(input: I) -> NoemaResult<I::Output>;
}

/// Send a request through the pipeline and handler (MediatR `Send` equivalent).
pub async fn send<I: Input + Send + Sync + 'static>(input: I) -> Result<I::Output, BoxDynError>
where
    Container: Mediator<I>,
{
    <Container as Mediator<I>>::send(input).await
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicU32, Ordering};

    use async_trait::async_trait;
    use futures::executor::block_on;

    use crate::__noema_emit_globals;
    use crate::core::*;

    use super::*;

    static ORDER: AtomicU32 = AtomicU32::new(0);

    fn record(step: u32) {
        ORDER.store(step, Ordering::SeqCst);
    }

    #[test]
    fn basic_request_without_pipeline() {
        struct Increaser;
        impl Increaser {
            fn add(&self, value: u32) -> u32 {
                value + 1
            }
        }

        #[input(u32)]
        struct TestInput {
            pub value: u32,
        }

        #[derive(Injectable)]
        struct TestInputHandler {
            increaser: Arc<Increaser>,
        }

        #[async_trait]
        impl InputHandler<TestInput> for TestInputHandler {
            async fn handle(
                &self,
                input: Arc<TestInput>,
            ) -> NoemaResult<<TestInput as Input>::Output> {
                Ok(self.increaser.add(input.value))
            }
        }

        impl Resolver<Increaser> for Container {
            fn resolve() -> Arc<Increaser> {
                Arc::new(Increaser)
            }
        }

        crate::request!(TestInput: TestInputHandler);

        let output = block_on(send(TestInput { value: 41 })).unwrap();
        assert_eq!(output, 42);
    }

    #[test]
    fn per_input_pipeline_order() {
        #[input(u32)]
        struct OrderInput {
            value: u32,
        }

        #[derive(Injectable)]
        struct OrderHandlerInjectable {}

        #[async_trait]
        impl InputHandler<OrderInput> for OrderHandlerInjectable {
            async fn handle(&self, input: Arc<OrderInput>) -> NoemaResult<u32> {
                record(3);
                Ok(input.value)
            }
        }

        struct BehaviorA;
        #[async_trait]
        impl PipelineBehavior<OrderInput> for BehaviorA {
            async fn handle(
                &self,
                input: Arc<OrderInput>,
                next: Next<OrderInput>,
            ) -> NoemaResult<u32> {
                record(1);
                next.run(input).await
            }
        }

        struct BehaviorB;
        #[async_trait]
        impl PipelineBehavior<OrderInput> for BehaviorB {
            async fn handle(
                &self,
                input: Arc<OrderInput>,
                next: Next<OrderInput>,
            ) -> NoemaResult<u32> {
                record(2);
                next.run(input).await
            }
        }

        impl Injectable<Container> for BehaviorA {
            fn inject(_: &Container) -> Self {
                BehaviorA
            }
        }
        impl Injectable<Container> for BehaviorB {
            fn inject(_: &Container) -> Self {
                BehaviorB
            }
        }

        crate::request!(OrderInput: OrderHandlerInjectable => [BehaviorA, BehaviorB]);

        block_on(send(OrderInput { value: 7 })).unwrap();
        assert_eq!(ORDER.load(Ordering::SeqCst), 3);
    }

    #[test]
    fn short_circuit_skips_handler() {
        #[input(u32)]
        struct ShortInput;

        #[derive(Injectable)]
        struct NeverCalledHandlerInjectable {}

        #[async_trait]
        impl InputHandler<ShortInput> for NeverCalledHandlerInjectable {
            async fn handle(&self, _: Arc<ShortInput>) -> NoemaResult<u32> {
                panic!("handler must not run");
            }
        }

        struct ShortCircuit;
        #[async_trait]
        impl PipelineBehavior<ShortInput> for ShortCircuit {
            async fn handle(
                &self,
                _: Arc<ShortInput>,
                _next: Next<ShortInput>,
            ) -> NoemaResult<u32> {
                Ok(99)
            }
        }

        impl Injectable<Container> for ShortCircuit {
            fn inject(_: &Container) -> Self {
                ShortCircuit
            }
        }

        crate::request!(ShortInput: NeverCalledHandlerInjectable => [ShortCircuit]);

        let out = block_on(send(ShortInput)).unwrap();
        assert_eq!(out, 99);
    }
}