GraphQL-WS crate and Warp subscriptions update (#721)

* update pre-existing juniper_warp::subscriptions

* initial draft

* finish up, update example

* polish + timing test

* fix pre-existing bug

* rebase updates

* address comments

* add release.toml

* makefile and initial changelog

* add new Cargo.toml to juniper/release.toml

Author: ccbrown

Cargo.toml

@@ -15,6 +15,7 @@ members = [
   "juniper_rocket",
   "juniper_rocket_async",
   "juniper_subscriptions",
+  "juniper_graphql_ws",
   "juniper_warp",
   "juniper_actix",
 ]

examples/warp_subscriptions/Cargo.toml

@@ -13,6 +13,6 @@ serde_json = "1.0"
 tokio = { version = "0.2", features = ["rt-core", "macros"] }
 warp = "0.2.1"
 
-juniper = { git = "https://github.com/graphql-rust/juniper" }
-juniper_subscriptions = { git = "https://github.com/graphql-rust/juniper" }
-juniper_warp = { git = "https://github.com/graphql-rust/juniper", features = ["subscriptions"] }
+juniper = { path = "../../juniper" }
+juniper_graphql_ws = { path = "../../juniper_graphql_ws" }
+juniper_warp = { path = "../../juniper_warp", features = ["subscriptions"] }

examples/warp_subscriptions/src/main.rs

@@ -2,10 +2,10 @@
 
 use std::{env, pin::Pin, sync::Arc, time::Duration};
 
-use futures::{Future, FutureExt as _, Stream};
+use futures::{FutureExt as _, Stream};
 use juniper::{DefaultScalarValue, EmptyMutation, FieldError, RootNode};
-use juniper_subscriptions::Coordinator;
-use juniper_warp::{playground_filter, subscriptions::graphql_subscriptions};
+use juniper_graphql_ws::ConnectionConfig;
+use juniper_warp::{playground_filter, subscriptions::serve_graphql_ws};
 use warp::{http::Response, Filter};
 
 #[derive(Clone)]
@@ -151,30 +151,24 @@ async fn main() {
     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()));
+    let root_node = Arc::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)
-                        .map(|r| {
-                            if let Err(e) = r {
-                                println!("Websocket error: {}", e);
-                            }
-                        })
-                        .boxed()
-                })
-            },
-        ))
+        .map(move |ws: warp::ws::Ws| {
+            let root_node = root_node.clone();
+            ws.on_upgrade(move |websocket| async move {
+                serve_graphql_ws(websocket, root_node, ConnectionConfig::new(Context {}))
+                    .map(|r| {
+                        if let Err(e) = r {
+                            println!("Websocket error: {}", e);
+                        }
+                    })
+                    .await
+            })
+        }))
     .map(|reply| {
         // TODO#584: remove this workaround
         warp::reply::with_header(reply, "Sec-WebSocket-Protocol", "graphql-ws")

juniper/release.toml

@@ -30,6 +30,8 @@ pre-release-replacements = [
   {file="../juniper_warp/Cargo.toml", search="\\[dev-dependencies\\.juniper\\]\nversion = \"[^\"]+\"", replace="[dev-dependencies.juniper]\nversion = \"{{version}}\""},
   # Subscriptions
   {file="../juniper_subscriptions/Cargo.toml", search="juniper = \\{ version = \"[^\"]+\"", replace="juniper = { version = \"{{version}}\""},
+  # GraphQL-WS
+  {file="../juniper_graphql_ws/Cargo.toml", search="juniper = \\{ version = \"[^\"]+\"", replace="juniper = { version = \"{{version}}\""},
   # Actix-Web
   {file="../juniper_actix/Cargo.toml", search="juniper = \\{ version = \"[^\"]+\"", replace="juniper = { version = \"{{version}}\""},
   {file="../juniper_actix/Cargo.toml", search="\\[dev-dependencies\\.juniper\\]\nversion = \"[^\"]+\"", replace="[dev-dependencies.juniper]\nversion = \"{{version}}\""},

juniper_graphql_ws/CHANGELOG.md

@@ -0,0 +1,3 @@
+# master
+
+- Initial Release

juniper_graphql_ws/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "juniper_graphql_ws"
+version = "0.1.0"
+authors = ["Christopher Brown <[email protected]>"]
+license = "BSD-2-Clause"
+description = "Graphql-ws protocol implementation for Juniper"
+documentation = "https://docs.rs/juniper_graphql_ws"
+repository = "https://github.com/graphql-rust/juniper"
+keywords = ["graphql-ws", "juniper", "graphql", "apollo"]
+edition = "2018"
+
+[dependencies]
+juniper = { version = "0.14.2", path = "../juniper", default-features = false }
+juniper_subscriptions = { path = "../juniper_subscriptions" }
+serde = { version = "1.0.8", features = ["derive"] }
+tokio = { version = "0.2", features = ["macros", "rt-core", "time"] }
+
+[dev-dependencies]
+serde_json = { version = "1.0.2" }

juniper_graphql_ws/Makefile.toml

@@ -0,0 +1,20 @@
+[env]
+CARGO_MAKE_CARGO_ALL_FEATURES = ""
+
+[tasks.build-verbose]
+condition = { rust_version = { min = "1.29.0" } }
+
+[tasks.build-verbose.windows]
+condition = { rust_version = { min = "1.29.0" }, env = { "TARGET" = "x86_64-pc-windows-msvc" } }
+
+[tasks.test-verbose]
+condition = { rust_version = { min = "1.29.0" } }
+
+[tasks.test-verbose.windows]
+condition = { rust_version = { min = "1.29.0" }, env = { "TARGET" = "x86_64-pc-windows-msvc" } }
+
+[tasks.ci-coverage-flow]
+condition = { rust_version = { min = "1.29.0" } }
+
+[tasks.ci-coverage-flow.windows]
+disabled = true

juniper_graphql_ws/release.toml

@@ -0,0 +1,8 @@
+no-dev-version = true
+pre-release-commit-message = "Release {{crate_name}} {{version}}"
+pro-release-commit-message = "Bump {{crate_name}} version to {{next_version}}"
+tag-message = "Release {{crate_name}} {{version}}"
+upload-doc = false
+pre-release-replacements = [
+  {file="src/lib.rs", search="docs.rs/juniper_graphql_ws/[a-z0-9\\.-]+", replace="docs.rs/juniper_graphql_ws/{{version}}"},
+]

juniper_graphql_ws/src/client_message.rs

@@ -0,0 +1,131 @@
+use juniper::{ScalarValue, Variables};
+
+/// The payload for a client's "start" message. This triggers execution of a query, mutation, or
+/// subscription.
+#[derive(Debug, Deserialize, PartialEq)]
+#[serde(bound(deserialize = "S: ScalarValue"))]
+#[serde(rename_all = "camelCase")]
+pub struct StartPayload<S: ScalarValue> {
+    /// The document body.
+    pub query: String,
+
+    /// The optional variables.
+    #[serde(default)]
+    pub variables: Variables<S>,
+
+    /// The optional operation name (required if the document contains multiple operations).
+    pub operation_name: Option<String>,
+}
+
+/// ClientMessage defines the message types that clients can send.
+#[derive(Debug, Deserialize, PartialEq)]
+#[serde(bound(deserialize = "S: ScalarValue"))]
+#[serde(rename_all = "snake_case")]
+#[serde(tag = "type")]
+pub enum ClientMessage<S: ScalarValue> {
+    /// ConnectionInit is sent by the client upon connecting.
+    ConnectionInit {
+        /// Optional parameters of any type sent from the client. These are often used for
+        /// authentication.
+        #[serde(default)]
+        payload: Variables<S>,
+    },
+    /// Start messages are used to execute a GraphQL operation.
+    Start {
+        /// The id of the operation. This can be anything, but must be unique. If there are other
+        /// in-flight operations with the same id, the message will be ignored or cause an error.
+        id: String,
+
+        /// The query, variables, and operation name.
+        payload: StartPayload<S>,
+    },
+    /// Stop messages are used to unsubscribe from a subscription.
+    Stop {
+        /// The id of the operation to stop.
+        id: String,
+    },
+    /// ConnectionTerminate is used to terminate the connection.
+    ConnectionTerminate,
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use juniper::{DefaultScalarValue, InputValue};
+
+    #[test]
+    fn test_deserialization() {
+        type ClientMessage = super::ClientMessage<DefaultScalarValue>;
+
+        assert_eq!(
+            ClientMessage::ConnectionInit {
+                payload: [("foo".to_string(), InputValue::scalar("bar"))]
+                    .iter()
+                    .cloned()
+                    .collect(),
+            },
+            serde_json::from_str(r##"{"type": "connection_init", "payload": {"foo": "bar"}}"##)
+                .unwrap(),
+        );
+
+        assert_eq!(
+            ClientMessage::ConnectionInit {
+                payload: Variables::default(),
+            },
+            serde_json::from_str(r##"{"type": "connection_init"}"##).unwrap(),
+        );
+
+        assert_eq!(
+            ClientMessage::Start {
+                id: "foo".to_string(),
+                payload: StartPayload {
+                    query: "query MyQuery { __typename }".to_string(),
+                    variables: [("foo".to_string(), InputValue::scalar("bar"))]
+                        .iter()
+                        .cloned()
+                        .collect(),
+                    operation_name: Some("MyQuery".to_string()),
+                },
+            },
+            serde_json::from_str(
+                r##"{"type": "start", "id": "foo", "payload": {
+                "query": "query MyQuery { __typename }",
+                "variables": {
+                    "foo": "bar"
+                },
+                "operationName": "MyQuery"
+            }}"##
+            )
+            .unwrap(),
+        );
+
+        assert_eq!(
+            ClientMessage::Start {
+                id: "foo".to_string(),
+                payload: StartPayload {
+                    query: "query MyQuery { __typename }".to_string(),
+                    variables: Variables::default(),
+                    operation_name: None,
+                },
+            },
+            serde_json::from_str(
+                r##"{"type": "start", "id": "foo", "payload": {
+                "query": "query MyQuery { __typename }"
+            }}"##
+            )
+            .unwrap(),
+        );
+
+        assert_eq!(
+            ClientMessage::Stop {
+                id: "foo".to_string()
+            },
+            serde_json::from_str(r##"{"type": "stop", "id": "foo"}"##).unwrap(),
+        );
+
+        assert_eq!(
+            ClientMessage::ConnectionTerminate,
+            serde_json::from_str(r##"{"type": "connection_terminate"}"##).unwrap(),
+        );
+    }
+}