finchers 0.13.5

A combinator library for builidng asynchronous HTTP services
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

//! Components for working with Finchers runtime.

use std::cell::Cell;

use futures::sync::oneshot;
use futures::{Async, Future, Poll};
use tokio_threadpool;

use error::fail;
use error::Error;

// re-exports
#[doc(no_inline)]
pub use futures::sync::oneshot::SpawnHandle;
#[doc(no_inline)]
pub use tokio::executor::DefaultExecutor;
#[doc(no_inline)]
pub use tokio::spawn;
#[doc(no_inline)]
pub use tokio_threadpool::BlockingError;

// ====

#[derive(Debug, Copy, Clone)]
pub(crate) enum RuntimeMode {
    ThreadPool,
    CurrentThread,
}

thread_local!(static MODE: Cell<Option<RuntimeMode>> = Cell::new(None));

pub(crate) fn with_set_runtime_mode<R>(mode: RuntimeMode, f: impl FnOnce() -> R) -> R {
    #[allow(missing_debug_implementations)]
    struct SetOnDrop(Option<RuntimeMode>);

    impl Drop for SetOnDrop {
        fn drop(&mut self) {
            MODE.with(|mode| mode.set(self.0));
        }
    }

    let mode = MODE.with(|m| m.replace(Some(mode)));
    let _prev = SetOnDrop(mode);
    match mode {
        Some(..) => panic!("The runtime mode has already set on the current context."),
        None => f(),
    }
}

/// Enter a blocking section of code.
///
/// See also the documentation of tokio-threadpool's [`blocking`] for details.
///
/// [`blocking`]: https://docs.rs/tokio-threadpool/0.1/tokio_threadpool/fn.blocking.html
pub fn blocking<R>(f: impl FnOnce() -> R) -> Poll<R, BlockingError> {
    match MODE.with(|mode| mode.get()) {
        Some(RuntimeMode::ThreadPool) | None => tokio_threadpool::blocking(f),
        Some(RuntimeMode::CurrentThread) => Ok(Async::Ready(f())),
    }
}

/// A helper function to create a future from a function which represents a blocking section.
///
/// # Example
///
/// ```
/// # #[macro_use]
/// # extern crate finchers;
/// # extern crate failure;
/// # use finchers::prelude::*;
/// # use finchers::rt::blocking_section;
/// fn get_post_sync(id: u32) -> failure::Fallible<String> {
///     // ...
/// #    drop(id);
/// #    Ok("".into())
/// }
///
/// # fn main() {
/// let endpoint = path!(@get / u32 /)
///     .and_then(|id: u32| {
///         blocking_section(move || get_post_sync(id))
///     });
/// # drop(endpoint);
/// # }
/// ```
pub fn blocking_section<F, T, E>(f: F) -> BlockingSection<F>
where
    F: FnOnce() -> Result<T, E>,
    E: Into<Error>,
{
    BlockingSection { op: Some(f) }
}

/// A `Future` which executes a blocking section with annotation.
#[derive(Debug)]
pub struct BlockingSection<F> {
    op: Option<F>,
}

impl<F, T, E> Future for BlockingSection<F>
where
    F: FnOnce() -> Result<T, E>,
    E: Into<Error>,
{
    type Item = T;
    type Error = Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        let result = try_ready!(
            blocking(|| {
                let op = self.op.take().unwrap();
                op()
            }).map_err(fail)
        );
        result.map(Async::Ready).map_err(Into::into)
    }
}

/// Spawns a future onto the default executor and returns its handle.
#[inline]
pub fn spawn_with_handle<F>(future: F) -> SpawnHandle<F::Item, F::Error>
where
    F: Future + Send + 'static,
    F::Item: Send + 'static,
    F::Error: Send + 'static,
{
    oneshot::spawn(future, &DefaultExecutor::current())
}