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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

#![doc = include_str!("../README.md")]
use std::{future::Future, time::Duration};

use async_scoped::spawner::{Blocker, FuncSpawner, Spawner};
use async_scoped::Scope;
use derive_builder::Builder;
use error::Result;
use once_cell::sync::Lazy;

use tokio::runtime::{Builder as TokioBuilder, Handle, Runtime};
use tokio::task as tokio_task;
use transport::{Transport, TransportReceiver, TransportSender};

pub mod error;
pub mod transport;

static DEFAULT_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
    let runtime = TokioBuilder::new_multi_thread()
        .worker_threads(4)
        .thread_keep_alive(Duration::from_secs(60))
        .enable_time()
        .enable_io()
        .build()
        .unwrap();
    runtime
});

fn default_rt_handle() -> Handle {
    DEFAULT_RUNTIME.handle().clone()
}

#[derive(Clone, Builder)]
#[builder(pattern = "owned")]
pub struct TokioExecutor<TR> {
    #[builder(default = "default_rt_handle()")]
    rt_handle: Handle,
    transport: TR,
}

impl<TR> TokioExecutor<TR> {
    pub fn new(transport: TR) -> Self {
        Self {
            transport,
            rt_handle: default_rt_handle(),
        }
    }
}

impl<TR: Default> Default for TokioExecutor<TR> {
    fn default() -> Self {
        Self {
            rt_handle: default_rt_handle(),
            transport: TR::default(),
        }
    }
}

impl<TR: Transport> TokioExecutor<TR> {
    /// Drive given future to completion on the configured tokio runtime.
    /// It might yield the current fiber, but might as well block it.
    /// It depends on the transport you use.
    pub fn exec<T: Send, F: Future<Output = T> + Send>(&self, f: F) -> Result<T> {
        let (tx, rx) = self.transport.create_channel();

        let mut scope = unsafe { Scope::create(self.spawner()) };
        let future = async move {
            tx.send(f.await)
                .expect("Failed to send future result through transport sender")
        };
        scope.spawn(future);
        rx.receive()
    }

    fn spawner(&self) -> TokioSpawner {
        TokioSpawner(self.rt_handle.clone())
    }
}

/// Spawner for tokio, needed only for `async-scoped` traits implementation.
///
/// There is an ongoing effort to push stateful tokio spawner impl to async-scoped.
/// See [PR](https://github.com/rmanoka/async-scoped/pull/18) for more info.
///
/// TODO: If `PR` is accepted, this should be easily removed.
struct TokioSpawner(Handle);

unsafe impl<T: Send + 'static> Spawner<T> for TokioSpawner {
    type FutureOutput = std::result::Result<T, tokio_task::JoinError>;
    type SpawnHandle = tokio_task::JoinHandle<T>;

    fn spawn<F: Future<Output = T> + Send + 'static>(&self, f: F) -> Self::SpawnHandle {
        self.0.spawn(f)
    }
}

unsafe impl<T: Send + 'static> FuncSpawner<T> for TokioSpawner {
    type FutureOutput = std::result::Result<T, tokio_task::JoinError>;
    type SpawnHandle = tokio_task::JoinHandle<T>;

    fn spawn_func<F: FnOnce() -> T + Send + 'static>(&self, f: F) -> Self::SpawnHandle {
        self.0.spawn_blocking(f)
    }
}

unsafe impl Blocker for TokioSpawner {
    fn block_on<T, F: Future<Output = T>>(&self, f: F) -> T {
        self.0.block_on(f)
    }
}