WIP async/await implementation

Cargo.toml

@@ -4,6 +4,7 @@ members = [
   "juniper_codegen",
   "juniper",
   "integration_tests/juniper_tests",
+  "integration_tests/async_await",
   "juniper_hyper",
   "juniper_iron",
   "juniper_rocket",

integration_tests/async_await/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "async_await"
+version = "0.1.0"
+authors = ["Christoph Herzog <[email protected]>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+juniper = { path = "../../juniper", features = ["async"] }
+futures-preview = "0.3.0-alpha.18"
+tokio = "0.2.0-alpha.2"

integration_tests/async_await/src/main.rs

@@ -0,0 +1,126 @@
+#![feature(async_await, async_closure)]
+
+use juniper::{graphql_value, RootNode, Value};
+
+#[derive(juniper::GraphQLEnum)]
+enum UserKind {
+    Admin,
+    User,
+    Guest,
+}
+
+struct User {
+    id: u64,
+    name: String,
+    kind: UserKind,
+}
+
+#[juniper::object]
+impl User {
+    async fn name(&self) -> &str {
+        &self.name
+    }
+
+    async fn friends(&self) -> Vec<User> {
+        let friends = (0..10)
+            .map(|index| User {
+                id: index,
+                name: format!("user{}", index),
+                kind: UserKind::User,
+            })
+            .collect();
+        friends
+    }
+
+    async fn kind(&self) -> &UserKind {
+        &self.kind
+    }
+
+    async fn delayed() -> bool {
+        let when = tokio::clock::now() + std::time::Duration::from_millis(100);
+        tokio::timer::Delay::new(when).await;
+        true
+    }
+}
+
+struct Query;
+
+#[juniper::object]
+impl Query {
+    fn field_sync(&self) -> &'static str {
+        "field_sync"
+    }
+
+    async fn field_async_plain() -> String {
+        "field_async_plain".to_string()
+    }
+
+    fn user(id: String) -> User {
+        User {
+            id: 1,
+            name: id,
+            kind: UserKind::User,
+        }
+    }
+
+    async fn delayed() -> bool {
+        let when = tokio::clock::now() + std::time::Duration::from_millis(100);
+        tokio::timer::Delay::new(when).await;
+        true
+    }
+}
+
+struct Mutation;
+
+#[juniper::object]
+impl Mutation {}
+
+fn run<O>(f: impl std::future::Future<Output = O>) -> O {
+    tokio::runtime::current_thread::Runtime::new()
+        .unwrap()
+        .block_on(f)
+}
+
+#[test]
+fn async_simple() {
+    let schema = RootNode::new(Query, Mutation);
+    let doc = r#"
+        query { 
+            fieldSync
+            fieldAsyncPlain 
+            delayed  
+            user(id: "user1") {
+                kind
+                name
+                delayed
+            }
+        }
+    "#;
+
+    let vars = Default::default();
+    let f = juniper::execute_async(doc, None, &schema, &vars, &());
+
+    let (res, errs) = run(f).unwrap();
+
+    assert!(errs.is_empty());
+
+    let mut obj = res.into_object().unwrap();
+    obj.sort_by_field();
+    let value = Value::Object(obj);
+
+    assert_eq!(
+        value,
+        graphql_value!({
+            "delayed": true,
+            "fieldAsyncPlain": "field_async_plain",
+            "fieldSync": "field_sync",
+            "user": {
+                "delayed": true,
+                "kind": "USER",
+                "name": "user1",
+            },
+        }),
+    );
+}
+
+fn main() {}

juniper/Cargo.toml

@@ -24,6 +24,7 @@ harness = false
 path = "benches/bench.rs"
 
 [features]
+async = ["juniper_codegen/async", "futures-preview"]
 expose-test-schema = []
 default = [
     "chrono",
@@ -44,6 +45,9 @@ serde_json = { version="1.0.2", optional = true }
 url = { version = "2", optional = true }
 uuid = { version = "0.7", optional = true }
 
+futures-preview = { version = "0.3.0-alpha.18", optional = true, features = ["nightly", "async-await"] }
+
 [dev-dependencies]
 bencher = "0.1.2"
 serde_json = { version = "1.0.2" }
+tokio = "0.2.0-alpha.2"

juniper/src/executor/mod.rs

@@ -372,6 +372,37 @@ where
         Ok(value.resolve(info, self.current_selection_set, self))
     }
 
+    /// Resolve a single arbitrary value into an `ExecutionResult`
+    #[cfg(feature = "async")]
+    pub async fn resolve_async<T>(&self, info: &T::TypeInfo, value: &T) -> ExecutionResult<S>
+    where
+        T: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+        T::TypeInfo: Send + Sync,
+        CtxT: Send + Sync,
+        S: Send + Sync,
+    {
+        Ok(value
+            .resolve_async(info, self.current_selection_set, self)
+            .await)
+    }
+
+    /// Resolve a single arbitrary value, mapping the context to a new type
+    #[cfg(feature = "async")]
+    pub async fn resolve_with_ctx_async<NewCtxT, T>(
+        &self,
+        info: &T::TypeInfo,
+        value: &T,
+    ) -> ExecutionResult<S>
+    where
+        T: crate::GraphQLTypeAsync<S, Context = NewCtxT> + Send + Sync,
+        T::TypeInfo: Send + Sync,
+        S: Send + Sync,
+        NewCtxT: FromContext<CtxT> + Send + Sync,
+    {
+        let e = self.replaced_context(<NewCtxT as FromContext<CtxT>>::from(self.context));
+        e.resolve_async(info, value).await
+    }
+
     /// Resolve a single arbitrary value into a return value
     ///
     /// If the field fails to resolve, `null` will be returned.
@@ -388,6 +419,26 @@ where
         }
     }
 
+    /// Resolve a single arbitrary value into a return value
+    ///
+    /// If the field fails to resolve, `null` will be returned.
+    #[cfg(feature = "async")]
+    pub async fn resolve_into_value_async<T>(&self, info: &T::TypeInfo, value: &T) -> Value<S>
+    where
+        T: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+        T::TypeInfo: Send + Sync,
+        CtxT: Send + Sync,
+        S: Send + Sync,
+    {
+        match self.resolve_async(info, value).await {
+            Ok(v) => v,
+            Err(e) => {
+                self.push_error(e);
+                Value::null()
+            }
+        }
+    }
+
     /// Derive a new executor by replacing the context
     ///
     /// This can be used to connect different types, e.g. from different Rust
@@ -480,7 +531,7 @@ where
     }
 
     #[doc(hidden)]
-    pub fn fragment_by_name(&self, name: &str) -> Option<&'a Fragment<S>> {
+    pub fn fragment_by_name(&'a self, name: &str) -> Option<&'a Fragment<'a, S>> {
         self.fragments.get(name).cloned()
     }
 
@@ -720,6 +771,121 @@ where
     Ok((value, errors))
 }
 
+#[cfg(feature = "async")]
+pub async fn execute_validated_query_async<'a, QueryT, MutationT, CtxT, S>(
+    document: Document<'a, S>,
+    operation_name: Option<&str>,
+    root_node: &RootNode<'a, QueryT, MutationT, S>,
+    variables: &Variables<S>,
+    context: &CtxT,
+) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+where
+    S: ScalarValue + Send + Sync,
+    QueryT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    MutationT::TypeInfo: Send + Sync,
+    CtxT: Send + Sync,
+    for<'b> &'b S: ScalarRefValue<'b>,
+{
+    let mut fragments = vec![];
+    let mut operation = None;
+
+    for def in document {
+        match def {
+            Definition::Operation(op) => {
+                if operation_name.is_none() && operation.is_some() {
+                    return Err(GraphQLError::MultipleOperationsProvided);
+                }
+
+                let move_op = operation_name.is_none()
+                    || op.item.name.as_ref().map(|s| s.item) == operation_name;
+
+                if move_op {
+                    operation = Some(op);
+                }
+            }
+            Definition::Fragment(f) => fragments.push(f),
+        };
+    }
+
+    let op = match operation {
+        Some(op) => op,
+        None => return Err(GraphQLError::UnknownOperationName),
+    };
+
+    let default_variable_values = op.item.variable_definitions.map(|defs| {
+        defs.item
+            .items
+            .iter()
+            .filter_map(|&(ref name, ref def)| {
+                def.default_value
+                    .as_ref()
+                    .map(|i| (name.item.to_owned(), i.item.clone()))
+            })
+            .collect::<HashMap<String, InputValue<S>>>()
+    });
+
+    let errors = RwLock::new(Vec::new());
+    let value;
+
+    {
+        let mut all_vars;
+        let mut final_vars = variables;
+
+        if let Some(defaults) = default_variable_values {
+            all_vars = variables.clone();
+
+            for (name, value) in defaults {
+                all_vars.entry(name).or_insert(value);
+            }
+
+            final_vars = &all_vars;
+        }
+
+        let root_type = match op.item.operation_type {
+            OperationType::Query => root_node.schema.query_type(),
+            OperationType::Mutation => root_node
+                .schema
+                .mutation_type()
+                .expect("No mutation type found"),
+        };
+
+        let executor = Executor {
+            fragments: &fragments
+                .iter()
+                .map(|f| (f.item.name.item, &f.item))
+                .collect(),
+            variables: final_vars,
+            current_selection_set: Some(&op.item.selection_set[..]),
+            parent_selection_set: None,
+            current_type: root_type,
+            schema: &root_node.schema,
+            context,
+            errors: &errors,
+            field_path: FieldPath::Root(op.start),
+        };
+
+        value = match op.item.operation_type {
+            OperationType::Query => {
+                executor
+                    .resolve_into_value_async(&root_node.query_info, &root_node)
+                    .await
+            }
+            OperationType::Mutation => {
+                executor
+                    .resolve_into_value_async(&root_node.mutation_info, &root_node.mutation_type)
+                    .await
+            }
+        };
+    }
+
+    let mut errors = errors.into_inner().unwrap();
+    errors.sort();
+
+    Ok((value, errors))
+}
+
 impl<'r, S> Registry<'r, S>
 where
     S: ScalarValue + 'r,