Make all tests passed (async-graphql).

derive/src/subscription.rs

@@ -290,13 +290,13 @@ pub fn generate(object_args: &args::Object, item_impl: &mut ItemImpl) -> Result<
                                     extensions: None,
                                     cache_control: Default::default(),
                                     error: None,
-                                }
+                                },
                                 Err(err) => err.into(),
                             };
                             if resp.is_err() {
                                 *state = false;
                             }
-                            return #crate_name::futures::future::ready(Some(resp));
+                            #crate_name::futures::future::ready(Some(resp))
                         });
                         return Ok(Box::pin(stream));
                     }

src/http/mod.rs

@@ -10,10 +10,9 @@ pub use multipart::{receive_multipart, MultipartOptions};
 pub use playground_source::{playground_source, GraphQLPlaygroundConfig};
 pub use stream_body::StreamBody;
 
-use crate::{Data, ParseRequestError, Pos, QueryError, Request, Variables};
+use crate::{Data, ParseRequestError, Request, Variables};
 use futures::io::AsyncRead;
 use futures::AsyncReadExt;
-use serde::ser::{SerializeMap, SerializeSeq};
 use serde::Deserialize;
 
 /// Deserializable GraphQL Request object
@@ -47,7 +46,7 @@ impl From<GQLRequest> for Request {
 /// Receive a GraphQL request from a content type and body.
 pub async fn receive_body(
     content_type: Option<impl AsRef<str>>,
-    mut body: impl AsyncRead + Send + Unpin + 'static,
+    mut body: impl AsyncRead + Unpin + Send + 'static,
     opts: MultipartOptions,
 ) -> Result<Request, ParseRequestError> {
     if let Some(Ok(boundary)) = content_type.map(multer::parse_boundary) {

src/http/multipart.rs

@@ -22,7 +22,7 @@ pub struct MultipartOptions {
 
 /// Receive a multipart request.
 pub async fn receive_multipart(
-    body: impl AsyncRead + Send + Unpin + 'static,
+    body: impl AsyncRead + Unpin + Send + 'static,
     boundary: impl Into<String>,
     opts: MultipartOptions,
 ) -> Result<Request, ParseRequestError> {
@@ -104,7 +104,7 @@ pub async fn receive_multipart(
 }
 
 fn reader_stream(
-    mut reader: impl AsyncRead + Send + Unpin + 'static,
+    mut reader: impl AsyncRead + Unpin + Send + 'static,
 ) -> impl Stream<Item = io::Result<Bytes>> + Unpin + Send + 'static {
     let mut buf = [0u8; 2048];
 

src/lib.rs

@@ -153,7 +153,7 @@ pub use response::Response;
 pub use scalars::{Any, Json, OutputJson, ID};
 pub use schema::{Schema, SchemaBuilder, SchemaEnv};
 pub use serde_json::Number;
-pub use subscription::{ConnectionTransport, SimpleBroker, SubscriptionStreams};
+pub use subscription::SimpleBroker;
 pub use types::{
     connection, EmptyMutation, EmptySubscription, MaybeUndefined, MergedObject,
     MergedObjectSubscriptionTail, MergedObjectTail, Upload,

src/look_ahead.rs

@@ -101,7 +101,7 @@ mod tests {
 
         let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 1) {
@@ -110,9 +110,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 1) {
@@ -121,9 +121,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 2) {
@@ -134,9 +134,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 3) {
@@ -145,9 +145,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 1) {
@@ -158,9 +158,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 2) {
@@ -173,9 +173,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 1) {
@@ -188,9 +188,9 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
 
-        schema
+        assert!(!schema
             .execute(
                 r#"{
             obj(n: 2) {
@@ -205,6 +205,6 @@ mod tests {
         }"#,
             )
             .await
-            .unwrap();
+            .is_err());
     }
 }

src/request.rs

@@ -1,16 +1,9 @@
 use crate::parser::types::UploadValue;
-use crate::{http, Data, ParseRequestError, Value, Variables};
-use bytes::Bytes;
-use futures::stream;
-use futures::task::Poll;
-use futures::{AsyncRead, AsyncReadExt, Stream};
-use multer::{Constraints, Multipart, SizeLimit};
+use crate::{Data, Value, Variables};
 use std::any::Any;
-use std::collections::HashMap;
 use std::fs::File;
-use std::io::{self, Seek, SeekFrom, Write};
-use std::pin::Pin;
 
+/// GraphQL query request
 pub struct Request {
     pub(crate) query: String,
     pub(crate) operation_name: Option<String>,

src/response.rs

@@ -1,4 +1,4 @@
-use crate::{CacheControl, Error};
+use crate::{CacheControl, Error, Result};
 
 /// Query response
 #[derive(Debug)]
@@ -17,15 +17,73 @@ pub struct Response {
 }
 
 impl Response {
+    /// Returns `true` if the response is ok.
+    #[inline]
+    pub fn is_ok(&self) -> bool {
+        self.error.is_none()
+    }
+
+    /// Returns `true` if the response is error.
     #[inline]
     pub fn is_err(&self) -> bool {
         self.error.is_some()
     }
 
+    /// Get self.
+    ///
+    /// Panics
+    ///
+    /// It will panic when the response is error.
+    #[inline]
+    pub fn unwrap(self) -> Self {
+        self
+    }
+
+    /// Get the error object.
+    ///
+    /// Panics
+    ///
+    /// It will panic when the response is ok.
     #[inline]
     pub fn unwrap_err(self) -> Error {
         self.error.unwrap()
     }
+
+    /// Returns the contained error, consuming the self value.
+    ///
+    /// Panics
+    ///
+    /// Panics if the response is ok, with a panic message including the passed message.
+    #[inline]
+    pub fn expect_err(self, msg: &str) -> Error {
+        match self.error {
+            Some(err) => err,
+            None => panic!("{}", msg),
+        }
+    }
+
+    /// Returns self, consuming the self value.
+    ///
+    /// Panics
+    ///
+    /// Panics if the response is errror, with a panic message including the passed message.
+    #[inline]
+    pub fn expect(self, msg: &str) -> Self {
+        match self.error {
+            Some(_) => panic!("{}", msg),
+            None => self,
+        }
+    }
+
+    /// Convert response to `Result<Response>`.
+    #[inline]
+    pub fn into_result(self) -> Result<Self> {
+        if self.is_err() {
+            Err(self.error.unwrap())
+        } else {
+            Ok(self)
+        }
+    }
 }
 
 impl From<Error> for Response {

src/scalars/json.rs

@@ -128,7 +128,7 @@ mod test {
         let query = r#"{ obj(input: { a: 1, b: 2, c: { a: 11, b: 22 } } ) }"#;
         let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
         assert_eq!(
-            schema.execute(&query).await.unwrap().data,
+            schema.execute(query).await.unwrap().data,
             serde_json::json!({
              "obj": {
                  "a": 1,

src/schema.rs

@@ -358,17 +358,18 @@ where
         Ok((document, cache_control, extensions))
     }
 
-    /// Execute query without create the `QueryBuilder`.
-    pub async fn execute(&self, query: Request) -> Response {
+    /// Execute an GraphQL query.
+    pub async fn execute(&self, request: impl Into<Request>) -> Response {
+        let request = request.into();
         let (document, cache_control, extensions) =
-            try_query_result!(self.prepare_query(&query.query, &query.variables));
+            try_query_result!(self.prepare_query(&request.query, &request.variables));
 
         // execute
         let inc_resolve_id = AtomicUsize::default();
-        let document = match document.into_data(query.operation_name.as_deref()) {
+        let document = match document.into_data(request.operation_name.as_deref()) {
             Some(document) => document,
             None => {
-                let err = if let Some(operation_name) = query.operation_name {
+                let err = if let Some(operation_name) = request.operation_name {
                     Error::Query {
                         pos: Pos::default(),
                         path: None,
@@ -390,9 +391,9 @@ where
 
         let env = QueryEnv::new(
             extensions,
-            query.variables,
+            request.variables,
             document,
-            Arc::new(query.ctx_data),
+            Arc::new(request.ctx_data),
         );
         let ctx = ContextBase {
             path_node: None,
@@ -429,16 +430,18 @@ where
         }
     }
 
+    /// Execute an GraphQL subscription.
     pub async fn execute_stream(
         &self,
-        query: Request,
+        request: impl Into<Request>,
     ) -> Result<impl Stream<Item = Response> + Send> {
-        let (document, _, extensions) = self.prepare_query(&query.query, &query.variables)?;
+        let request = request.into();
+        let (document, _, extensions) = self.prepare_query(&request.query, &request.variables)?;
 
-        let document = match document.into_data(query.operation_name.as_deref()) {
+        let document = match document.into_data(request.operation_name.as_deref()) {
             Some(document) => document,
             None => {
-                let err = if let Some(name) = query.operation_name {
+                let err = if let Some(name) = request.operation_name {
                     QueryError::UnknownOperationNamed {
                         name: name.to_string(),
                     }
@@ -460,9 +463,9 @@ where
         let resolve_id = AtomicUsize::default();
         let env = QueryEnv::new(
             extensions,
-            query.variables,
+            request.variables,
             document,
-            Arc::new(query.ctx_data),
+            Arc::new(request.ctx_data),
         );
         let ctx = env.create_context(
             &self.env,

src/subscription/mod.rs

@@ -1,8 +1,8 @@
-mod connection;
+//mod connection;
 mod simple_broker;
 mod subscription_type;
 // mod ws_transport;
 
-pub use connection::{create_connection, ConnectionTransport, SubscriptionStreams};
+//pub use connection::{create_connection, ConnectionTransport, SubscriptionStreams};
 pub use simple_broker::SimpleBroker;
 pub use subscription_type::{create_subscription_stream, SubscriptionType};

src/types/maybe_undefined.rs

@@ -35,7 +35,7 @@ use std::borrow::Cow;
 ///             v3:value1
 ///         }"#;
 ///     assert_eq!(
-///         schema.execute(&query).await.unwrap().data,
+///         schema.execute(query).await.unwrap().data,
 ///         serde_json::json!({
 ///             "v1": 99,
 ///             "v2": 1,

tests/complexity_and_depth.rs

@@ -37,7 +37,7 @@ pub async fn test_complexity_and_depth() {
         .limit_complexity(2)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap_err(),
+        schema.execute(query).await.unwrap_err(),
         Error::Query {
             pos: Pos { line: 0, column: 0 },
             path: None,
@@ -50,7 +50,7 @@ pub async fn test_complexity_and_depth() {
         .limit_complexity(2)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "a": 1,
             "b": 1,
@@ -62,7 +62,7 @@ pub async fn test_complexity_and_depth() {
         .limit_complexity(2)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap_err(),
+        schema.execute(query).await.unwrap_err(),
         Error::Query {
             pos: Pos { line: 0, column: 0 },
             path: None,
@@ -75,7 +75,7 @@ pub async fn test_complexity_and_depth() {
         .limit_complexity(2)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "obj": { "a": 1 }
         })
@@ -95,7 +95,7 @@ pub async fn test_complexity_and_depth() {
         .limit_depth(2)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap_err(),
+        schema.execute(query).await.unwrap_err(),
         Error::Query {
             pos: Pos { line: 0, column: 0 },
             path: None,
@@ -117,7 +117,7 @@ pub async fn test_complexity_and_depth() {
         .limit_depth(3)
         .finish();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "obj": {
                 "a": 1,

tests/connection.rs

@@ -65,7 +65,6 @@ pub async fn test_connection_additional_fields() {
         schema
             .execute("{ numbers(first: 2) { totalCount edges { node diff } } }")
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "numbers": {
@@ -82,7 +81,6 @@ pub async fn test_connection_additional_fields() {
         schema
             .execute("{ numbers(last: 2) { edges { node diff } } }")
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "numbers": {

tests/default_value.rs

@@ -22,7 +22,7 @@ pub async fn test_default_value_arg() {
     let query = "{ value1 value2 value3 }";
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value1": 100,
             "value2": 0,
@@ -33,7 +33,7 @@ pub async fn test_default_value_arg() {
     let query = "{ value1(input: 1) value2(input: 2) value3(input: 3) }";
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value1": 1,
             "value2": 2,
@@ -79,7 +79,7 @@ pub async fn test_default_value_inputobject() {
     let query = "{ value(input: {}) { value1 value2 value3 } }";
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value": {
                 "value1": 100,
@@ -92,7 +92,7 @@ pub async fn test_default_value_inputobject() {
     let query = "{ value(input: { value1: 1, value2: 2, value3: 3 }) { value1 value2 value3 } }";
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value": {
                 "value1": 1,

tests/directive.rs

@@ -21,8 +21,7 @@ pub async fn test_directive_skip() {
             }
         "#,
         )
-        .await
-        .unwrap();
+        .await;
     assert_eq!(
         resp.data,
         serde_json::json!({
@@ -52,8 +51,7 @@ pub async fn test_directive_include() {
             }
         "#,
         )
-        .await
-        .unwrap();
+        .await;
     assert_eq!(
         resp.data,
         serde_json::json!({
@@ -92,8 +90,7 @@ pub async fn test_directive_ifdef() {
             }
         "#,
         )
-        .await
-        .unwrap();
+        .await;
     assert_eq!(
         resp.data,
         serde_json::json!({
@@ -110,8 +107,7 @@ pub async fn test_directive_ifdef() {
             }
         "#,
         )
-        .await
-        .unwrap();
+        .await;
     assert_eq!(
         resp.data,
         serde_json::json!({

tests/enum.rs

@@ -40,7 +40,7 @@ pub async fn test_enum_type() {
         }"#
     .to_owned();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(&query).await.data,
         serde_json::json!({
             "value": "A",
             "testArg": "A",

tests/federation.rs

@@ -79,7 +79,7 @@ pub async fn test_federation() {
             }
         }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "_entities": [
                 {"__typename": "Product", "upc": "B00005N5PF"},

tests/field_features.rs

@@ -68,7 +68,7 @@ pub async fn test_field_features() {
     let schema = Schema::new(QueryRoot, EmptyMutation, Subscription);
     let query = "{ value }";
     assert_eq!(
-        schema.execute(query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value": 10,
         })
@@ -76,7 +76,7 @@ pub async fn test_field_features() {
 
     let query = "{ valueBson }";
     assert_eq!(
-        schema.execute(query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "valueBson": 10,
         })
@@ -97,7 +97,7 @@ pub async fn test_field_features() {
 
     let query = "{ obj { value } }";
     assert_eq!(
-        schema.execute(query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "obj": { "value": 10 }
         })
@@ -105,7 +105,7 @@ pub async fn test_field_features() {
 
     let query = "{ obj { valueBson } }";
     assert_eq!(
-        schema.execute(query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "obj": { "valueBson": 10 }
         })
@@ -125,62 +125,44 @@ pub async fn test_field_features() {
     );
 
     let mut stream = schema
-        .create_subscription_stream(
-            "subscription { values }",
-            None,
-            Default::default(),
-            Default::default(),
-        )
+        .execute_stream("subscription { values }")
         .await
         .unwrap();
     assert_eq!(
-        stream.next().await,
-        Some(Ok(serde_json::json!({
+        stream.next().await.map(|resp| resp.data),
+        Some(serde_json::json!({
             "values": 10
-        })))
+        }))
     );
 
     let mut stream = schema
-        .create_subscription_stream(
-            "subscription { valuesBson }",
-            None,
-            Default::default(),
-            Default::default(),
-        )
+        .execute_stream("subscription { valuesBson }")
         .await
         .unwrap();
     assert_eq!(
-        stream.next().await,
-        Some(Ok(serde_json::json!({
+        stream.next().await.map(|resp| resp.data),
+        Some(serde_json::json!({
             "valuesBson": 10
-        })))
+        }))
     );
 
-    let res = schema
-        .create_subscription_stream(
-            "subscription { valuesAbc }",
-            None,
-            Default::default(),
-            Default::default(),
-        )
-        .await;
-    if let Err(err) = res {
-        assert_eq!(
-            err,
-            Error::Query {
-                pos: Pos {
-                    column: 16,
-                    line: 1
-                },
-                path: Some(serde_json::json!(["valuesAbc"])),
-                err: QueryError::FieldError {
-                    err: "`valuesAbc` is only available if the features `abc` are enabled"
-                        .to_string(),
-                    extended_error: None
-                }
+    let err = schema
+        .execute_stream("subscription { valuesAbc }")
+        .await
+        .map(|_| ())
+        .unwrap_err();
+    assert_eq!(
+        err,
+        Error::Query {
+            pos: Pos {
+                column: 16,
+                line: 1
+            },
+            path: Some(serde_json::json!(["valuesAbc"])),
+            err: QueryError::FieldError {
+                err: "`valuesAbc` is only available if the features `abc` are enabled".to_string(),
+                extended_error: None
             }
-        )
-    } else {
-        unreachable!()
-    }
+        }
+    );
 }

tests/fields_merge.rs

@@ -32,7 +32,7 @@ pub async fn test_field_merge() {
         }
     "#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "value1": 1,
             "value2": 2,
@@ -72,7 +72,7 @@ pub async fn test_field_object_merge() {
         }
     "#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "obj": {
                 "a": 1,

tests/guard.rs

@@ -1,7 +1,6 @@
 use async_graphql::guard::Guard;
 use async_graphql::*;
 use futures::{Stream, StreamExt};
-use std::sync::Arc;
 
 #[derive(Eq, PartialEq, Copy, Clone)]
 enum Role {
@@ -77,11 +76,9 @@ pub async fn test_guard() {
 
     let query = "{ obj { value } }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Role::Admin))
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "obj": {"value": 99}
@@ -90,9 +87,8 @@ pub async fn test_guard() {
 
     let query = "{ obj { value } }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Role::Guest))
             .await
             .unwrap_err(),
         Error::Query {
@@ -107,11 +103,9 @@ pub async fn test_guard() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Role::Admin))
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "value": 1,
@@ -120,9 +114,8 @@ pub async fn test_guard() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Role::Guest))
             .await
             .unwrap_err(),
         Error::Query {
@@ -137,42 +130,25 @@ pub async fn test_guard() {
 
     assert_eq!(
         schema
-            .create_subscription_stream(
-                "subscription { values }",
-                None,
-                Variables::default(),
-                Some(Arc::new({
-                    let mut data = Data::default();
-                    data.insert(Role::Admin);
-                    data
-                })),
-            )
+            .execute_stream(Request::new("subscription { values }").data(Role::Admin))
             .await
             .unwrap()
+            .map(|item| item.data)
             .collect::<Vec<_>>()
             .await,
         vec![
-            Ok(serde_json::json! ({"values": 1})),
-            Ok(serde_json::json! ({"values": 2})),
-            Ok(serde_json::json! ({"values": 3}))
+            serde_json::json! ({"values": 1}),
+            serde_json::json! ({"values": 2}),
+            serde_json::json! ({"values": 3})
         ]
     );
 
     assert_eq!(
         schema
-            .create_subscription_stream(
-                "subscription { values }",
-                None,
-                Variables::default(),
-                Some(Arc::new({
-                    let mut data = Data::default();
-                    data.insert(Role::Guest);
-                    data
-                })),
-            )
+            .execute_stream(Request::new("subscription { values }").data(Role::Guest))
             .await
-            .err()
-            .unwrap(),
+            .map(|_| ())
+            .unwrap_err(),
         Error::Query {
             pos: Pos {
                 line: 1,
@@ -199,22 +175,25 @@ pub async fn test_multiple_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Admin)
+                    .data(Username("test".to_string()))
+            )
             .await
-            .unwrap()
             .data,
         serde_json::json!({"value": 10})
     );
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Guest)
+                    .data(Username("test".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -229,10 +208,12 @@ pub async fn test_multiple_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Admin)
+                    .data(Username("test1".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -247,10 +228,12 @@ pub async fn test_multiple_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Guest)
+                    .data(Username("test1".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -295,20 +278,17 @@ pub async fn test_guard_forward_arguments() {
 
     let query = r#"{ user(id: "abc") }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("abc"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("abc")))
             .await
-            .unwrap()
             .data,
         serde_json::json!({"user": "abc"})
     );
 
     let query = r#"{ user(id: "abc") }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("aaa"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("aaa")))
             .await
             .unwrap_err(),
         Error::Query {

tests/input_object.rs

@@ -74,7 +74,7 @@ pub async fn test_input_object_default_value() {
         }"#
     .to_owned();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(&query).await.data,
         serde_json::json!({
             "a": {
                 "a": 999,

tests/input_validators.rs

@@ -63,7 +63,7 @@ pub async fn test_input_validator_string_min_length() {
                 schema
                     .execute(&field_query)
                     .await
-                    .expect_err(&should_fail_msg[..]),
+                    .expect_err(&should_fail_msg),
                 Error::Rule {
                     errors: vec!(RuleError {
                         locations: vec!(Pos {
@@ -1651,9 +1651,8 @@ pub async fn test_input_validator_variable() {
             );
 
             assert_eq!(
-                QueryBuilder::new(field_query)
-                    .variables(variables.clone())
-                    .execute(&schema)
+                schema
+                    .execute(Request::new(field_query).variables(variables.clone()))
                     .await
                     .expect_err(&should_fail_msg[..]),
                 Error::Rule {
@@ -1668,9 +1667,8 @@ pub async fn test_input_validator_variable() {
             );
 
             assert_eq!(
-                QueryBuilder::new(object_query)
-                    .variables(variables.clone())
-                    .execute(&schema)
+                schema
+                    .execute(Request::new(object_query).variables(variables.clone()))
                     .await
                     .expect_err(&should_fail_msg[..]),
                 Error::Rule {
@@ -1686,9 +1684,8 @@ pub async fn test_input_validator_variable() {
         } else {
             let error_msg = format!("Schema returned error with test_string = {}", case);
             assert_eq!(
-                QueryBuilder::new(field_query)
-                    .variables(variables.clone())
-                    .execute(&schema)
+                schema
+                    .execute(Request::new(field_query).variables(variables.clone()))
                     .await
                     .expect(&error_msg[..])
                     .data,
@@ -1698,9 +1695,8 @@ pub async fn test_input_validator_variable() {
             );
 
             assert_eq!(
-                QueryBuilder::new(object_query)
-                    .variables(variables.clone())
-                    .execute(&schema)
+                schema
+                    .execute(Request::new(object_query).variables(variables.clone()))
                     .await
                     .expect(&error_msg[..])
                     .data,

tests/input_value.rs

@@ -14,7 +14,7 @@ pub async fn test_input_value_custom_error() {
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     let query = r#"{ parseInt(n:289) }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap_err(),
+        schema.execute(query).await.unwrap_err(),
         Error::Query {
             pos: Pos {
                 line: 1,

tests/interface.rs

@@ -35,7 +35,7 @@ pub async fn test_interface_simple_object() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "id": 33,
@@ -79,7 +79,7 @@ pub async fn test_interface_simple_object2() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "id": 33,
@@ -143,7 +143,7 @@ pub async fn test_multiple_interfaces() {
             }
         }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "myObj": {
                 "valueA": 1,
@@ -219,7 +219,7 @@ pub async fn test_multiple_objects_in_multiple_interfaces() {
              }
          }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "myObj": [{
                 "valueA": 1,
@@ -266,7 +266,7 @@ pub async fn test_interface_field_result() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "value": 10,
@@ -315,7 +315,7 @@ pub async fn test_interface_field_method() {
     let query = "{ test { created_at } }";
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "test": {
                 "created_at": 1,

tests/json_type.rs

@@ -45,7 +45,7 @@ pub async fn test_json_scalar() {
     );
     let query = r#"{ data dataOutput dataOutputClone }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "data": { "a": 10, "b": 20},
             "dataOutput": { "a": 10, "b": 20},

tests/list.rs

@@ -55,7 +55,7 @@ pub async fn test_list_type() {
         json_value
     );
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(&query).await.data,
         serde_json::json!({
             "valueVec": vec![1, 2, 3, 4, 5],
             "valueSlice": vec![1, 2, 3, 4, 5],

tests/maybe_undefined.rs

@@ -44,7 +44,7 @@ pub async fn test_maybe_undefined_type() {
         }
     "#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "v1": 99,
             "v2": 1,

tests/merged_object.rs

@@ -45,7 +45,7 @@ pub async fn test_merged_object() {
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     let query = "{ obj { a b c } }";
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "obj": {
                 "a": 10,
@@ -73,7 +73,7 @@ pub async fn test_merged_object_macro() {
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     let query = "{ obj { a b c } }";
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "obj": {
                 "a": 10,
@@ -101,7 +101,7 @@ pub async fn test_merged_object_derive() {
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     let query = "{ obj { a b c } }";
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "obj": {
                 "a": 10,
@@ -150,7 +150,7 @@ pub async fn test_merged_object_default() {
     let schema = Schema::new(Query::default(), EmptyMutation, EmptySubscription);
     let query = "{ a b }";
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "a": 10,
             "b": 20,
@@ -192,14 +192,15 @@ pub async fn test_merged_subscription() {
 
     {
         let mut stream = schema
-            .create_subscription_stream("subscription { events1 }", None, Default::default(), None)
+            .execute_stream("subscription { events1 }")
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.unwrap().data);
         for i in 0i32..10 {
             assert_eq!(
-                Some(Ok(serde_json::json!({
+                Some(serde_json::json!({
                     "events1": i,
-                }))),
+                })),
                 stream.next().await
             );
         }
@@ -208,14 +209,15 @@ pub async fn test_merged_subscription() {
 
     {
         let mut stream = schema
-            .create_subscription_stream("subscription { events2 }", None, Default::default(), None)
+            .execute_stream("subscription { events2 }")
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.unwrap().data);
         for i in 10i32..20 {
             assert_eq!(
-                Some(Ok(serde_json::json!({
+                Some(serde_json::json!({
                     "events2": i,
-                }))),
+                })),
                 stream.next().await
             );
         }

tests/mut_args.rs

@@ -15,11 +15,7 @@ pub async fn test_mut_args() {
 
     let schema = Schema::build(Query, EmptyMutation, EmptySubscription).finish();
     assert_eq!(
-        schema
-            .execute("{ test(a: 10, b: \"abc\") }")
-            .await
-            .unwrap()
-            .data,
+        schema.execute("{ test(a: 10, b: \"abc\") }").await.data,
         serde_json::json!({
             "test": "11abca"
         })

tests/mutation.rs

@@ -31,10 +31,7 @@ pub async fn test_mutation_execution_order() {
     let schema = Schema::build(QueryRoot, MutationRoot, EmptySubscription)
         .data(list.clone())
         .finish();
-    schema
-        .execute("mutation { append1 append2 }")
-        .await
-        .unwrap();
+    schema.execute("mutation { append1 append2 }").await;
     assert_eq!(list.lock().await[0], 1);
     assert_eq!(list.lock().await[1], 2);
 }
@@ -66,8 +63,7 @@ pub async fn test_mutation_fragment() {
             }
         }"#,
         )
-        .await
-        .unwrap();
+        .await;
     assert_eq!(
         resp.data,
         serde_json::json!({

tests/optional.rs

@@ -59,7 +59,7 @@ pub async fn test_optional_type() {
             }"#
     .to_owned();
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(&query).await.data,
         serde_json::json!({
             "value1": 10,
             "value1Ref": 10,

tests/post_guard.rs

@@ -79,11 +79,9 @@ pub async fn test_post_guard() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Username("test".to_string())))
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "value": 99
@@ -92,9 +90,8 @@ pub async fn test_post_guard() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Username("test1".to_string())))
             .await
             .unwrap_err(),
         Error::Query {
@@ -109,11 +106,9 @@ pub async fn test_post_guard() {
 
     let query = "{ obj { value } }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Username("test".to_string())))
             .await
-            .unwrap()
             .data,
         serde_json::json!({
             "obj": { "value": 88 }
@@ -122,9 +117,8 @@ pub async fn test_post_guard() {
 
     let query = "{ obj { value } }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(Username("test1".to_string())))
             .await
             .unwrap_err(),
         Error::Query {
@@ -153,22 +147,25 @@ pub async fn test_multiple_post_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Admin)
+                    .data(Username("test".to_string()))
+            )
             .await
-            .unwrap()
             .data,
         serde_json::json!({"value": 10})
     );
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .data(Username("test".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Guest)
+                    .data(Username("test".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -183,10 +180,12 @@ pub async fn test_multiple_post_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Admin)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Admin)
+                    .data(Username("test1".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -201,10 +200,12 @@ pub async fn test_multiple_post_guards() {
 
     let query = "{ value }";
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(Role::Guest)
-            .data(Username("test1".to_string()))
-            .execute(&schema)
+        schema
+            .execute(
+                Request::new(query)
+                    .data(Role::Guest)
+                    .data(Username("test1".to_string()))
+            )
             .await
             .unwrap_err(),
         Error::Query {
@@ -250,20 +251,17 @@ pub async fn test_post_guard_forward_arguments() {
 
     let query = r#"{ user(id: "abc") }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("abc"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("abc")))
             .await
-            .unwrap()
             .data,
         serde_json::json!({"user": "haha"})
     );
 
     let query = r#"{ user(id: "abc") }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("aaa"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("aaa")))
             .await
             .unwrap_err(),
         Error::Query {
@@ -308,20 +306,17 @@ pub async fn test_post_guard_generic() {
 
     let query = r#"{ user }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("abc"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("abc")))
             .await
-            .unwrap()
             .data,
         serde_json::json!({"user": "haha"})
     );
 
     let query = r#"{ user }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .data(ID::from("aaa"))
-            .execute(&schema)
+        schema
+            .execute(Request::new(query).data(ID::from("aaa")))
             .await
             .unwrap_err(),
         Error::Query {

tests/raw_ident.rs

@@ -1,5 +1,5 @@
 use async_graphql::*;
-use futures::{Stream, StreamExt};
+use futures::{Stream, StreamExt, TryStreamExt};
 
 #[async_std::test]
 pub async fn test_input_value_custom_error() {
@@ -53,11 +53,7 @@ pub async fn test_input_value_custom_error() {
             enumValue(value: TYPE)
         }"#;
     assert_eq!(
-        QueryBuilder::new(query)
-            .execute(&schema)
-            .await
-            .unwrap()
-            .data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "type": 99,
             "obj": { "i32": 88 },
@@ -66,9 +62,11 @@ pub async fn test_input_value_custom_error() {
     );
 
     let mut stream = schema
-        .create_subscription_stream("subscription { type }", None, Default::default(), None)
+        .execute_stream("subscription { type }")
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.into_result())
+        .map_ok(|resp| resp.data);
     for i in 0..10 {
         assert_eq!(
             Some(Ok(serde_json::json!({ "type": i }))),

tests/subscription.rs

@@ -1,6 +1,5 @@
 use async_graphql::*;
-use futures::{Stream, StreamExt};
-use std::sync::Arc;
+use futures::{Stream, StreamExt, TryStreamExt};
 
 #[async_std::test]
 pub async fn test_subscription() {
@@ -32,17 +31,13 @@ pub async fn test_subscription() {
 
     {
         let mut stream = schema
-            .create_subscription_stream(
-                "subscription { values(start: 10, end: 20) }",
-                None,
-                Default::default(),
-                None,
-            )
+            .execute_stream("subscription { values(start: 10, end: 20) }")
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.unwrap().data);
         for i in 10..20 {
             assert_eq!(
-                Some(Ok(serde_json::json!({ "values": i }))),
+                Some(serde_json::json!({ "values": i })),
                 stream.next().await
             );
         }
@@ -51,17 +46,13 @@ pub async fn test_subscription() {
 
     {
         let mut stream = schema
-            .create_subscription_stream(
-                "subscription { events(start: 10, end: 20) { a b } }",
-                None,
-                Default::default(),
-                None,
-            )
+            .execute_stream("subscription { events(start: 10, end: 20) { a b } }")
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.unwrap().data);
         for i in 10..20 {
             assert_eq!(
-                Some(Ok(serde_json::json!({ "events": {"a": i, "b": i * 10} }))),
+                Some(serde_json::json!({ "events": {"a": i, "b": i * 10} })),
                 stream.next().await
             );
         }
@@ -103,23 +94,15 @@ pub async fn test_simple_broker() {
 
     let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
     let mut stream1 = schema
-        .create_subscription_stream(
-            "subscription { events1 { value } }",
-            None,
-            Default::default(),
-            None,
-        )
+        .execute_stream("subscription { events1 { value } }")
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.unwrap().data);
     let mut stream2 = schema
-        .create_subscription_stream(
-            "subscription { events2 { value } }",
-            None,
-            Default::default(),
-            None,
-        )
+        .execute_stream("subscription { events2 { value } }")
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.unwrap().data);
 
     SimpleBroker::publish(Event1 { value: 10 });
     SimpleBroker::publish(Event2 { value: 88 });
@@ -128,20 +111,20 @@ pub async fn test_simple_broker() {
 
     assert_eq!(
         stream1.next().await,
-        Some(Ok(serde_json::json!({ "events1": {"value": 10} })))
+        Some(serde_json::json!({ "events1": {"value": 10} }))
     );
     assert_eq!(
         stream1.next().await,
-        Some(Ok(serde_json::json!({ "events1": {"value": 15} })))
+        Some(serde_json::json!({ "events1": {"value": 15} }))
     );
 
     assert_eq!(
         stream2.next().await,
-        Some(Ok(serde_json::json!({ "events2": {"value": 88} })))
+        Some(serde_json::json!({ "events2": {"value": 88} }))
     );
     assert_eq!(
         stream2.next().await,
-        Some(Ok(serde_json::json!({ "events2": {"value": 99} })))
+        Some(serde_json::json!({ "events2": {"value": 99} }))
     );
 }
 
@@ -179,24 +162,16 @@ pub async fn test_subscription_with_ctx_data() {
 
     {
         let mut stream = schema
-            .create_subscription_stream(
-                "subscription { values objects { value } }",
-                None,
-                Default::default(),
-                Some(Arc::new({
-                    let mut data = Data::default();
-                    data.insert(100i32);
-                    data
-                })),
-            )
+            .execute_stream(Request::new("subscription { values objects { value } }").data(100i32))
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.data);
         assert_eq!(
-            Some(Ok(serde_json::json!({ "values": 100 }))),
+            Some(serde_json::json!({ "values": 100 })),
             stream.next().await
         );
         assert_eq!(
-            Some(Ok(serde_json::json!({ "objects": { "value": 100 } }))),
+            Some(serde_json::json!({ "objects": { "value": 100 } })),
             stream.next().await
         );
         assert!(stream.next().await.is_none());
@@ -228,20 +203,14 @@ pub async fn test_subscription_with_token() {
 
     {
         let mut stream = schema
-            .create_subscription_stream(
-                "subscription { values }",
-                None,
-                Default::default(),
-                Some(Arc::new({
-                    let mut data = Data::default();
-                    data.insert(Token("123456".to_string()));
-                    data
-                })),
+            .execute_stream(
+                Request::new("subscription { values }").data(Token("123456".to_string())),
             )
             .await
-            .unwrap();
+            .unwrap()
+            .map(|resp| resp.unwrap().data);
         assert_eq!(
-            Some(Ok(serde_json::json!({ "values": 100 }))),
+            Some(serde_json::json!({ "values": 100 })),
             stream.next().await
         );
         assert!(stream.next().await.is_none());
@@ -249,15 +218,8 @@ pub async fn test_subscription_with_token() {
 
     {
         assert!(schema
-            .create_subscription_stream(
-                "subscription { values }",
-                None,
-                Default::default(),
-                Some(Arc::new({
-                    let mut data = Data::default();
-                    data.insert(Token("654321".to_string()));
-                    data
-                })),
+            .execute_stream(
+                Request::new("subscription { values }").data(Token("654321".to_string()))
             )
             .await
             .is_err());
@@ -288,7 +250,7 @@ pub async fn test_subscription_inline_fragment() {
 
     let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
     let mut stream = schema
-        .create_subscription_stream(
+        .execute_stream(
             r#"
             subscription {
                 events(start: 10, end: 20) {
@@ -299,15 +261,13 @@ pub async fn test_subscription_inline_fragment() {
                 }
             }
             "#,
-            None,
-            Default::default(),
-            None,
         )
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.data);
     for i in 10..20 {
         assert_eq!(
-            Some(Ok(serde_json::json!({ "events": {"a": i, "b": i * 10} }))),
+            Some(serde_json::json!({ "events": {"a": i, "b": i * 10} })),
             stream.next().await
         );
     }
@@ -345,7 +305,7 @@ pub async fn test_subscription_fragment() {
         .register_type::<MyInterface>()
         .finish();
     let mut stream = schema
-        .create_subscription_stream(
+        .execute_stream(
             r#"
             subscription s {
                 events(start: 10, end: 20) {
@@ -356,15 +316,13 @@ pub async fn test_subscription_fragment() {
                 }
             }
             "#,
-            None,
-            Default::default(),
-            None,
         )
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.data);
     for i in 10..20 {
         assert_eq!(
-            Some(Ok(serde_json::json!({ "events": {"a": i, "b": i * 10} }))),
+            Some(serde_json::json!({ "events": {"a": i, "b": i * 10} })),
             stream.next().await
         );
     }
@@ -402,7 +360,7 @@ pub async fn test_subscription_fragment2() {
         .register_type::<MyInterface>()
         .finish();
     let mut stream = schema
-        .create_subscription_stream(
+        .execute_stream(
             r#"
             subscription s {
                 events(start: 10, end: 20) {
@@ -414,15 +372,13 @@ pub async fn test_subscription_fragment2() {
                 a b
             }
             "#,
-            None,
-            Default::default(),
-            None,
         )
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.data);
     for i in 10..20 {
         assert_eq!(
-            Some(Ok(serde_json::json!({ "events": {"a": i, "b": i * 10} }))),
+            Some(serde_json::json!({ "events": {"a": i, "b": i * 10} })),
             stream.next().await
         );
     }
@@ -462,14 +418,11 @@ pub async fn test_subscription_error() {
 
     let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
     let mut stream = schema
-        .create_subscription_stream(
-            "subscription { events { value } }",
-            None,
-            Default::default(),
-            None,
-        )
+        .execute_stream("subscription { events { value } }")
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.into_result())
+        .map_ok(|resp| resp.data);
     for i in 0i32..5 {
         assert_eq!(
             Some(Ok(serde_json::json!({ "events": { "value": i } }))),
@@ -516,9 +469,11 @@ pub async fn test_subscription_fieldresult() {
 
     let schema = Schema::new(QueryRoot, EmptyMutation, SubscriptionRoot);
     let mut stream = schema
-        .create_subscription_stream("subscription { values }", None, Default::default(), None)
+        .execute_stream("subscription { values }")
         .await
-        .unwrap();
+        .unwrap()
+        .map(|resp| resp.into_result())
+        .map_ok(|resp| resp.data);
     for i in 0i32..5 {
         assert_eq!(
             Some(Ok(serde_json::json!({ "values": i }))),

tests/union.rs

@@ -35,7 +35,7 @@ pub async fn test_union_simple_object() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "id": 33,
@@ -79,7 +79,7 @@ pub async fn test_union_simple_object2() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "id": 33,
@@ -149,7 +149,7 @@ pub async fn test_multiple_unions() {
              }
         }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "unionA": {
                 "valueA": 1,
@@ -229,7 +229,7 @@ pub async fn test_multiple_objects_in_multiple_unions() {
             }
          }"#;
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "myObj": [{
                 "valueA": 1,
@@ -276,7 +276,7 @@ pub async fn test_union_field_result() {
         }"#;
     let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
     assert_eq!(
-        schema.execute(&query).await.unwrap().data,
+        schema.execute(query).await.unwrap().data,
         serde_json::json!({
             "node": {
                 "value": 10,

tests/variables.rs

@@ -16,7 +16,7 @@ pub async fn test_variables() {
     }
 
     let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
-    let query = QueryBuilder::new(
+    let query = Request::new(
         r#"
             query QueryWithVariables($intVal: Int!, $intListVal: [Int!]!) {
                 intVal(value: $intVal)
@@ -28,9 +28,9 @@ pub async fn test_variables() {
         "intVal": 10,
          "intListVal": [1, 2, 3, 4, 5],
     })));
-    let resp = query.execute(&schema).await.unwrap();
+
     assert_eq!(
-        resp.data,
+        schema.execute(query).await.data,
         serde_json::json!({
             "intVal": 10,
             "intListVal": [1, 2, 3, 4, 5],
@@ -50,18 +50,17 @@ pub async fn test_variable_default_value() {
     }
 
     let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
-    let resp = schema
-        .execute(
-            r#"
+    assert_eq!(
+        schema
+            .execute(
+                r#"
             query QueryWithVariables($intVal: Int = 10) {
                 intVal(value: $intVal)
             }
-        "#,
-        )
-        .await
-        .unwrap();
-    assert_eq!(
-        resp.data,
+        "#
+            )
+            .await
+            .data,
         serde_json::json!({
             "intVal": 10,
         })
@@ -80,15 +79,16 @@ pub async fn test_variable_no_value() {
     }
 
     let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
-    let query = QueryBuilder::new(
-        r#"
+    let resp = schema
+        .execute(Request::new(
+            r#"
             query QueryWithVariables($intVal: Int) {
                 intVal(value: $intVal)
             }
         "#,
-    )
-    .variables(Variables::parse_from_json(serde_json::json!({})));
-    let resp = query.execute(&schema).await.unwrap();
+        ))
+        .await
+        .unwrap();
     assert_eq!(
         resp.data,
         serde_json::json!({
@@ -109,7 +109,7 @@ pub async fn test_variable_null() {
     }
 
     let schema = Schema::new(QueryRoot, EmptyMutation, EmptySubscription);
-    let query = QueryBuilder::new(
+    let query = Request::new(
         r#"
             query QueryWithVariables($intVal: Int) {
                 intVal(value: $intVal)
@@ -119,7 +119,7 @@ pub async fn test_variable_null() {
     .variables(Variables::parse_from_json(serde_json::json!({
         "intVal": null,
     })));
-    let resp = query.execute(&schema).await.unwrap();
+    let resp = schema.execute(query).await;
     assert_eq!(
         resp.data,
         serde_json::json!({
@@ -165,13 +165,13 @@ pub async fn test_variable_in_input_object() {
         query TestQuery($value: Int!) {
             test(input: {value: $value })
         }"#;
-        let resp = QueryBuilder::new(query)
-            .variables(Variables::parse_from_json(serde_json::json!({
-                "value": 10,
-            })))
-            .execute(&schema)
-            .await
-            .unwrap();
+        let resp = schema
+            .execute(
+                Request::new(query).variables(Variables::parse_from_json(serde_json::json!({
+                    "value": 10,
+                }))),
+            )
+            .await;
         assert_eq!(
             resp.data,
             serde_json::json!({
@@ -186,13 +186,13 @@ pub async fn test_variable_in_input_object() {
         query TestQuery($value: Int!) {
             test2(input: [{value: $value }, {value: $value }])
         }"#;
-        let resp = QueryBuilder::new(query)
-            .variables(Variables::parse_from_json(serde_json::json!({
-                "value": 3,
-            })))
-            .execute(&schema)
-            .await
-            .unwrap();
+        let resp = schema
+            .execute(
+                Request::new(query).variables(Variables::parse_from_json(serde_json::json!({
+                    "value": 3,
+                }))),
+            )
+            .await;
         assert_eq!(
             resp.data,
             serde_json::json!({
@@ -207,13 +207,13 @@ pub async fn test_variable_in_input_object() {
         mutation TestMutation($value: Int!) {
             test(input: {value: $value })
         }"#;
-        let resp = QueryBuilder::new(query)
-            .variables(Variables::parse_from_json(serde_json::json!({
-                "value": 10,
-            })))
-            .execute(&schema)
-            .await
-            .unwrap();
+        let resp = schema
+            .execute(
+                Request::new(query).variables(Variables::parse_from_json(serde_json::json!({
+                    "value": 10,
+                }))),
+            )
+            .await;
         assert_eq!(
             resp.data,
             serde_json::json!({