Add subscription support

Cargo.toml

@@ -9,12 +9,13 @@ members = [
   "juniper_hyper",
   "juniper_iron",
   "juniper_rocket",
+  "juniper_subscriptions",
+  "juniper_warp",
 ]
 exclude = [
   "docs/book/tests",
-  # TODO enable warp
-  "juniper_warp",
   "examples/warp_async",
+  "examples/warp_subscriptions",
   # TODO enable async tests
   "juniper_rocket_async",
 ]

benches/bench.rs

@@ -3,12 +3,16 @@ extern crate juniper;
 
 use bencher::Bencher;
 
-use juniper::{execute_sync, RootNode, EmptyMutation, Variables};
+use juniper::{execute_sync, RootNode, EmptyMutation, EmptySubscription, Variables};
 use juniper::tests::model::Database;
 
 fn query_type_name(b: &mut Bencher) {
     let database = Database::new();
-    let schema = RootNode::new(&database, EmptyMutation::<Database>::new());
+    let schema = RootNode::new(
+        &database,
+        EmptyMutation::<Database>::new(),
+        EmptySubscription::<Database>::new()
+    );
 
     let doc = r#"
         query IntrospectionQueryTypeQuery {
@@ -24,7 +28,11 @@ fn query_type_name(b: &mut Bencher) {
 
 fn introspection_query(b: &mut Bencher) {
     let database = Database::new();
-    let schema = RootNode::new(&database, EmptyMutation::<Database>::new());
+    let schema = RootNode::new(
+        &database,
+        EmptyMutation::<Database>::new(),
+        EmptySubscription::<Database>::new(),
+    );
 
     let doc = r#"
   query IntrospectionQuery {

docs/book/content/advanced/introspection.md

@@ -30,7 +30,7 @@ result can then be converted to JSON for use with tools and libraries such as
 [graphql-client](https://github.com/graphql-rust/graphql-client):
 
 ```rust
-use juniper::{EmptyMutation, FieldResult, IntrospectionFormat};
+use juniper::{EmptyMutation, EmptySubscription, FieldResult, IntrospectionFormat};
 
 // Define our schema.
 
@@ -53,15 +53,20 @@ impl Query {
    }
 }
 
-type Schema = juniper::RootNode<'static, Query, EmptyMutation<Context>>;
+type Schema = juniper::RootNode<
+    'static, 
+    Query, 
+    EmptyMutation<Context>, 
+    EmptySubscription<Context>
+>;
 
 fn main() {
     // Create a context object.
     let ctx = Context{};
 
     // Run the built-in introspection query.
     let (res, _errors) = juniper::introspect(
-        &Schema::new(Query, EmptyMutation::new()),
+        &Schema::new(Query, EmptyMutation::new(), EmptySubscription::new()),
         &ctx,
         IntrospectionFormat::default(),
     ).unwrap();

docs/book/content/quickstart.md

@@ -24,7 +24,7 @@ types to a GraphQL schema. The most important one is the
 resolvers, which you will use for the `Query` and `Mutation` roots.
 
 ```rust
-use juniper::{FieldResult};
+use juniper::{FieldResult, EmptySubscription};
 
 # struct DatabasePool;
 # impl DatabasePool {
@@ -119,10 +119,10 @@ impl Mutation {
 
 // A root schema consists of a query and a mutation.
 // Request queries can be executed against a RootNode.
-type Schema = juniper::RootNode<'static, Query, Mutation>;
+type Schema = juniper::RootNode<'static, Query, Mutation, EmptySubscription<Context>>;
 
 # fn main() {
-#   let _ = Schema::new(Query, Mutation{});
+#   let _ = Schema::new(Query, Mutation{}, EmptySubscription::new());
 # }
 ```
 
@@ -139,7 +139,7 @@ You can invoke `juniper::execute` directly to run a GraphQL query:
 ```rust
 # // Only needed due to 2018 edition because the macro is not accessible.
 # #[macro_use] extern crate juniper;
-use juniper::{FieldResult, Variables, EmptyMutation};
+use juniper::{FieldResult, Variables, EmptyMutation, EmptySubscription};
 
 
 #[derive(juniper::GraphQLEnum, Clone, Copy)]
@@ -168,7 +168,7 @@ impl Query {
 
 // A root schema consists of a query and a mutation.
 // Request queries can be executed against a RootNode.
-type Schema = juniper::RootNode<'static, Query, EmptyMutation<Ctx>>;
+type Schema = juniper::RootNode<'static, Query, EmptyMutation<Ctx>, EmptySubscription<Ctx>>;
 
 fn main() {
     // Create a context object.
@@ -178,7 +178,7 @@ fn main() {
     let (res, _errors) = juniper::execute_sync(
         "query { favoriteEpisode }",
         None,
-        &Schema::new(Query, EmptyMutation::new()),
+        &Schema::new(Query, EmptyMutation::new(), EmptySubscription::new()),
         &Variables::new(),
         &ctx,
     ).unwrap();

examples/warp_async/Cargo.toml

@@ -12,6 +12,7 @@ env_logger = "0.6.2"
 warp = "0.1.19"
 futures = { version = "0.3.1", features = ["compat"] }
 reqwest = "0.9.19"
+tokio = { version = "0.2", features = ["rt-core", "macros"] }
 
 juniper_codegen = { git = "https://github.com/graphql-rust/juniper", branch = "async-await", features = ["async"] }
 juniper = { git = "https://github.com/graphql-rust/juniper", branch = "async-await", features = ["async"] }

examples/warp_async/src/main.rs

@@ -2,7 +2,7 @@
 //! This example demonstrates async/await usage with warp.
 //! NOTE: this uses tokio 0.1 , not the alpha tokio 0.2.
 
-use juniper::{EmptyMutation, RootNode, FieldError};
+use juniper::{EmptyMutation, EmptySubscription, RootNode, FieldError};
 use warp::{http::Response, Filter};
 
 #[derive(Clone)]
@@ -73,13 +73,14 @@ impl Query {
     }
 }
 
-type Schema = RootNode<'static, Query, EmptyMutation<Context>>;
+type Schema = RootNode<'static, Query, EmptyMutation<Context>, EmptySubscription<Context>>;
 
 fn schema() -> Schema {
-    Schema::new(Query, EmptyMutation::<Context>::new())
+    Schema::new(Query, EmptyMutation::<Context>::new(), EmptySubscription::<Context>::new())
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
     ::std::env::set_var("RUST_LOG", "warp_async");
     env_logger::init();
 
@@ -96,15 +97,16 @@ fn main() {
     log::info!("Listening on 127.0.0.1:8080");
 
     let state = warp::any().map(move || Context{} );
-    let graphql_filter = juniper_warp::make_graphql_filter_async(schema(), state.boxed());
+    let graphql_filter = juniper_warp::make_graphql_filter(schema(), state.boxed());
 
     warp::serve(
-        warp::get2()
+        warp::get()
             .and(warp::path("graphiql"))
             .and(juniper_warp::graphiql_filter("/graphql"))
             .or(homepage)
             .or(warp::path("graphql").and(graphql_filter))
             .with(log),
     )
-    .run(([127, 0, 0, 1], 8080));
+    .run(([127, 0, 0, 1], 8080))
+    .await
 }

examples/warp_subscriptions/.gitignore

@@ -0,0 +1 @@
+target

examples/warp_subscriptions/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "warp_subscriptions"
+version = "0.1.0"
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+env_logger = "0.6.2"
+futures = { version = "=0.3.1" }
+log = "0.4.8"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+tokio = { version = "0.2", features = ["rt-core", "macros"] }
+warp = "0.2.1"
+
+# TODO#433: get crates from GitHub
+juniper = { path = "../../juniper" }
+juniper_subscriptions = { path = "../../juniper_subscriptions"}
+juniper_warp = { path = "../../juniper_warp", features = ["subscriptions"] }

examples/warp_subscriptions/src/main.rs

@@ -0,0 +1,184 @@
+//! This example demonstrates asynchronous subscriptions with warp and tokio 0.2
+
+use std::{pin::Pin, sync::Arc, time::Duration};
+
+use futures::{Future, FutureExt as _, Stream};
+use juniper::{DefaultScalarValue, EmptyMutation, FieldError, RootNode};
+use juniper_subscriptions::Coordinator;
+use juniper_warp::{playground_filter, subscriptions::graphql_subscriptions};
+use warp::{http::Response, Filter};
+
+#[derive(Clone)]
+struct Context {}
+
+impl juniper::Context for Context {}
+
+#[derive(Clone, Copy, juniper::GraphQLEnum)]
+enum UserKind {
+    Admin,
+    User,
+    Guest,
+}
+
+struct User {
+    id: i32,
+    kind: UserKind,
+    name: String,
+}
+
+// Field resolvers implementation
+#[juniper::graphql_object(Context = Context)]
+impl User {
+    fn id(&self) -> i32 {
+        self.id
+    }
+
+    fn kind(&self) -> UserKind {
+        self.kind
+    }
+
+    fn name(&self) -> &str {
+        &self.name
+    }
+
+    async fn friends(&self) -> Vec<User> {
+        if self.id == 1 {
+            return vec![
+                User {
+                    id: 11,
+                    kind: UserKind::User,
+                    name: "user11".into(),
+                },
+                User {
+                    id: 12,
+                    kind: UserKind::Admin,
+                    name: "user12".into(),
+                },
+                User {
+                    id: 13,
+                    kind: UserKind::Guest,
+                    name: "user13".into(),
+                },
+            ];
+        } else if self.id == 2 {
+            return vec![User {
+                id: 21,
+                kind: UserKind::User,
+                name: "user21".into(),
+            }];
+        } else if self.id == 3 {
+            return vec![
+                User {
+                    id: 31,
+                    kind: UserKind::User,
+                    name: "user31".into(),
+                },
+                User {
+                    id: 32,
+                    kind: UserKind::Guest,
+                    name: "user32".into(),
+                },
+            ];
+        } else {
+            return vec![];
+        }
+    }
+}
+
+struct Query;
+
+#[juniper::graphql_object(Context = Context)]
+impl Query {
+    async fn users(id: i32) -> Vec<User> {
+        vec![User {
+            id,
+            kind: UserKind::Admin,
+            name: "User Name".into(),
+        }]
+    }
+}
+
+type UsersStream = Pin<Box<dyn Stream<Item = Result<User, FieldError>> + Send>>;
+
+struct Subscription;
+
+#[juniper::graphql_subscription(Context = Context)]
+impl Subscription {
+    async fn users() -> UsersStream {
+        let mut counter = 0;
+        let stream = tokio::time::interval(Duration::from_secs(5)).map(move |_| {
+            counter += 1;
+            if counter == 2 {
+                Err(FieldError::new(
+                    "some field error from handler",
+                    Value::Scalar(DefaultScalarValue::String(
+                        "some additional string".to_string(),
+                    )),
+                ))
+            } else {
+                Ok(User {
+                    id: counter,
+                    kind: UserKind::Admin,
+                    name: "stream user".to_string(),
+                })
+            }
+        });
+
+        Box::pin(stream)
+    }
+}
+
+type Schema = RootNode<'static, Query, EmptyMutation<Context>, Subscription>;
+
+fn schema() -> Schema {
+    Schema::new(Query, EmptyMutation::new(), Subscription)
+}
+
+#[tokio::main]
+async fn main() {
+    ::std::env::set_var("RUST_LOG", "warp_subscriptions");
+    env_logger::init();
+
+    let log = warp::log("warp_server");
+
+    let homepage = warp::path::end().map(|| {
+        Response::builder()
+            .header("content-type", "text/html")
+            .body(format!(
+                "<html><h1>juniper_subscriptions demo</h1><div>visit <a href=\"/playground\">graphql playground</a></html>"
+            ))
+    });
+
+    let qm_schema = schema();
+    let qm_state = warp::any().map(move || Context {});
+    let qm_graphql_filter = juniper_warp::make_graphql_filter(qm_schema, qm_state.boxed());
+
+    let sub_state = warp::any().map(move || Context {});
+    let coordinator = Arc::new(juniper_subscriptions::Coordinator::new(schema()));
+
+    log::info!("Listening on 127.0.0.1:8080");
+
+    let routes = (warp::path("subscriptions")
+        .and(warp::ws())
+        .and(sub_state.clone())
+        .and(warp::any().map(move || Arc::clone(&coordinator)))
+        .map(
+            |ws: warp::ws::Ws,
+             ctx: Context,
+             coordinator: Arc<Coordinator<'static, _, _, _, _, _>>| {
+                ws.on_upgrade(|websocket| -> Pin<Box<dyn Future<Output = ()> + Send>> {
+                    graphql_subscriptions(websocket, coordinator, ctx).boxed()
+                })
+            },
+        ))
+    .or(warp::post()
+        .and(warp::path("graphql"))
+        .and(qm_graphql_filter))
+    .or(warp::get()
+        .and(warp::path("playground"))
+        .and(playground_filter("/graphql", Some("/subscriptions"))))
+    .or(homepage)
+    .with(log);
+
+    warp::serve(routes).run(([127, 0, 0, 1], 8080)).await;
+}