sena 0.2.0

Library for composable event handlers
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

use crate::csp::{
    comm::{Responder, RxChan, VoidTx},
    message::Message,
    shutdown::NoShutdown,
};

use super::handler::Handler;

with_rt! {
    use either::Either;
    use crate::{
        csp::{
            comm::OutputTx,
            shutdown::ShutdownRx,
        },
        utils::poll_biased::PollBiased,
    };
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Server<H, X> {
    pub handler: H,
    pub rx: X,
}

pub struct ServeOptions<OTx, SRx> {
    /// Token that can be used to gracefully shutdown server
    pub shutdown_rx: SRx,

    /// Channel where all output will be thrown, outputs will be sent here
    /// only if output is not forwarded as a response (client doesn't wants response)
    pub output_tx: OTx,
}

impl<SRx> ServeOptions<VoidTx, SRx> {
    pub const fn with_shutdown_token(shutdown_rx: SRx) -> Self {
        Self {
            shutdown_rx,
            output_tx: VoidTx,
        }
    }
}

impl Default for ServeOptions<VoidTx, NoShutdown> {
    fn default() -> Self {
        Self {
            shutdown_rx: NoShutdown,
            output_tx: VoidTx,
        }
    }
}

impl<H, X> Server<H, X> {
    #[cfg(feature = "tokio")]
    pub async fn serve<T, E, OTx, SRx>(
        &mut self,
        mut options: ServeOptions<OTx, SRx>,
    ) -> Result<(), E>
    where
        T: Send + 'static,
        E: Send + 'static,
        OTx: OutputTx<H::Output, E> + 'static,
        H: Handler<T, E> + Clone + 'static,
        X: RxChan<T, E, H::Output>,
        X::Responder: Send + 'static,
        SRx: ShutdownRx<E>,
    {
        loop {
            let result = PollBiased::new(options.shutdown_rx.wait_shutdown(), self.rx.recv()).await;
            match result {
                Either::Left(_) => return Ok(()),
                Either::Right(message) => {
                    let message = message?;

                    let out_tx = options.output_tx.clone();
                    let handler = self.handler.clone();

                    tokio::spawn(async move {
                        let output = handler.handle(message.data).await;
                        if let Some(responder) = message.responder {
                            responder.respond_with(output).await?;
                        } else {
                            out_tx.send(output).await?;
                        }

                        Ok::<_, E>(())
                    });
                }
            }
        }
    }

    pub async fn handle_message<T, E>(
        &mut self,
        message: Message<T, X::Responder>,
    ) -> Result<Option<Result<H::Output, E>>, E>
    where
        H: Handler<T, E>,
        X: RxChan<T, E, H::Output>,
    {
        let output = self.handler.handle(message.data).await;

        if let Some(responder) = message.responder {
            responder.respond_with(output).await?;
            Ok(None)
        } else {
            Ok(Some(output))
        }
    }
}