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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

use std::sync::Arc;
use std::time::Duration;

use derive_builder::Builder;
use parking_lot::Mutex;
use tarantool::fiber;

use crate::error::Error;
use crate::error::Result;

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

/// `PollTransport` is most widely supported and less efficient transport available.
/// Unlike [CBus][super::cbus::CBusTransport], it can be used in every tarantool version.
///
/// To determine, when result is ready,
/// it polls underlying channel with the configurable interval.
#[derive(Builder, Debug)]
pub struct PollTransport {
    #[builder(default = "default_poll_interval()")]
    interval: Duration,
}

impl Default for PollTransport {
    fn default() -> Self {
        Self {
            interval: default_poll_interval(),
        }
    }
}

fn default_poll_interval() -> Duration {
    Duration::from_millis(10)
}

impl Transport for PollTransport {
    type Sender<T> = Sender<T> where T: Send;

    type Receiver<T> = Receiver<T>
    where
        T: Send;

    fn create_channel<T: Send>(&self) -> (Self::Sender<T>, Self::Receiver<T>) {
        let result: SharedResult<T> = Default::default();

        (
            Sender(result.clone()),
            Receiver {
                result,
                interval: self.interval,
            },
        )
    }
}

pub struct Sender<T>(SharedResult<T>);

impl<T: Send> TransportSender<T> for Sender<T> {
    fn send(self, data: T) -> Result<()> {
        *self.0.lock() = Some(data);
        Ok(())
    }
}

pub struct Receiver<T> {
    result: SharedResult<T>,
    interval: Duration,
}

impl<T: Send> TransportReceiver<T> for Receiver<T> {
    fn receive(self) -> Result<T> {
        loop {
            {
                let mut result = self.result.lock();
                if let Some(result) = result.take() {
                    return Ok(result);
                }

                // Check that sender part is still online via ref counter.
                //
                // It is done atomically with the lock above,
                // because sender part can successfully exit the function between lock and this check.
                if Arc::strong_count(&self.result) != 2 {
                    return Err(Error::sender_disconnected());
                }
            }
            fiber::sleep(self.interval)
        }
    }
}

type SharedResult<T> = Arc<Mutex<Option<T>>>;