finchers_juniper/execute/
nonblocking.rs

1use finchers::endpoint;
2use finchers::endpoint::wrapper::Wrapper;
3use finchers::endpoint::{ApplyContext, ApplyResult, Endpoint, IntoEndpoint};
4use finchers::error::Error;
5use finchers::rt;
6
7use futures::future;
8use futures::{Future, Poll};
9use std::sync::Arc;
10
11use super::shared::SharedSchema;
12use request::{GraphQLRequestEndpoint, GraphQLResponse, RequestFuture};
13
14/// Create a GraphQL executor from the specified `RootNode`.
15///
16/// The endpoint created by this wrapper will spawn a task which executes the GraphQL queries
17/// after receiving the request, by using tokio's `DefaultExecutor`, and notify the start of
18/// the blocking section by using tokio_threadpool's blocking API.
19pub fn nonblocking<S>(schema: S) -> Nonblocking<S>
20where
21    S: SharedSchema,
22{
23    Nonblocking { schema }
24}
25
26#[allow(missing_docs)]
27#[derive(Debug)]
28pub struct Nonblocking<S> {
29    schema: S,
30}
31
32impl<'a, S> IntoEndpoint<'a> for Nonblocking<S>
33where
34    S: SharedSchema<Context = ()>,
35{
36    type Output = (GraphQLResponse,);
37    type Endpoint = NonblockingEndpoint<endpoint::Cloned<()>, S>;
38
39    fn into_endpoint(self) -> Self::Endpoint {
40        NonblockingEndpoint {
41            context: endpoint::cloned(()),
42            request: ::request::graphql_request(),
43            schema: Arc::new(self.schema),
44        }
45    }
46}
47
48impl<'a, E, S> Wrapper<'a, E> for Nonblocking<S>
49where
50    E: Endpoint<'a, Output = (S::Context,)>,
51    S: SharedSchema,
52{
53    type Output = (GraphQLResponse,);
54    type Endpoint = NonblockingEndpoint<E, S>;
55
56    fn wrap(self, endpoint: E) -> Self::Endpoint {
57        NonblockingEndpoint {
58            context: endpoint,
59            request: ::request::graphql_request(),
60            schema: Arc::new(self.schema),
61        }
62    }
63}
64
65#[derive(Debug)]
66pub struct NonblockingEndpoint<E, S> {
67    context: E,
68    request: GraphQLRequestEndpoint,
69    schema: Arc<S>,
70}
71
72impl<'a, E, S> Endpoint<'a> for NonblockingEndpoint<E, S>
73where
74    E: Endpoint<'a, Output = (S::Context,)>,
75    S: SharedSchema,
76{
77    type Output = (GraphQLResponse,);
78    type Future = NonblockingFuture<'a, E, S>;
79
80    fn apply(&'a self, cx: &mut ApplyContext<'_>) -> ApplyResult<Self::Future> {
81        let context = self.context.apply(cx)?;
82        let request = self.request.apply(cx)?;
83        Ok(NonblockingFuture {
84            inner: context.join(request),
85            handle: None,
86            endpoint: self,
87        })
88    }
89}
90
91#[allow(missing_debug_implementations)]
92pub struct NonblockingFuture<'a, E: Endpoint<'a>, S: 'a> {
93    inner: future::Join<E::Future, RequestFuture<'a>>,
94    handle: Option<rt::SpawnHandle<GraphQLResponse, Error>>,
95    endpoint: &'a NonblockingEndpoint<E, S>,
96}
97
98impl<'a, E, S> Future for NonblockingFuture<'a, E, S>
99where
100    E: Endpoint<'a, Output = (S::Context,)>,
101    S: SharedSchema,
102{
103    type Item = (GraphQLResponse,);
104    type Error = Error;
105
106    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
107        loop {
108            match self.handle {
109                Some(ref mut handle) => return handle.poll().map(|x| x.map(|response| (response,))),
110                None => {
111                    let ((context,), (request,)) = try_ready!(self.inner.poll());
112
113                    trace!("spawn a GraphQL task using the default executor");
114                    let schema = self.endpoint.schema.clone();
115                    let future = rt::blocking_section(move || -> ::finchers::error::Result<_> {
116                        Ok(request.execute(schema.as_root_node(), &context))
117                    });
118                    self.handle = Some(rt::spawn_with_handle(future));
119                }
120            }
121        }
122    }
123}
finchers_juniper/execute/nonblocking.rs

finchers_juniper/execute/
nonblocking.rs
