Add subscription support

juniper/src/http/mod.rs

@@ -14,7 +14,9 @@ use crate::{
     executor::ExecutionError,
     value::{DefaultScalarValue, ScalarRefValue, ScalarValue},
     FieldError, GraphQLError, GraphQLType, RootNode, Value, Variables,
+    ValuesIterator, ValuesStream,
 };
+use crate::executor::OwnedExecutor;
 
 /// The expected structure of the decoded JSON document for either POST or GET requests.
 ///
@@ -70,19 +72,75 @@ where
         }
     }
 
-    /// Execute a GraphQL request using the specified schema and context
+    /// Execute a GraphQL subscription using the specified schema and context
     ///
-    /// This is a simple wrapper around the `execute` function exposed at the
+    /// This is a simple wrapper around the `subscribe` function exposed at the
     /// top level of this crate.
-    pub fn execute<'a, CtxT, QueryT, MutationT>(
+    pub fn subscribe<'a, CtxT, QueryT, MutationT, SubscriptionT>(
         &'a self,
-        root_node: &'a RootNode<QueryT, MutationT, S>,
+        root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
+        context: &'a CtxT,
+        executor_variables: &'a mut OwnedExecutor<'a, CtxT, S>,
+        fragments: &'a mut Vec<crate::parser::Spanning<crate::ast::Fragment<'a, S>>>,
+        executor: &'a mut crate::executor::OptionalExecutor<'a, CtxT, S>,
+    ) -> IteratorGraphQLResponse<'a, S>
+    where
+        S: ScalarValue + Send + Sync + 'static,
+        QueryT: GraphQLType<S, Context = CtxT>,
+        MutationT: GraphQLType<S, Context = CtxT>,
+        SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
+        for<'b> &'b S: ScalarRefValue<'b>,
+    {
+        IteratorGraphQLResponse(crate::subscribe(
+            &self.query,
+            self.operation_name(),
+            root_node,
+            self.variables(),
+            context,
+            executor_variables,
+            fragments,
+            executor,
+        ))
+    }
+
+    /// Execute a GraphQL subscription using the specified schema and context
+    ///
+    /// This is a simple wrapper around the `subscribe_async` function exposed at the
+    /// top level of this crate.
+    #[cfg(feature = "async")]
+    pub async fn subscribe_async<'a, CtxT, QueryT, MutationT, SubscriptionT>(
+        &'a self,
+        root_node: &'a RootNode<'a, QueryT, MutationT, SubscriptionT, S>,
+        context: &'a CtxT,
+    ) -> StreamGraphQLResponse<'a, S>
+    where
+        S: ScalarValue + Send + Sync + 'static,
+        QueryT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+        QueryT::TypeInfo: Send + Sync,
+        MutationT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+        MutationT::TypeInfo: Send + Sync,
+        SubscriptionT: crate::SubscriptionHandlerAsync<S, Context = CtxT> + Send + Sync,
+        SubscriptionT::TypeInfo: Send + Sync,
+        CtxT: Send + Sync,
+        for<'b> &'b S: ScalarRefValue<'b>,
+    {
+        let op = self.operation_name();
+        let vars = &self.variables();
+        let res = crate::subscribe_async(&self.query, op, root_node, vars, context).await;
+
+        StreamGraphQLResponse(res)
+    }
+
+    pub fn execute<'a, CtxT, QueryT, MutationT, SubscriptionT>(
+        &'a self,
+        root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
         context: &CtxT,
     ) -> GraphQLResponse<'a, S>
     where
-        S: ScalarValue,
+        S: ScalarValue + Send + Sync + 'static,
         QueryT: GraphQLType<S, Context = CtxT>,
         MutationT: GraphQLType<S, Context = CtxT>,
+        SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
         for<'b> &'b S: ScalarRefValue<'b>,
     {
         GraphQLResponse(crate::execute(
@@ -95,17 +153,19 @@ where
     }
 
     #[cfg(feature = "async")]
-    pub async fn execute_async<'a, CtxT, QueryT, MutationT>(
+    pub async fn execute_async<'a, CtxT, QueryT, MutationT, SubscriptionT>(
         &'a self,
-        root_node: &'a RootNode<'a, QueryT, MutationT, S>,
+        root_node: &'a RootNode<'a, QueryT, MutationT, SubscriptionT, S>,
         context: &'a CtxT,
     ) -> GraphQLResponse<'a, S>
     where
-        S: ScalarValue + Send + Sync,
+        S: ScalarValue + Send + Sync + 'static,
         QueryT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
         QueryT::TypeInfo: Send + Sync,
         MutationT: crate::GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
         MutationT::TypeInfo: Send + Sync,
+        SubscriptionT: crate::SubscriptionHandlerAsync<S, Context = CtxT> + Send + Sync,
+        SubscriptionT::TypeInfo: Send + Sync,
         CtxT: Send + Sync,
         for<'b> &'b S: ScalarRefValue<'b>,
     {
@@ -122,9 +182,25 @@ where
 /// to JSON and send it over the wire. Use the `is_ok` method to determine
 /// whether to send a 200 or 400 HTTP status code.
 pub struct GraphQLResponse<'a, S = DefaultScalarValue>(
-    Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>,
+    //todo: remove pub (pub is used in playground to access result)
+    pub Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>,
 );
 
+//todo: remove pub (pub is used in playground to access result)
+pub struct IteratorGraphQLResponse<'a, S = DefaultScalarValue>(
+    pub Result<(Value<ValuesIterator<'a, S>>, Vec<ExecutionError<S>>), GraphQLError<'a>>,
+)
+where
+    S: 'static;
+
+#[cfg(feature = "async")]
+pub struct StreamGraphQLResponse<'a, S = DefaultScalarValue>(
+    //todo: remove pub (pub is used in playground to access result)
+    pub Result<(Value<ValuesStream<S>>, Vec<ExecutionError<S>>), GraphQLError<'a>>,
+)
+where
+    S: 'static;
+
 impl<'a, S> GraphQLResponse<'a, S>
 where
     S: ScalarValue,

juniper/src/integrations/serde.rs

@@ -74,6 +74,10 @@ impl<'a> ser::Serialize for GraphQLError<'a> {
                 message: "Expected query, got subscription",
             }]
             .serialize(serializer),
+            GraphQLError::NotSubscription => [SerializeHelper {
+                message: "Expected subscription, got query",
+            }]
+            .serialize(serializer),
         }
     }
 }
@@ -408,7 +412,7 @@ mod tests {
     use super::{ExecutionError, GraphQLError};
     use crate::{
         ast::InputValue,
-        value::{DefaultScalarValue, Object},
+        value::{base_object::SyncObject, DefaultScalarValue, Object},
         FieldError, Value,
     };
     use serde_json::{from_str, to_string};

juniper/src/lib.rs

@@ -138,6 +138,9 @@ pub mod integrations;
 // TODO: remove this alias export in 0.10. (breaking change)
 pub use crate::http::graphiql;
 
+pub use crate::executor::OptionalExecutor;
+
+
 #[cfg(all(test, not(feature = "expose-test-schema")))]
 mod tests;
 #[cfg(feature = "expose-test-schema")]
@@ -175,12 +178,27 @@ pub use crate::{
     },
 };
 
+pub use executor::OwnedExecutor;
+
 /// A pinned, boxed future that can be polled.
 pub type BoxFuture<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + 'a + Send>>;
 
 #[cfg(feature = "async")]
 pub use crate::types::async_await::GraphQLTypeAsync;
 
+#[cfg(feature = "async")]
+pub use crate::executor::SubscriptionResultAsync;
+
+pub use crate::executor::SubscriptionResult;
+#[cfg(feature = "async")]
+pub use crate::types::async_await::SubscriptionHandlerAsync;
+
+pub use crate::types::base::SubscriptionHandler;
+
+pub use crate::executor::ValuesIterator;
+#[cfg(feature = "async")]
+pub use crate::executor::ValuesStream;
+
 /// An error that prevented query execution
 #[derive(Debug, PartialEq)]
 #[allow(missing_docs)]
@@ -191,21 +209,20 @@ pub enum GraphQLError<'a> {
     MultipleOperationsProvided,
     UnknownOperationName,
     IsSubscription,
+    NotSubscription,
 }
 
-/// Execute a query in a provided schema
-pub fn execute<'a, S, CtxT, QueryT, MutationT>(
+fn parse_and_validate_document<'a, QueryT, MutationT, SubscriptionT, CtxT, S>(
     document_source: &'a str,
-    operation_name: Option<&str>,
-    root_node: &'a RootNode<QueryT, MutationT, S>,
+    root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
     variables: &Variables<S>,
-    context: &CtxT,
-) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+) -> Result<Vec<crate::ast::Definition<'a, S>>, GraphQLError<'a>>
 where
-    S: ScalarValue,
+    S: ScalarValue + Send + Sync + 'static,
     for<'b> &'b S: ScalarRefValue<'b>,
     QueryT: GraphQLType<S, Context = CtxT>,
     MutationT: GraphQLType<S, Context = CtxT>,
+    SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
 {
     let document = parse_document_source(document_source, &root_node.schema)?;
     {
@@ -226,26 +243,25 @@ where
         }
     }
 
-    executor::execute_validated_query(document, operation_name, root_node, variables, context)
+    Ok(document)
 }
 
-/// Execute a query in a provided schema
-#[cfg(feature = "async")]
-pub async fn execute_async<'a, S, CtxT, QueryT, MutationT>(
+fn parse_and_validate_document_async<'a, QueryT, MutationT, SubscriptionT, CtxT, S>(
     document_source: &'a str,
-    operation_name: Option<&str>,
-    root_node: &'a RootNode<'a, QueryT, MutationT, S>,
+    root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
     variables: &Variables<S>,
-    context: &CtxT,
-) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+) -> Result<Vec<crate::ast::Definition<'a, S>>, GraphQLError<'a>>
 where
-    S: ScalarValue + Send + Sync,
-    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    S: ScalarValue + Send + Sync + 'static,
+    for<'b> &'b S: ScalarRefValue<'b>,
+    QueryT: GraphQLTypeAsync<S, Context = CtxT>,
     QueryT::TypeInfo: Send + Sync,
-    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT>,
     MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: crate::SubscriptionHandlerAsync<S, Context = CtxT>,
+    SubscriptionT::Context: Send + Sync,
+    SubscriptionT::TypeInfo: Send + Sync,
     CtxT: Send + Sync,
-    for<'b> &'b S: ScalarRefValue<'b>,
 {
     let document = parse_document_source(document_source, &root_node.schema)?;
     {
@@ -266,21 +282,120 @@ where
         }
     }
 
+    Ok(document)
+}
+
+/// Execute a query in a provided schema
+pub fn execute<'a, S, CtxT, QueryT, MutationT, SubscriptionT>(
+    document_source: &'a str,
+    operation_name: Option<&str>,
+    root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
+    variables: &Variables<S>,
+    context: &CtxT,
+) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+where
+    S: ScalarValue + Send + Sync + 'static,
+    for<'b> &'b S: ScalarRefValue<'b>,
+    QueryT: GraphQLType<S, Context = CtxT>,
+    MutationT: GraphQLType<S, Context = CtxT>,
+    SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
+{
+    let document = parse_and_validate_document(document_source, root_node, variables)?;
+
+    executor::execute_validated_query(document, operation_name, root_node, variables, context)
+}
+
+pub fn subscribe<'a, S, CtxT, QueryT, MutationT, SubscriptionT>(
+    document_source: &'a str,
+    operation_name: Option<&str>,
+    root_node: &'a RootNode<'a, QueryT, MutationT, SubscriptionT, S>,
+    variables: Variables<S>,
+    context: &'a CtxT,
+    executor_variables: &'a mut OwnedExecutor<'a, CtxT, S>,
+    fragments: &'a mut Vec<Spanning<crate::ast::Fragment<'a, S>>>,
+    executor: &'a mut crate::executor::OptionalExecutor<'a, CtxT, S>,
+) -> Result<(Value<ValuesIterator<'a, S>>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+where
+    S: ScalarValue + Send + Sync + 'static,
+    for<'b> &'b S: ScalarRefValue<'b>,
+    QueryT: GraphQLType<S, Context = CtxT>,
+    MutationT: GraphQLType<S, Context = CtxT>,
+    SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
+{
+    let document = parse_and_validate_document(document_source, root_node, &variables)?;
+
+    executor::execute_validated_subcription(document, operation_name, root_node, variables, context, executor_variables, fragments, executor)
+}
+
+/// Execute a query in a provided schema
+#[cfg(feature = "async")]
+pub async fn execute_async<'a, S, CtxT, QueryT, MutationT, SubscriptionT>(
+    document_source: &'a str,
+    operation_name: Option<&str>,
+    root_node: &'a RootNode<'a, QueryT, MutationT, SubscriptionT, S>,
+    variables: &Variables<S>,
+    context: &CtxT,
+) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+where
+    S: ScalarValue + Send + Sync + 'static,
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: SubscriptionHandlerAsync<S, Context = CtxT> + Send + Sync,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Send + Sync,
+    for<'b> &'b S: ScalarRefValue<'b>,
+{
+    let document = parse_and_validate_document_async(document_source, root_node, variables)?;
+
     executor::execute_validated_query_async(document, operation_name, root_node, variables, context)
         .await
 }
 
+#[cfg(feature = "async")]
+pub async fn subscribe_async<'a, S, CtxT, QueryT, MutationT, SubscriptionT>(
+    document_source: &'a str,
+    operation_name: Option<&str>,
+    root_node: &'a RootNode<'a, QueryT, MutationT, SubscriptionT, S>,
+    variables: &Variables<S>,
+    context: &CtxT,
+) -> Result<(Value<ValuesStream<S>>, Vec<ExecutionError<S>>), GraphQLError<'a>>
+where
+    S: ScalarValue + Send + Sync + 'static,
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + Sync,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: SubscriptionHandlerAsync<S, Context = CtxT> + Send + Sync,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Send + Sync,
+    for<'b> &'b S: ScalarRefValue<'b>,
+{
+    let document = parse_and_validate_document_async(document_source, root_node, variables)?;
+
+    executor::execute_validated_subscription_async(
+        document,
+        operation_name,
+        root_node,
+        variables,
+        context,
+    )
+    .await
+}
+
 /// Execute the reference introspection query in the provided schema
-pub fn introspect<'a, S, CtxT, QueryT, MutationT>(
-    root_node: &'a RootNode<QueryT, MutationT, S>,
+pub fn introspect<'a, S, CtxT, QueryT, MutationT, SubscriptionT>(
+    root_node: &'a RootNode<QueryT, MutationT, SubscriptionT, S>,
     context: &CtxT,
     format: IntrospectionFormat,
 ) -> Result<(Value<S>, Vec<ExecutionError<S>>), GraphQLError<'a>>
 where
-    S: ScalarValue,
+    S: ScalarValue + Send + Sync + 'static,
     for<'b> &'b S: ScalarRefValue<'b>,
     QueryT: GraphQLType<S, Context = CtxT>,
     MutationT: GraphQLType<S, Context = CtxT>,
+    SubscriptionT: crate::SubscriptionHandler<S, Context = CtxT>,
 {
     execute(
         match format {