tros 0.6.0

Your tros from tarantool TX thread to tokio(..and others!)
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

use std::sync::Once;

use tarantool::{
    cbus::{self, oneshot, RecvError},
    fiber,
};

use crate::error::{Error, Result};

use super::{Transport, TransportReceiver, TransportSender};

const DEFAULT_CBUS_ENDPONT: &str = "tros_cbus_endpoint";
static DEFAULT_LOOP_FIBER_INIT: Once = Once::new();

/// Data is being sent via cbus-channel.
/// Most efficient implementation of the transport,
/// as it puts current fiber to sleep without further manual polling.
#[derive(Debug)]
pub struct CBusTransport<'a> {
    cbus_endpoint: &'a str,
}

impl<'a> Default for CBusTransport<'a> {
    fn default() -> Self {
        DEFAULT_LOOP_FIBER_INIT.call_once(|| {
            fiber::Builder::new()
                .name("tros_cbus_loop_f")
                .func(move || {
                    let cbus_endpoint = cbus::Endpoint::new(DEFAULT_CBUS_ENDPONT)
                        .expect("error on start cbus endpoint");
                    cbus_endpoint.cbus_loop();
                })
                .start_non_joinable()
                .unwrap();
        });
        Self::new_unchecked(DEFAULT_CBUS_ENDPONT)
    }
}

impl<'a> CBusTransport<'a> {
    /// Initialize `CBusTransport` with given custom endpoint.
    /// Take a note that it is **your** responsibility to initialize
    /// given endpoint(and its loop fiber) - hence the name "unchecked".
    ///
    /// If you want automatic endpoint init, prefer `Default::default` instead -
    /// - it would take care of all heavy lifting.
    pub fn new_unchecked(cbus_endpoint: &'a str) -> Self {
        Self { cbus_endpoint }
    }
}

impl<'a> Transport for CBusTransport<'a> {
    type Sender<T> = oneshot::Sender<T> where T: Send;
    type Receiver<T> = oneshot::EndpointReceiver<T> where T: Send;

    fn create_channel<T: Send>(&self) -> (Self::Sender<T>, Self::Receiver<T>) {
        oneshot::channel(self.cbus_endpoint)
    }
}

impl<T> TransportSender<T> for oneshot::Sender<T> {
    fn send(self, data: T) -> Result<()> {
        oneshot::Sender::send(self, data);
        Ok(())
    }
}

impl<T> TransportReceiver<T> for oneshot::EndpointReceiver<T> {
    fn receive(self) -> Result<T> {
        oneshot::EndpointReceiver::receive(self).map_err(|err| match err {
            RecvError::Disconnected => Error::sender_disconnected(),
        })
    }
}