Repository: graphql-rust/juniper (language: rust)

Issue:
juniper_warp::subscriptions rewrite wishlist
The `juniper_warp::subscriptions` module has been pretty neglected and has enough shortcomings to warrant a complete rewrite imo.

Things that should be improved in such a rewrite:

* `connection_init` handling should parse parameters sent from the client, and these should be exposed in a way that allows users of the library to perform authentication based on them. (Related: https://github.com/graphql-rust/juniper/pull/597)
* The server should respond to `connection_init` with GQL_CONNECTION_ACK + GQL_CONNECTION_KEEP_ALIVE or GQL_CONNECTION_ERROR (e.g. if authentication failed).
* It should be possible to execute queries and mutations over the connection, in addition to subscriptions.
* The server should send periodic keep-alives to the client (possibly with a configurable interval).
* Stopping a subscription should have more well-defined timing characteristics. For example, the values stream should be dropped immediately, and we shouldn't send the GQL_COMPLETE message until we can guarantee that no more events will be sent for that subscription.
* It should probably be possible for the bulk of the code to be framework-agnostic. I.e. the graphql-ws protocol implementation should be usable with frameworks other than Warp.
* The module needs extensive tests.

I think I'm pretty likely to attempt a rewrite if no one else beats me to it, but regardless of who does it, these are the things I would say need to be addressed.


Discussion / hints:
Starting on this now. Plan is to introduce a new crate that implements the graphql-ws protocol in a reusable way. Then juniper_warp, juniper_rocket, etc. can use that to provide subscription functionality with just a few lines of code.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 79429a10e..d37670be7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,6 +15,7 @@ members = [
   "juniper_rocket",
   "juniper_rocket_async",
   "juniper_subscriptions",
+  "juniper_graphql_ws",
   "juniper_warp",
   "juniper_actix",
 ]
diff --git a/examples/warp_subscriptions/Cargo.toml b/examples/warp_subscriptions/Cargo.toml
index 7fc8fb4a1..5c69129eb 100644
--- a/examples/warp_subscriptions/Cargo.toml
+++ b/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"] }
diff --git a/examples/warp_subscriptions/src/main.rs b/examples/warp_subscriptions/src/main.rs
index f0f9f7373..0d4f31a67 100644
--- a/examples/warp_subscriptions/src/main.rs
+++ b/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")
diff --git a/juniper/release.toml b/juniper/release.toml
index 723911494..ab15f4e50 100644
--- a/juniper/release.toml
+++ b/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}}\""},
diff --git a/juniper_graphql_ws/CHANGELOG.md b/juniper_graphql_ws/CHANGELOG.md
new file mode 100644
index 000000000..052324725
--- /dev/null
+++ b/juniper_graphql_ws/CHANGELOG.md
@@ -0,0 +1,3 @@
+# master
+
+- Initial Release
diff --git a/juniper_graphql_ws/Cargo.toml b/juniper_graphql_ws/Cargo.toml
new file mode 100644
index 000000000..8bf19ee76
--- /dev/null
+++ b/juniper_graphql_ws/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+name = "juniper_graphql_ws"
+version = "0.1.0"
+authors = ["Christopher Brown <ccbrown112@gmail.com>"]
+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" }
diff --git a/juniper_graphql_ws/Makefile.toml b/juniper_graphql_ws/Makefile.toml
new file mode 100644
index 000000000..ba858470d
--- /dev/null
+++ b/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
diff --git a/juniper_graphql_ws/release.toml b/juniper_graphql_ws/release.toml
new file mode 100644
index 000000000..98e705946
--- /dev/null
+++ b/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}}"},
+]
diff --git a/juniper_graphql_ws/src/client_message.rs b/juniper_graphql_ws/src/client_message.rs
new file mode 100644
index 000000000..1e20caef1
--- /dev/null
+++ b/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(),
+        );
+    }
+}
diff --git a/juniper_graphql_ws/src/lib.rs b/juniper_graphql_ws/src/lib.rs
new file mode 100644
index 000000000..7cf73eecb
--- /dev/null
+++ b/juniper_graphql_ws/src/lib.rs
@@ -0,0 +1,1073 @@
+/*!
+
+# juniper_graphql_ws
+
+This crate contains an implementation of the [graphql-ws protocol](https://github.com/apollographql/subscriptions-transport-ws/blob/263844b5c1a850c1e29814564eb62cb587e5eaaf/PROTOCOL.md), as used by Apollo.
+
+*/
+
+#![deny(missing_docs)]
+#![deny(warnings)]
+
+#[macro_use]
+extern crate serde;
+
+mod client_message;
+pub use client_message::*;
+
+mod server_message;
+pub use server_message::*;
+
+mod schema;
+pub use schema::*;
+
+use juniper::{
+    futures::{
+        channel::oneshot,
+        future::{self, BoxFuture, Either, Future, FutureExt, TryFutureExt},
+        stream::{self, BoxStream, SelectAll, StreamExt},
+        task::{Context, Poll, Waker},
+        Sink, Stream,
+    },
+    GraphQLError, RuleError, ScalarValue, Variables,
+};
+use std::{
+    collections::HashMap,
+    convert::{Infallible, TryInto},
+    error::Error,
+    marker::PhantomPinned,
+    pin::Pin,
+    sync::Arc,
+    time::Duration,
+};
+
+struct ExecutionParams<S: Schema> {
+    start_payload: StartPayload<S::ScalarValue>,
+    config: Arc<ConnectionConfig<S::Context>>,
+    schema: S,
+}
+
+/// ConnectionConfig is used to configure the connection once the client sends the ConnectionInit
+/// message.
+pub struct ConnectionConfig<CtxT> {
+    context: CtxT,
+    max_in_flight_operations: usize,
+    keep_alive_interval: Duration,
+}
+
+impl<CtxT> ConnectionConfig<CtxT> {
+    /// Constructs the configuration required for a connection to be accepted.
+    pub fn new(context: CtxT) -> Self {
+        Self {
+            context,
+            max_in_flight_operations: 0,
+            keep_alive_interval: Duration::from_secs(30),
+        }
+    }
+
+    /// Specifies the maximum number of in-flight operations that a connection can have. If this
+    /// number is exceeded, attempting to start more will result in an error. By default, there is
+    /// no limit to in-flight operations.
+    pub fn with_max_in_flight_operations(mut self, max: usize) -> Self {
+        self.max_in_flight_operations = max;
+        self
+    }
+
+    /// Specifies the interval at which to send keep-alives. Specifying a zero duration will
+    /// disable keep-alives. By default, keep-alives are sent every
+    /// 30 seconds.
+    pub fn with_keep_alive_interval(mut self, interval: Duration) -> Self {
+        self.keep_alive_interval = interval;
+        self
+    }
+}
+
+impl<S: ScalarValue, CtxT: Unpin + Send + 'static> Init<S, CtxT> for ConnectionConfig<CtxT> {
+    type Error = Infallible;
+    type Future = future::Ready<Result<Self, Self::Error>>;
+
+    fn init(self, _params: Variables<S>) -> Self::Future {
+        future::ready(Ok(self))
+    }
+}
+
+enum Reaction<S: Schema> {
+    ServerMessage(ServerMessage<S::ScalarValue>),
+    EndStream,
+}
+
+impl<S: Schema> Reaction<S> {
+    /// Converts the reaction into a one-item stream.
+    fn to_stream(self) -> BoxStream<'static, Self> {
+        stream::once(future::ready(self)).boxed()
+    }
+}
+
+/// Init defines the requirements for types that can provide connection configurations when
+/// ConnectionInit messages are received. Implementations are provided for `ConnectionConfig` and
+/// closures that meet the requirements.
+pub trait Init<S: ScalarValue, CtxT>: Unpin + 'static {
+    /// The error that is returned on failure. The formatted error will be used as the contents of
+    /// the "message" field sent back to the client.
+    type Error: Error;
+
+    /// The future configuration type.
+    type Future: Future<Output = Result<ConnectionConfig<CtxT>, Self::Error>> + Send + 'static;
+
+    /// Returns a future for the configuration to use.
+    fn init(self, params: Variables<S>) -> Self::Future;
+}
+
+impl<F, S, CtxT, Fut, E> Init<S, CtxT> for F
+where
+    S: ScalarValue,
+    F: FnOnce(Variables<S>) -> Fut + Unpin + 'static,
+    Fut: Future<Output = Result<ConnectionConfig<CtxT>, E>> + Send + 'static,
+    E: Error,
+{
+    type Error = E;
+    type Future = Fut;
+
+    fn init(self, params: Variables<S>) -> Fut {
+        self(params)
+    }
+}
+
+enum ConnectionState<S: Schema, I: Init<S::ScalarValue, S::Context>> {
+    /// PreInit is the state before a ConnectionInit message has been accepted.
+    PreInit { init: I, schema: S },
+    /// Active is the state after a ConnectionInit message has been accepted.
+    Active {
+        config: Arc<ConnectionConfig<S::Context>>,
+        stoppers: HashMap<String, oneshot::Sender<()>>,
+        schema: S,
+    },
+    /// Terminated is the state after a ConnectionInit message has been rejected.
+    Terminated,
+}
+
+impl<S: Schema, I: Init<S::ScalarValue, S::Context>> ConnectionState<S, I> {
+    // Each message we receive results in a stream of zero or more reactions. For example, a
+    // ConnectionTerminate message results in a one-item stream with the EndStream reaction.
+    async fn handle_message(
+        self,
+        msg: ClientMessage<S::ScalarValue>,
+    ) -> (Self, BoxStream<'static, Reaction<S>>) {
+        if let ClientMessage::ConnectionTerminate = msg {
+            return (self, Reaction::EndStream.to_stream());
+        }
+
+        match self {
+            Self::PreInit { init, schema } => match msg {
+                ClientMessage::ConnectionInit { payload } => match init.init(payload).await {
+                    Ok(config) => {
+                        let keep_alive_interval = config.keep_alive_interval;
+
+                        let mut s = stream::iter(vec![Reaction::ServerMessage(
+                            ServerMessage::ConnectionAck,
+                        )])
+                        .boxed();
+
+                        if keep_alive_interval > Duration::from_secs(0) {
+                            s = s
+                                .chain(
+                                    Reaction::ServerMessage(ServerMessage::ConnectionKeepAlive)
+                                        .to_stream(),
+                                )
+                                .boxed();
+                            s = s
+                                .chain(stream::unfold((), move |_| async move {
+                                    tokio::time::delay_for(keep_alive_interval).await;
+                                    Some((
+                                        Reaction::ServerMessage(ServerMessage::ConnectionKeepAlive),
+                                        (),
+                                    ))
+                                }))
+                                .boxed();
+                        }
+
+                        (
+                            Self::Active {
+                                config: Arc::new(config),
+                                stoppers: HashMap::new(),
+                                schema,
+                            },
+                            s,
+                        )
+                    }
+                    Err(e) => (
+                        Self::Terminated,
+                        stream::iter(vec![
+                            Reaction::ServerMessage(ServerMessage::ConnectionError {
+                                payload: ConnectionErrorPayload {
+                                    message: e.to_string(),
+                                },
+                            }),
+                            Reaction::EndStream,
+                        ])
+                        .boxed(),
+                    ),
+                },
+                _ => (Self::PreInit { init, schema }, stream::empty().boxed()),
+            },
+            Self::Active {
+                config,
+                mut stoppers,
+                schema,
+            } => {
+                let reactions = match msg {
+                    ClientMessage::Start { id, payload } => {
+                        if stoppers.contains_key(&id) {
+                            // We already have an operation with this id, so we can't start a new
+                            // one.
+                            stream::empty().boxed()
+                        } else {
+                            // Go ahead and prune canceled stoppers before adding a new one.
+                            stoppers.retain(|_, tx| !tx.is_canceled());
+
+                            if config.max_in_flight_operations > 0
+                                && stoppers.len() >= config.max_in_flight_operations
+                            {
+                                // Too many in-flight operations. Just send back a validation error.
+                                stream::iter(vec![
+                                    Reaction::ServerMessage(ServerMessage::Error {
+                                        id: id.clone(),
+                                        payload: GraphQLError::ValidationError(vec![
+                                            RuleError::new("Too many in-flight operations.", &[]),
+                                        ])
+                                        .into(),
+                                    }),
+                                    Reaction::ServerMessage(ServerMessage::Complete { id }),
+                                ])
+                                .boxed()
+                            } else {
+                                // Create a channel that we can use to cancel the operation.
+                                let (tx, rx) = oneshot::channel::<()>();
+                                stoppers.insert(id.clone(), tx);
+
+                                // Create the operation stream. This stream will emit Data and Error
+                                // messages, but will not emit Complete – that part is up to us.
+                                let s = Self::start(
+                                    id.clone(),
+                                    ExecutionParams {
+                                        start_payload: payload,
+                                        config: config.clone(),
+                                        schema: schema.clone(),
+                                    },
+                                )
+                                .into_stream()
+                                .flatten();
+
+                                // Combine this with our oneshot channel so that the stream ends if the
+                                // oneshot is ever fired.
+                                let s = stream::unfold((rx, s.boxed()), |(rx, mut s)| async move {
+                                    let next = match future::select(rx, s.next()).await {
+                                        Either::Left(_) => None,
+                                        Either::Right((r, rx)) => r.map(|r| (r, rx)),
+                                    };
+                                    next.map(|(r, rx)| (r, (rx, s)))
+                                });
+
+                                // Once the stream ends, send the Complete message.
+                                let s = s.chain(
+                                    Reaction::ServerMessage(ServerMessage::Complete { id })
+                                        .to_stream(),
+                                );
+
+                                s.boxed()
+                            }
+                        }
+                    }
+                    ClientMessage::Stop { id } => {
+                        stoppers.remove(&id);
+                        stream::empty().boxed()
+                    }
+                    _ => stream::empty().boxed(),
+                };
+                (
+                    Self::Active {
+                        config,
+                        stoppers,
+                        schema,
+                    },
+                    reactions,
+                )
+            }
+            Self::Terminated => (self, stream::empty().boxed()),
+        }
+    }
+
+    async fn start(id: String, params: ExecutionParams<S>) -> BoxStream<'static, Reaction<S>> {
+        // TODO: This could be made more efficient if juniper exposed functionality to allow us to
+        // parse and validate the query, determine whether it's a subscription, and then execute
+        // it. For now, the query gets parsed and validated twice.
+
+        let params = Arc::new(params);
+
+        // Try to execute this as a query or mutation.
+        match juniper::execute(
+            &params.start_payload.query,
+            params
+                .start_payload
+                .operation_name
+                .as_ref()
+                .map(|s| s.as_str()),
+            params.schema.root_node(),
+            &params.start_payload.variables,
+            &params.config.context,
+        )
+        .await
+        {
+            Ok((data, errors)) => {
+                return Reaction::ServerMessage(ServerMessage::Data {
+                    id: id.clone(),
+                    payload: DataPayload { data, errors },
+                })
+                .to_stream();
+            }
+            Err(GraphQLError::IsSubscription) => {}
+            Err(e) => {
+                return Reaction::ServerMessage(ServerMessage::Error {
+                    id: id.clone(),
+                    // e only references data owned by params. The new ErrorPayload will continue to keep that data alive.
+                    payload: unsafe { ErrorPayload::new_unchecked(Box::new(params.clone()), e) },
+                })
+                .to_stream();
+            }
+        }
+
+        // Try to execute as a subscription.
+        SubscriptionStart::new(id, params.clone()).boxed()
+    }
+}
+
+struct InterruptableStream<S> {
+    stream: S,
+    rx: oneshot::Receiver<()>,
+}
+
+impl<S: Stream + Unpin> Stream for InterruptableStream<S> {
+    type Item = S::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        match Pin::new(&mut self.rx).poll(cx) {
+            Poll::Ready(_) => return Poll::Ready(None),
+            Poll::Pending => {}
+        }
+        Pin::new(&mut self.stream).poll_next(cx)
+    }
+}
+
+/// SubscriptionStartState is the state for a subscription operation.
+enum SubscriptionStartState<S: Schema> {
+    /// Init is the start before being polled for the first time.
+    Init { id: String },
+    /// ResolvingIntoStream is the state after being polled for the first time. In this state,
+    /// we're parsing, validating, and getting the actual event stream.
+    ResolvingIntoStream {
+        id: String,
+        future: BoxFuture<
+            'static,
+            Result<
+                juniper_subscriptions::Connection<'static, S::ScalarValue>,
+                GraphQLError<'static>,
+            >,
+        >,
+    },
+    /// Streaming is the state after we've successfully obtained the event stream for the
+    /// subscription. In this state, we're just forwarding events back to the client.
+    Streaming {
+        id: String,
+        stream: juniper_subscriptions::Connection<'static, S::ScalarValue>,
+    },
+    /// Terminated is the state once we're all done.
+    Terminated,
+}
+
+/// SubscriptionStart is the stream for a subscription operation.
+struct SubscriptionStart<S: Schema> {
+    params: Arc<ExecutionParams<S>>,
+    state: SubscriptionStartState<S>,
+    _marker: PhantomPinned,
+}
+
+impl<S: Schema> SubscriptionStart<S> {
+    fn new(id: String, params: Arc<ExecutionParams<S>>) -> Pin<Box<Self>> {
+        Box::pin(Self {
+            params,
+            state: SubscriptionStartState::Init { id },
+            _marker: PhantomPinned,
+        })
+    }
+}
+
+impl<S: Schema> Stream for SubscriptionStart<S> {
+    type Item = Reaction<S>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        let (params, state) = unsafe {
+            // XXX: The execution parameters are referenced by state and must not be modified.
+            // Modifying state is fine though.
+            let inner = self.get_unchecked_mut();
+            (&inner.params, &mut inner.state)
+        };
+
+        loop {
+            match state {
+                SubscriptionStartState::Init { id } => {
+                    // XXX: resolve_into_stream returns a Future that references the execution
+                    // parameters, and the returned stream also references them. We can guarantee
+                    // that everything has the same lifetime in this self-referential struct.
+                    let params = Arc::as_ptr(params);
+                    *state = SubscriptionStartState::ResolvingIntoStream {
+                        id: id.clone(),
+                        future: unsafe {
+                            juniper::resolve_into_stream(
+                                &(*params).start_payload.query,
+                                (*params)
+                                    .start_payload
+                                    .operation_name
+                                    .as_ref()
+                                    .map(|s| s.as_str()),
+                                (*params).schema.root_node(),
+                                &(*params).start_payload.variables,
+                                &(*params).config.context,
+                            )
+                        }
+                        .map_ok(|(stream, errors)| {
+                            juniper_subscriptions::Connection::from_stream(stream, errors)
+                        })
+                        .boxed(),
+                    };
+                }
+                SubscriptionStartState::ResolvingIntoStream {
+                    ref id,
+                    ref mut future,
+                } => match future.as_mut().poll(cx) {
+                    Poll::Ready(r) => match r {
+                        Ok(stream) => {
+                            *state = SubscriptionStartState::Streaming {
+                                id: id.clone(),
+                                stream,
+                            }
+                        }
+                        Err(e) => {
+                            return Poll::Ready(Some(Reaction::ServerMessage(
+                                ServerMessage::Error {
+                                    id: id.clone(),
+                                    // e only references data owned by params. The new ErrorPayload will continue to keep that data alive.
+                                    payload: unsafe {
+                                        ErrorPayload::new_unchecked(Box::new(params.clone()), e)
+                                    },
+                                },
+                            )));
+                        }
+                    },
+                    Poll::Pending => return Poll::Pending,
+                },
+                SubscriptionStartState::Streaming {
+                    ref id,
+                    ref mut stream,
+                } => match Pin::new(stream).poll_next(cx) {
+                    Poll::Ready(Some(output)) => {
+                        return Poll::Ready(Some(Reaction::ServerMessage(ServerMessage::Data {
+                            id: id.clone(),
+                            payload: DataPayload {
+                                data: output.data,
+                                errors: output.errors,
+                            },
+                        })));
+                    }
+                    Poll::Ready(None) => {
+                        *state = SubscriptionStartState::Terminated;
+                        return Poll::Ready(None);
+                    }
+                    Poll::Pending => return Poll::Pending,
+                },
+                SubscriptionStartState::Terminated => return Poll::Ready(None),
+            }
+        }
+    }
+}
+
+enum ConnectionSinkState<S: Schema, I: Init<S::ScalarValue, S::Context>> {
+    Ready {
+        state: ConnectionState<S, I>,
+    },
+    HandlingMessage {
+        result: BoxFuture<'static, (ConnectionState<S, I>, BoxStream<'static, Reaction<S>>)>,
+    },
+    Closed,
+}
+
+/// Implements the graphql-ws protocol. This is a sink for `TryInto<ClientMessage>` and a stream of
+/// `ServerMessage`.
+pub struct Connection<S: Schema, I: Init<S::ScalarValue, S::Context>> {
+    reactions: SelectAll<BoxStream<'static, Reaction<S>>>,
+    stream_waker: Option<Waker>,
+    sink_state: ConnectionSinkState<S, I>,
+}
+
+impl<S, I> Connection<S, I>
+where
+    S: Schema,
+    I: Init<S::ScalarValue, S::Context>,
+{
+    /// Creates a new connection, which is a sink for `TryInto<ClientMessage>` and a stream of `ServerMessage`.
+    ///
+    /// The `schema` argument should typically be an `Arc<RootNode<...>>`.
+    ///
+    /// The `init` argument is used to provide the context and additional configuration for
+    /// connections. This can be a `ConnectionConfig` if the context and configuration are already
+    /// known, or it can be a closure that gets executed asynchronously when the client sends the
+    /// ConnectionInit message. Using a closure allows you to perform authentication based on the
+    /// parameters provided by the client.
+    pub fn new(schema: S, init: I) -> Self {
+        Self {
+            reactions: SelectAll::new(),
+            stream_waker: None,
+            sink_state: ConnectionSinkState::Ready {
+                state: ConnectionState::PreInit { init, schema },
+            },
+        }
+    }
+}
+
+impl<S, I, T> Sink<T> for Connection<S, I>
+where
+    T: TryInto<ClientMessage<S::ScalarValue>>,
+    T::Error: Error,
+    S: Schema,
+    I: Init<S::ScalarValue, S::Context> + Send,
+{
+    type Error = Infallible;
+
+    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
+        match &mut self.sink_state {
+            ConnectionSinkState::Ready { .. } => Poll::Ready(Ok(())),
+            ConnectionSinkState::HandlingMessage { ref mut result } => {
+                match Pin::new(result).poll(cx) {
+                    Poll::Ready((state, reactions)) => {
+                        self.reactions.push(reactions);
+                        self.sink_state = ConnectionSinkState::Ready { state };
+                        Poll::Ready(Ok(()))
+                    }
+                    Poll::Pending => Poll::Pending,
+                }
+            }
+            ConnectionSinkState::Closed => panic!("poll_ready called after close"),
+        }
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+        let s = self.get_mut();
+        let state = &mut s.sink_state;
+        *state = match std::mem::replace(state, ConnectionSinkState::Closed) {
+            ConnectionSinkState::Ready { state } => {
+                match item.try_into() {
+                    Ok(msg) => ConnectionSinkState::HandlingMessage {
+                        result: state.handle_message(msg).boxed(),
+                    },
+                    Err(e) => {
+                        // If we weren't able to parse the message, send back an error.
+                        s.reactions.push(
+                            Reaction::ServerMessage(ServerMessage::ConnectionError {
+                                payload: ConnectionErrorPayload {
+                                    message: e.to_string(),
+                                },
+                            })
+                            .to_stream(),
+                        );
+                        ConnectionSinkState::Ready { state }
+                    }
+                }
+            }
+            _ => panic!("start_send called when not ready"),
+        };
+        Ok(())
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
+        <Self as Sink<T>>::poll_ready(self, cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
+        self.sink_state = ConnectionSinkState::Closed;
+        if let Some(waker) = self.stream_waker.take() {
+            // Wake up the stream so it can close too.
+            waker.wake();
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<S, I> Stream for Connection<S, I>
+where
+    S: Schema,
+    I: Init<S::ScalarValue, S::Context>,
+{
+    type Item = ServerMessage<S::ScalarValue>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        self.stream_waker = Some(cx.waker().clone());
+
+        if let ConnectionSinkState::Closed = self.sink_state {
+            return Poll::Ready(None);
+        }
+
+        // Poll the reactions for new outgoing messages.
+        loop {
+            if !self.reactions.is_empty() {
+                match Pin::new(&mut self.reactions).poll_next(cx) {
+                    Poll::Ready(Some(reaction)) => match reaction {
+                        Reaction::ServerMessage(msg) => return Poll::Ready(Some(msg)),
+                        Reaction::EndStream => return Poll::Ready(None),
+                    },
+                    Poll::Ready(None) => {
+                        // In rare cases, the reaction stream may terminate. For example, this will
+                        // happen if the first message we receive does not require any reaction. Just
+                        // recreate it in that case.
+                        self.reactions = SelectAll::new();
+                        return Poll::Pending;
+                    }
+                    Poll::Pending => return Poll::Pending,
+                }
+            } else {
+                return Poll::Pending;
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use juniper::{
+        futures::sink::SinkExt,
+        parser::{ParseError, Spanning, Token},
+        DefaultScalarValue, EmptyMutation, FieldError, FieldResult, InputValue, RootNode, Value,
+    };
+    use std::{convert::Infallible, io};
+
+    struct Context(i32);
+
+    struct Query;
+
+    #[juniper::graphql_object(Context=Context)]
+    impl Query {
+        /// context just resolves to the current context.
+        async fn context(context: &Context) -> i32 {
+            context.0
+        }
+    }
+
+    struct Subscription;
+
+    #[juniper::graphql_subscription(Context=Context)]
+    impl Subscription {
+        /// never never emits anything.
+        async fn never(context: &Context) -> BoxStream<'static, FieldResult<i32>> {
+            tokio::time::delay_for(Duration::from_secs(10000))
+                .map(|_| unreachable!())
+                .into_stream()
+                .boxed()
+        }
+
+        /// context emits the current context once, then never emits anything else.
+        async fn context(context: &Context) -> BoxStream<'static, FieldResult<i32>> {
+            stream::once(future::ready(Ok(context.0)))
+                .chain(
+                    tokio::time::delay_for(Duration::from_secs(10000))
+                        .map(|_| unreachable!())
+                        .into_stream(),
+                )
+                .boxed()
+        }
+
+        /// error emits an error once, then never emits anything else.
+        async fn error(context: &Context) -> BoxStream<'static, FieldResult<i32>> {
+            stream::once(future::ready(Err(FieldError::new(
+                "field error",
+                Value::null(),
+            ))))
+            .chain(
+                tokio::time::delay_for(Duration::from_secs(10000))
+                    .map(|_| unreachable!())
+                    .into_stream(),
+            )
+            .boxed()
+        }
+    }
+
+    type ClientMessage = super::ClientMessage<DefaultScalarValue>;
+    type ServerMessage = super::ServerMessage<DefaultScalarValue>;
+
+    fn new_test_schema() -> Arc<RootNode<'static, Query, EmptyMutation<Context>, Subscription>> {
+        Arc::new(RootNode::new(Query, EmptyMutation::new(), Subscription))
+    }
+
+    #[tokio::test]
+    async fn test_query() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_secs(0)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "{context}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(
+            ServerMessage::Data {
+                id: "foo".to_string(),
+                payload: DataPayload {
+                    data: Value::Object(
+                        [("context", Value::Scalar(DefaultScalarValue::Int(1)))]
+                            .iter()
+                            .cloned()
+                            .collect()
+                    ),
+                    errors: vec![],
+                },
+            },
+            conn.next().await.unwrap()
+        );
+
+        assert_eq!(
+            ServerMessage::Complete {
+                id: "foo".to_string(),
+            },
+            conn.next().await.unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_subscriptions() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_secs(0)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "subscription Foo {context}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(
+            ServerMessage::Data {
+                id: "foo".to_string(),
+                payload: DataPayload {
+                    data: Value::Object([("context", Value::scalar(1))].iter().cloned().collect()),
+                    errors: vec![],
+                },
+            },
+            conn.next().await.unwrap()
+        );
+
+        conn.send(ClientMessage::Start {
+            id: "bar".to_string(),
+            payload: StartPayload {
+                query: "subscription Bar {context}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(
+            ServerMessage::Data {
+                id: "bar".to_string(),
+                payload: DataPayload {
+                    data: Value::Object([("context", Value::scalar(1))].iter().cloned().collect()),
+                    errors: vec![],
+                },
+            },
+            conn.next().await.unwrap()
+        );
+
+        conn.send(ClientMessage::Stop {
+            id: "foo".to_string(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(
+            ServerMessage::Complete {
+                id: "foo".to_string(),
+            },
+            conn.next().await.unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_init_params_ok() {
+        let mut conn = Connection::new(new_test_schema(), |params: Variables| async move {
+            assert_eq!(params.get("foo"), Some(&InputValue::scalar("bar")));
+            Ok(ConnectionConfig::new(Context(1))) as Result<_, Infallible>
+        });
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: [("foo".to_string(), InputValue::scalar("bar".to_string()))]
+                .iter()
+                .cloned()
+                .collect(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_init_params_error() {
+        let mut conn = Connection::new(new_test_schema(), |params: Variables| async move {
+            assert_eq!(params.get("foo"), Some(&InputValue::scalar("bar")));
+            Err(io::Error::new(io::ErrorKind::Other, "init error"))
+        });
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: [("foo".to_string(), InputValue::scalar("bar".to_string()))]
+                .iter()
+                .cloned()
+                .collect(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(
+            ServerMessage::ConnectionError {
+                payload: ConnectionErrorPayload {
+                    message: "init error".to_string(),
+                },
+            },
+            conn.next().await.unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_max_in_flight_operations() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1))
+                .with_keep_alive_interval(Duration::from_secs(0))
+                .with_max_in_flight_operations(1),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "subscription Foo {never}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        conn.send(ClientMessage::Start {
+            id: "bar".to_string(),
+            payload: StartPayload {
+                query: "subscription Bar {never}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        match conn.next().await.unwrap() {
+            ServerMessage::Error { id, .. } => {
+                assert_eq!(id, "bar");
+            }
+            msg @ _ => panic!("expected error, got: {:?}", msg),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_parse_error() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_secs(0)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "asd".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        match conn.next().await.unwrap() {
+            ServerMessage::Error { id, payload } => {
+                assert_eq!(id, "foo");
+                match payload.graphql_error() {
+                    GraphQLError::ParseError(Spanning {
+                        item: ParseError::UnexpectedToken(Token::Name("asd")),
+                        ..
+                    }) => {}
+                    p @ _ => panic!("expected graphql parse error, got: {:?}", p),
+                }
+            }
+            msg @ _ => panic!("expected error, got: {:?}", msg),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_keep_alives() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_millis(20)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        for _ in 0..10 {
+            assert_eq!(
+                ServerMessage::ConnectionKeepAlive,
+                conn.next().await.unwrap()
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_slow_init() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_secs(0)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        // If we send the start message before the init is handled, we should still get results.
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "{context}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        assert_eq!(
+            ServerMessage::Data {
+                id: "foo".to_string(),
+                payload: DataPayload {
+                    data: Value::Object(
+                        [("context", Value::Scalar(DefaultScalarValue::Int(1)))]
+                            .iter()
+                            .cloned()
+                            .collect()
+                    ),
+                    errors: vec![],
+                },
+            },
+            conn.next().await.unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_subscription_field_error() {
+        let mut conn = Connection::new(
+            new_test_schema(),
+            ConnectionConfig::new(Context(1)).with_keep_alive_interval(Duration::from_secs(0)),
+        );
+
+        conn.send(ClientMessage::ConnectionInit {
+            payload: Variables::default(),
+        })
+        .await
+        .unwrap();
+
+        assert_eq!(ServerMessage::ConnectionAck, conn.next().await.unwrap());
+
+        conn.send(ClientMessage::Start {
+            id: "foo".to_string(),
+            payload: StartPayload {
+                query: "subscription Foo {error}".to_string(),
+                variables: Variables::default(),
+                operation_name: None,
+            },
+        })
+        .await
+        .unwrap();
+
+        match conn.next().await.unwrap() {
+            ServerMessage::Data {
+                id,
+                payload: DataPayload { data, errors },
+            } => {
+                assert_eq!(id, "foo");
+                assert_eq!(
+                    data,
+                    Value::Object([("error", Value::null())].iter().cloned().collect())
+                );
+                assert_eq!(errors.len(), 1);
+            }
+            msg @ _ => panic!("expected data, got: {:?}", msg),
+        }
+    }
+}
diff --git a/juniper_graphql_ws/src/schema.rs b/juniper_graphql_ws/src/schema.rs
new file mode 100644
index 000000000..68d282f0b
--- /dev/null
+++ b/juniper_graphql_ws/src/schema.rs
@@ -0,0 +1,131 @@
+use juniper::{GraphQLSubscriptionType, GraphQLTypeAsync, RootNode, ScalarValue};
+use std::sync::Arc;
+
+/// Schema defines the requirements for schemas that can be used for operations. Typically this is
+/// just an `Arc<RootNode<...>>` and you should not have to implement it yourself.
+pub trait Schema: Unpin + Clone + Send + Sync + 'static {
+    /// The context type.
+    type Context: Unpin + Send + Sync;
+
+    /// The scalar value type.
+    type ScalarValue: ScalarValue + Send + Sync;
+
+    /// The query type info.
+    type QueryTypeInfo: Send + Sync;
+
+    /// The query type.
+    type Query: GraphQLTypeAsync<Self::ScalarValue, Context = Self::Context, TypeInfo = Self::QueryTypeInfo>
+        + Send;
+
+    /// The mutation type info.
+    type MutationTypeInfo: Send + Sync;
+
+    /// The mutation type.
+    type Mutation: GraphQLTypeAsync<
+            Self::ScalarValue,
+            Context = Self::Context,
+            TypeInfo = Self::MutationTypeInfo,
+        > + Send;
+
+    /// The subscription type info.
+    type SubscriptionTypeInfo: Send + Sync;
+
+    /// The subscription type.
+    type Subscription: GraphQLSubscriptionType<
+            Self::ScalarValue,
+            Context = Self::Context,
+            TypeInfo = Self::SubscriptionTypeInfo,
+        > + Send;
+
+    /// Returns the root node for the schema.
+    fn root_node(
+        &self,
+    ) -> &RootNode<'static, Self::Query, Self::Mutation, Self::Subscription, Self::ScalarValue>;
+}
+
+/// This exists as a work-around for this issue: https://github.com/rust-lang/rust/issues/64552
+///
+/// It can be used in generators where using Arc directly would result in an error.
+// TODO: Remove this once that issue is resolved.
+#[doc(hidden)]
+pub struct ArcSchema<QueryT, MutationT, SubscriptionT, CtxT, S>(
+    pub Arc<RootNode<'static, QueryT, MutationT, SubscriptionT, S>>,
+)
+where
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Unpin + Send + Sync,
+    S: ScalarValue + Send + Sync + 'static;
+
+impl<QueryT, MutationT, SubscriptionT, CtxT, S> Clone
+    for ArcSchema<QueryT, MutationT, SubscriptionT, CtxT, S>
+where
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Unpin + Send + Sync,
+    S: ScalarValue + Send + Sync + 'static,
+{
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}
+
+impl<QueryT, MutationT, SubscriptionT, CtxT, S> Schema
+    for ArcSchema<QueryT, MutationT, SubscriptionT, CtxT, S>
+where
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Unpin + Send + Sync + 'static,
+    S: ScalarValue + Send + Sync + 'static,
+{
+    type Context = CtxT;
+    type ScalarValue = S;
+    type QueryTypeInfo = QueryT::TypeInfo;
+    type Query = QueryT;
+    type MutationTypeInfo = MutationT::TypeInfo;
+    type Mutation = MutationT;
+    type SubscriptionTypeInfo = SubscriptionT::TypeInfo;
+    type Subscription = SubscriptionT;
+
+    fn root_node(&self) -> &RootNode<'static, QueryT, MutationT, SubscriptionT, S> {
+        &self.0
+    }
+}
+
+impl<QueryT, MutationT, SubscriptionT, CtxT, S> Schema
+    for Arc<RootNode<'static, QueryT, MutationT, SubscriptionT, S>>
+where
+    QueryT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    QueryT::TypeInfo: Send + Sync,
+    MutationT: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+    MutationT::TypeInfo: Send + Sync,
+    SubscriptionT: GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
+    SubscriptionT::TypeInfo: Send + Sync,
+    CtxT: Unpin + Send + Sync,
+    S: ScalarValue + Send + Sync + 'static,
+{
+    type Context = CtxT;
+    type ScalarValue = S;
+    type QueryTypeInfo = QueryT::TypeInfo;
+    type Query = QueryT;
+    type MutationTypeInfo = MutationT::TypeInfo;
+    type Mutation = MutationT;
+    type SubscriptionTypeInfo = SubscriptionT::TypeInfo;
+    type Subscription = SubscriptionT;
+
+    fn root_node(&self) -> &RootNode<'static, QueryT, MutationT, SubscriptionT, S> {
+        self
+    }
+}
diff --git a/juniper_graphql_ws/src/server_message.rs b/juniper_graphql_ws/src/server_message.rs
new file mode 100644
index 000000000..3c3531649
--- /dev/null
+++ b/juniper_graphql_ws/src/server_message.rs
@@ -0,0 +1,191 @@
+use juniper::{ExecutionError, GraphQLError, ScalarValue, Value};
+use serde::{Serialize, Serializer};
+use std::{any::Any, fmt, marker::PhantomPinned};
+
+/// The payload for errors that are not associated with a GraphQL operation.
+#[derive(Debug, Serialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct ConnectionErrorPayload {
+    /// The error message.
+    pub message: String,
+}
+
+/// Sent after execution of an operation. For queries and mutations, this is sent to the client
+/// once. For subscriptions, this is sent for every event in the event stream.
+#[derive(Debug, Serialize, PartialEq)]
+#[serde(bound(serialize = "S: ScalarValue"))]
+#[serde(rename_all = "camelCase")]
+pub struct DataPayload<S> {
+    /// The result data.
+    pub data: Value<S>,
+
+    /// The errors that have occurred during execution. Note that parse and validation errors are
+    /// not included here. They are sent via Error messages.
+    pub errors: Vec<ExecutionError<S>>,
+}
+
+/// A payload for errors that can happen before execution. Errors that happen during execution are
+/// instead sent to the client via `DataPayload`. `ErrorPayload` is a wrapper for an owned
+/// `GraphQLError`.
+// XXX: Think carefully before deriving traits. This is self-referential (error references
+// _execution_params).
+pub struct ErrorPayload {
+    _execution_params: Option<Box<dyn Any + Send>>,
+    error: GraphQLError<'static>,
+    _marker: PhantomPinned,
+}
+
+impl ErrorPayload {
+    /// For this to be okay, the caller must guarantee that the error can only reference data from
+    /// execution_params and that execution_params has not been modified or moved.
+    pub(crate) unsafe fn new_unchecked<'a>(
+        execution_params: Box<dyn Any + Send>,
+        error: GraphQLError<'a>,
+    ) -> Self {
+        Self {
+            _execution_params: Some(execution_params),
+            error: std::mem::transmute(error),
+            _marker: PhantomPinned,
+        }
+    }
+
+    /// Returns the contained GraphQLError.
+    pub fn graphql_error<'a>(&'a self) -> &GraphQLError<'a> {
+        &self.error
+    }
+}
+
+impl fmt::Debug for ErrorPayload {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.error.fmt(f)
+    }
+}
+
+impl PartialEq for ErrorPayload {
+    fn eq(&self, other: &Self) -> bool {
+        self.error.eq(&other.error)
+    }
+}
+
+impl Serialize for ErrorPayload {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.error.serialize(serializer)
+    }
+}
+
+impl From<GraphQLError<'static>> for ErrorPayload {
+    fn from(error: GraphQLError<'static>) -> Self {
+        Self {
+            _execution_params: None,
+            error,
+            _marker: PhantomPinned,
+        }
+    }
+}
+
+/// ServerMessage defines the message types that servers can send.
+#[derive(Debug, Serialize, PartialEq)]
+#[serde(bound(serialize = "S: ScalarValue"))]
+#[serde(rename_all = "snake_case")]
+#[serde(tag = "type")]
+pub enum ServerMessage<S: ScalarValue> {
+    /// ConnectionError is used for errors that are not associated with a GraphQL operation. For
+    /// example, this will be used when:
+    ///
+    ///   * The server is unable to parse a client's message.
+    ///   * The client's initialization parameters are rejected.
+    ConnectionError {
+        /// The error that occurred.
+        payload: ConnectionErrorPayload,
+    },
+    /// ConnectionAck is sent in response to a client's ConnectionInit message if the server accepted a
+    /// connection.
+    ConnectionAck,
+    /// Data contains the result of a query, mutation, or subscription event.
+    Data {
+        /// The id of the operation that the data is for.
+        id: String,
+
+        /// The data and errors that occurred during execution.
+        payload: DataPayload<S>,
+    },
+    /// Error contains an error that occurs before execution, such as validation errors.
+    Error {
+        /// The id of the operation that triggered this error.
+        id: String,
+
+        /// The error(s).
+        payload: ErrorPayload,
+    },
+    /// Complete indicates that no more data will be sent for the given operation.
+    Complete {
+        /// The id of the operation that has completed.
+        id: String,
+    },
+    /// ConnectionKeepAlive is sent periodically after accepting a connection.
+    #[serde(rename = "ka")]
+    ConnectionKeepAlive,
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use juniper::DefaultScalarValue;
+
+    #[test]
+    fn test_serialization() {
+        type ServerMessage = super::ServerMessage<DefaultScalarValue>;
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::ConnectionError {
+                payload: ConnectionErrorPayload {
+                    message: "foo".to_string(),
+                },
+            })
+            .unwrap(),
+            r##"{"type":"connection_error","payload":{"message":"foo"}}"##,
+        );
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::ConnectionAck).unwrap(),
+            r##"{"type":"connection_ack"}"##,
+        );
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::Data {
+                id: "foo".to_string(),
+                payload: DataPayload {
+                    data: Value::null(),
+                    errors: vec![],
+                },
+            })
+            .unwrap(),
+            r##"{"type":"data","id":"foo","payload":{"data":null,"errors":[]}}"##,
+        );
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::Error {
+                id: "foo".to_string(),
+                payload: GraphQLError::UnknownOperationName.into(),
+            })
+            .unwrap(),
+            r##"{"type":"error","id":"foo","payload":[{"message":"Unknown operation"}]}"##,
+        );
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::Complete {
+                id: "foo".to_string(),
+            })
+            .unwrap(),
+            r##"{"type":"complete","id":"foo"}"##,
+        );
+
+        assert_eq!(
+            serde_json::to_string(&ServerMessage::ConnectionKeepAlive).unwrap(),
+            r##"{"type":"ka"}"##,
+        );
+    }
+}
diff --git a/juniper_subscriptions/src/lib.rs b/juniper_subscriptions/src/lib.rs
index 0e78279ca..3418c0551 100644
--- a/juniper_subscriptions/src/lib.rs
+++ b/juniper_subscriptions/src/lib.rs
@@ -222,19 +222,25 @@ where
                 }
 
                 if filled_count == obj_len {
+                    let mut errors = vec![];
                     filled_count = 0;
                     let new_vec = (0..obj_len).map(|_| None).collect::<Vec<_>>();
                     let ready_vec = std::mem::replace(&mut ready_vec, new_vec);
                     let ready_vec_iterator = ready_vec.into_iter().map(|el| {
                         let (name, val) = el.unwrap();
-                        if let Ok(value) = val {
-                            (name, value)
-                        } else {
-                            (name, Value::Null)
+                        match val {
+                            Ok(value) => (name, value),
+                            Err(e) => {
+                                errors.push(e);
+                                (name, Value::Null)
+                            }
                         }
                     });
                     let obj = Object::from_iter(ready_vec_iterator);
-                    Poll::Ready(Some(ExecutionOutput::from_data(Value::Object(obj))))
+                    Poll::Ready(Some(ExecutionOutput {
+                        data: Value::Object(obj),
+                        errors,
+                    }))
                 } else {
                     Poll::Pending
                 }
diff --git a/juniper_warp/Cargo.toml b/juniper_warp/Cargo.toml
index cf14ae322..f2fcb5b50 100644
--- a/juniper_warp/Cargo.toml
+++ b/juniper_warp/Cargo.toml
@@ -9,7 +9,7 @@ repository = "https://github.com/graphql-rust/juniper"
 edition = "2018"
 
 [features]
-subscriptions = ["juniper_subscriptions"]
+subscriptions = ["juniper_graphql_ws"]
 
 [dependencies]
 bytes = "0.5"
@@ -17,7 +17,7 @@ anyhow = "1.0"
 thiserror = "1.0"
 futures = "0.3.1"
 juniper = { version = "0.14.2", path = "../juniper", default-features = false  }
-juniper_subscriptions = { path = "../juniper_subscriptions", optional = true }
+juniper_graphql_ws = { path = "../juniper_graphql_ws", optional = true }
 serde = { version = "1.0.75", features = ["derive"] }
 serde_json = "1.0.24"
 tokio = { version = "0.2", features = ["blocking", "rt-core"] }
diff --git a/juniper_warp/src/lib.rs b/juniper_warp/src/lib.rs
index 291ba011a..06898304f 100644
--- a/juniper_warp/src/lib.rs
+++ b/juniper_warp/src/lib.rs
@@ -393,224 +393,103 @@ fn playground_response(
 /// [1]: https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md
 #[cfg(feature = "subscriptions")]
 pub mod subscriptions {
-    use std::{
-        collections::HashMap,
-        sync::{
-            atomic::{AtomicBool, Ordering},
-            Arc,
+    use juniper::{
+        futures::{
+            future::{self, Either},
+            sink::SinkExt,
+            stream::StreamExt,
         },
+        GraphQLSubscriptionType, GraphQLTypeAsync, RootNode, ScalarValue,
     };
+    use juniper_graphql_ws::{ArcSchema, ClientMessage, Connection, Init};
+    use std::{convert::Infallible, fmt, sync::Arc};
 
-    use anyhow::anyhow;
-    use futures::{channel::mpsc, Future, StreamExt as _, TryFutureExt as _, TryStreamExt as _};
-    use juniper::{http::GraphQLRequest, InputValue, ScalarValue, SubscriptionCoordinator as _};
-    use juniper_subscriptions::Coordinator;
-    use serde::{Deserialize, Serialize};
-    use warp::ws::Message;
-
-    /// Listen to incoming messages and do one of the following:
-    ///  - execute subscription and return values from stream
-    ///  - stop stream and close ws connection
-    #[allow(dead_code)]
-    pub fn graphql_subscriptions<Query, Mutation, Subscription, CtxT, S>(
-        websocket: warp::ws::WebSocket,
-        coordinator: Arc<Coordinator<'static, Query, Mutation, Subscription, CtxT, S>>,
-        context: CtxT,
-    ) -> impl Future<Output = Result<(), anyhow::Error>> + Send
-    where
-        Query: juniper::GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
-        Query::TypeInfo: Send + Sync,
-        Mutation: juniper::GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
-        Mutation::TypeInfo: Send + Sync,
-        Subscription: juniper::GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
-        Subscription::TypeInfo: Send + Sync,
-        CtxT: Send + Sync + 'static,
-        S: ScalarValue + Send + Sync + 'static,
-    {
-        let (sink_tx, sink_rx) = websocket.split();
-        let (ws_tx, ws_rx) = mpsc::unbounded();
-        tokio::task::spawn(
-            ws_rx
-                .take_while(|v: &Option<_>| futures::future::ready(v.is_some()))
-                .map(|x| x.unwrap())
-                .forward(sink_tx),
-        );
+    struct Message(warp::ws::Message);
 
-        let context = Arc::new(context);
-        let got_close_signal = Arc::new(AtomicBool::new(false));
-        let got_close_signal2 = got_close_signal.clone();
+    impl<S: ScalarValue> std::convert::TryFrom<Message> for ClientMessage<S> {
+        type Error = serde_json::Error;
 
-        struct SubscriptionState {
-            should_stop: AtomicBool,
+        fn try_from(msg: Message) -> serde_json::Result<Self> {
+            serde_json::from_slice(msg.0.as_bytes())
         }
-        let subscription_states = HashMap::<String, Arc<SubscriptionState>>::new();
+    }
 
-        sink_rx
-            .map_err(move |e| {
-                got_close_signal2.store(true, Ordering::Relaxed);
-                anyhow!("Websocket error: {}", e)
-            })
-            .try_fold(subscription_states, move |mut subscription_states, msg| {
-                let coordinator = coordinator.clone();
-                let context = context.clone();
-                let got_close_signal = got_close_signal.clone();
-                let ws_tx = ws_tx.clone();
-
-                async move {
-                    if msg.is_close() {
-                        return Ok(subscription_states);
-                    }
-
-                    let msg = msg
-                        .to_str()
-                        .map_err(|_| anyhow!("Non-text messages are not accepted"))?;
-                    let request: WsPayload<S> = serde_json::from_str(msg)
-                        .map_err(|e| anyhow!("Invalid WsPayload: {}", e))?;
-
-                    match request.type_name.as_str() {
-                        "connection_init" => {}
-                        "start" => {
-                            if got_close_signal.load(Ordering::Relaxed) {
-                                return Ok(subscription_states);
-                            }
-
-                            let request_id = request.id.clone().unwrap_or("1".to_owned());
-
-                            if let Some(existing) = subscription_states.get(&request_id) {
-                                existing.should_stop.store(true, Ordering::Relaxed);
-                            }
-                            let state = Arc::new(SubscriptionState {
-                                should_stop: AtomicBool::new(false),
-                            });
-                            subscription_states.insert(request_id.clone(), state.clone());
-
-                            let ws_tx = ws_tx.clone();
-
-                            if let Some(ref payload) = request.payload {
-                                if payload.query.is_none() {
-                                    return Err(anyhow!("Query not found"));
-                                }
-                            } else {
-                                return Err(anyhow!("Payload not found"));
-                            }
-
-                            tokio::task::spawn(async move {
-                                let payload = request.payload.unwrap();
-
-                                let graphql_request = GraphQLRequest::<S>::new(
-                                    payload.query.unwrap(),
-                                    None,
-                                    payload.variables,
-                                );
-
-                                let values_stream = match coordinator
-                                    .subscribe(&graphql_request, &context)
-                                    .await
-                                {
-                                    Ok(s) => s,
-                                    Err(err) => {
-                                        let _ =
-                                            ws_tx.unbounded_send(Some(Ok(Message::text(format!(
-                                                r#"{{"type":"error","id":"{}","payload":{}}}"#,
-                                                request_id,
-                                                serde_json::ser::to_string(&err).unwrap_or(
-                                                    "Error deserializing GraphQLError".to_owned()
-                                                )
-                                            )))));
-
-                                        let close_message = format!(
-                                            r#"{{"type":"complete","id":"{}","payload":null}}"#,
-                                            request_id
-                                        );
-                                        let _ = ws_tx
-                                            .unbounded_send(Some(Ok(Message::text(close_message))));
-                                        // close channel
-                                        let _ = ws_tx.unbounded_send(None);
-                                        return;
-                                    }
-                                };
-
-                                values_stream
-                                    .take_while(move |response| {
-                                        let request_id = request_id.clone();
-                                        let should_stop = state.should_stop.load(Ordering::Relaxed)
-                                            || got_close_signal.load(Ordering::Relaxed);
-                                        if !should_stop {
-                                            let mut response_text = serde_json::to_string(
-                                                &response,
-                                            )
-                                            .unwrap_or("Error deserializing response".to_owned());
-
-                                            response_text = format!(
-                                                r#"{{"type":"data","id":"{}","payload":{} }}"#,
-                                                request_id, response_text
-                                            );
-
-                                            let _ = ws_tx.unbounded_send(Some(Ok(Message::text(
-                                                response_text,
-                                            ))));
-                                        }
-
-                                        async move { !should_stop }
-                                    })
-                                    .for_each(|_| async {})
-                                    .await;
-                            });
-                        }
-                        "stop" => {
-                            let request_id = request.id.unwrap_or("1".to_owned());
-                            if let Some(existing) = subscription_states.get(&request_id) {
-                                existing.should_stop.store(true, Ordering::Relaxed);
-                                subscription_states.remove(&request_id);
-                            }
-
-                            let close_message = format!(
-                                r#"{{"type":"complete","id":"{}","payload":null}}"#,
-                                request_id
-                            );
-                            let _ = ws_tx.unbounded_send(Some(Ok(Message::text(close_message))));
-
-                            // close channel
-                            let _ = ws_tx.unbounded_send(None);
-                        }
-                        _ => {}
-                    }
-
-                    Ok(subscription_states)
-                }
-            })
-            .map_ok(|_| ())
+    /// Errors that can happen while serving a connection.
+    #[derive(Debug)]
+    pub enum Error {
+        /// Errors that can happen in Warp while serving a connection.
+        Warp(warp::Error),
+
+        /// Errors that can happen while serializing outgoing messages. Note that errors that occur
+        /// while deserializing internal messages are handled internally by the protocol.
+        Serde(serde_json::Error),
     }
 
-    #[derive(Deserialize)]
-    #[serde(bound = "GraphQLPayload<S>: Deserialize<'de>")]
-    struct WsPayload<S>
-    where
-        S: ScalarValue + Send + Sync,
-    {
-        id: Option<String>,
-        #[serde(rename(deserialize = "type"))]
-        type_name: String,
-        payload: Option<GraphQLPayload<S>>,
+    impl fmt::Display for Error {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match self {
+                Self::Warp(e) => write!(f, "warp error: {}", e),
+                Self::Serde(e) => write!(f, "serde error: {}", e),
+            }
+        }
     }
 
-    #[derive(Debug, Deserialize)]
-    #[serde(bound = "InputValue<S>: Deserialize<'de>")]
-    struct GraphQLPayload<S>
-    where
-        S: ScalarValue + Send + Sync,
-    {
-        variables: Option<InputValue<S>>,
-        extensions: Option<HashMap<String, String>>,
-        #[serde(rename(deserialize = "operationName"))]
-        operaton_name: Option<String>,
-        query: Option<String>,
+    impl std::error::Error for Error {}
+
+    impl From<warp::Error> for Error {
+        fn from(err: warp::Error) -> Self {
+            Self::Warp(err)
+        }
     }
 
-    #[derive(Serialize)]
-    struct Output {
-        data: String,
-        variables: String,
+    impl From<Infallible> for Error {
+        fn from(_err: Infallible) -> Self {
+            unreachable!()
+        }
+    }
+
+    /// Serves the graphql-ws protocol over a WebSocket connection.
+    ///
+    /// The `init` argument is used to provide the context and additional configuration for
+    /// connections. This can be a `juniper_graphql_ws::ConnectionConfig` if the context and
+    /// configuration are already known, or it can be a closure that gets executed asynchronously
+    /// when the client sends the ConnectionInit message. Using a closure allows you to perform
+    /// authentication based on the parameters provided by the client.
+    pub async fn serve_graphql_ws<Query, Mutation, Subscription, CtxT, S, I>(
+        websocket: warp::ws::WebSocket,
+        root_node: Arc<RootNode<'static, Query, Mutation, Subscription, S>>,
+        init: I,
+    ) -> Result<(), Error>
+    where
+        Query: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+        Query::TypeInfo: Send + Sync,
+        Mutation: GraphQLTypeAsync<S, Context = CtxT> + Send + 'static,
+        Mutation::TypeInfo: Send + Sync,
+        Subscription: GraphQLSubscriptionType<S, Context = CtxT> + Send + 'static,
+        Subscription::TypeInfo: Send + Sync,
+        CtxT: Unpin + Send + Sync + 'static,
+        S: ScalarValue + Send + Sync + 'static,
+        I: Init<S, CtxT> + Send,
+    {
+        let (ws_tx, ws_rx) = websocket.split();
+        let (s_tx, s_rx) = Connection::new(ArcSchema(root_node), init).split();
+
+        let ws_rx = ws_rx.map(|r| r.map(|msg| Message(msg)));
+        let s_rx = s_rx.map(|msg| {
+            serde_json::to_string(&msg)
+                .map(|t| warp::ws::Message::text(t))
+                .map_err(|e| Error::Serde(e))
+        });
+
+        match future::select(
+            ws_rx.forward(s_tx.sink_err_into()),
+            s_rx.forward(ws_tx.sink_err_into()),
+        )
+        .await
+        {
+            Either::Left((r, _)) => r.map_err(|e| e.into()),
+            Either::Right((r, _)) => r,
+        }
     }
 }
 


Test patch: