Add subscription support

Cargo.toml

@@ -9,12 +9,15 @@ 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",
 ]

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();
 
@@ -99,12 +100,13 @@ fn main() {
     let graphql_filter = juniper_warp::make_graphql_filter_async(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" }

examples/warp_subscriptions/src/main.rs

@@ -0,0 +1,187 @@
+//! 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;
+use warp::{http::Response, Filter};
+use warp_subscriptions::*;
+
+mod warp_subscriptions;
+
+#[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_async(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_async(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;
+}