ioserverlib 0.1.2

IO messages serving library
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

use std::io::{Read, Write};
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use crate::serializer::Serializer;
use crate::channel::OwnedChannel;

/// Spawn new thread with a new `Server` struct which will listen to incoming
/// messages in a loop, process them using the `messages_handler` and use
/// `errors_handler` on any occuring errors.
///
/// If `errors_handler` returns `true`, then the thread will be closed.
pub fn daemon<R, W, S, C, H, E>(
    channel: C,
    messages_handler: H,
    errors_handler: E
) -> Daemon
where
    R: Read + Send + 'static,
    W: Write + Send + 'static,
    S: Serializer<R, W> + Send + 'static,
    C: OwnedChannel<R, W, S> + Send + 'static,
    H: Fn(S::Message) -> Option<S::Message> + Send + 'static,
    E: Fn(S::Error) -> bool + Send + 'static
{
    let mut server = Server::new(channel, messages_handler);

    let alive = Arc::new(AtomicBool::new(true));

    {
        let alive = alive.clone();

        std::thread::spawn(move || {
            while alive.load(Ordering::Relaxed) {
                if let Err(err) = server.update() {
                    if (errors_handler)(err) {
                        alive.store(false, Ordering::Release);

                        break;
                    }
                }
            }
        });
    }

    Daemon(alive)
}

pub struct Server<R, W, S, C, H> {
    _reader: PhantomData<R>,
    _writer: PhantomData<W>,
    _serializer: PhantomData<S>,
    channel: C,
    handler: H
}

impl<R, W, S, C, H> Server<R, W, S, C, H>
where
    R: Read,
    W: Write,
    S: Serializer<R, W>,
    C: OwnedChannel<R, W, S>,
    H: Fn(S::Message) -> Option<S::Message>
{
    #[inline(always)]
    pub const fn new(channel: C, handler: H) -> Self {
        Self {
            _reader: PhantomData,
            _writer: PhantomData,
            _serializer: PhantomData,
            channel,
            handler
        }
    }

    #[inline(always)]
    pub fn channel(&mut self) -> &mut C {
        &mut self.channel
    }

    pub fn update(&mut self) -> Result<(), S::Error> {
        let message = self.channel.read()?;

        if let Some(response) = (self.handler)(message) {
            self.channel.write(response)?;
        }

        Ok(())
    }
}

#[derive(Debug, Clone)]
pub struct Daemon(Arc<AtomicBool>);

impl Daemon {
    /// Check if the underlying server thread is still running.
    #[inline]
    pub fn is_alive(&self) -> bool {
        self.0.load(Ordering::Relaxed)
    }

    /// Kill the underlying server thread.
    #[inline]
    pub fn kill(self) {
        self.0.store(false, Ordering::Release);
    }
}