idemio 0.1.1

Core framework for idemio products
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use crate::exchange::{Exchange, ExchangeError};
use crate::router::path::LoadedChain;
use crate::status::ExchangeState;
use async_trait::async_trait;
use std::sync::Arc;
use thiserror::Error;
use crate::handler::Handler;

/// A trait for executing handler chains in an asynchronous request/response processing system.
///
/// This trait defines the contract for executing a series of handlers on an exchange,
/// providing a standardized way to process requests through various middleware and
/// termination handlers.
#[async_trait]
pub trait HandlerExecutor<Exchange>
where
    Exchange: Send + Sync,
{
    /// The output type that this executor returns after processing
    type Output: Send + Sync;

    // TODO - look into making return status generic. Different runtimes might require different status returns.
    async fn execute_handler(handlers: &Vec<Arc<dyn Handler<Exchange>>>, exchange: &mut Exchange) -> Result<bool, ExecutorError>
    where
        Exchange: ExtractOutput<Self::Output> + Send + Sync,
    {
        for handler in handlers {
            let status = handler.exec(exchange).await.unwrap();
            if status.code.is_in_flight() {
                continue;
            } else if status.code.is_completed() || status.code.is_error() {
                return Ok(true);
            } else {
                return Err(ExecutorError::unknown_exchange_state(status.code));
            }
        }
        Ok(false)
    }

    async fn execute_handlers(
        &self,
        executables: Arc<LoadedChain<Exchange>>,
        exchange: &mut Exchange,
    ) -> Result<Self::Output, ExecutorError>;

    async fn return_output(exchange: &mut Exchange) -> Result<Self::Output, ExecutorError>;
}


#[derive(Error, Debug)]
pub enum ExecutorError {

    #[error("Exchange is in an unknown state: {state}")]
    State { state: ExchangeState },

    #[error("Failed to extract output from exchange.")]
    Output {
        #[source]
        source: ExchangeError,
    },
}

impl ExecutorError {

    pub fn output_read_error(err: ExchangeError) -> Self {
        ExecutorError::Output { source: err }
    }

    pub fn unknown_exchange_state(state: ExchangeState) -> Self {
        ExecutorError::State { state }
    }
}

pub struct DefaultExecutor<Output> {
    pub _phantom: std::marker::PhantomData<Output>,
}

#[async_trait]
impl<E, Output> HandlerExecutor<E> for DefaultExecutor<Output>
where
    E: ExtractOutput<Output> + Send + Sync,
    Output: Send + Sync,

{
    type Output = Output;

    async fn execute_handlers(
        &self,
        executables: Arc<LoadedChain<E>>,
        exchange: &mut E,
    ) -> Result<Self::Output, ExecutorError> {
        let request_handlers = executables.request_handlers();
        let request_handlers_result = Self::execute_handler(&request_handlers, exchange).await?;

        // Early exit
        if request_handlers_result {
            return Ok(Self::return_output(exchange).await?);
        }

        let termination_handler = executables.termination_handler();
        let status = termination_handler.exec(exchange).await.unwrap();
        if status.code.is_completed() || status.code.is_error() {
            return Ok(Self::return_output(exchange).await?);
        } else if !status.code.is_in_flight() {
            return Err(ExecutorError::unknown_exchange_state(status.code));
        }

        let response_handlers = executables.response_handlers();
        Self::execute_handler(&response_handlers, exchange).await?;
        exchange
            .extract_output()
            .await
            .map_err(|e| ExecutorError::output_read_error(e))
    }

    async fn return_output(exchange: &mut E) -> Result<Self::Output, ExecutorError> {
        exchange.extract_output().await.map_err(|e| ExecutorError::output_read_error(e))
    }
}

/// Helper trait to extract output from an exchange
/// This abstracts the output extraction logic and makes it testable
#[async_trait]
pub trait ExtractOutput<Output> {
    async fn extract_output(&mut self) -> Result<Output, ExchangeError>;
}

#[async_trait]
impl<In, Out, Meta> ExtractOutput<Out> for Exchange<In, Out, Meta>
where
    In: Send + Sync,
    Out: Send + Sync,
    Meta: Send + Sync,
{
    async fn extract_output(&mut self) -> Result<Out, ExchangeError> {
        self.take_output().await
    }
}