From 2540e871582e247f564b40d29aa2ba5beb2e48e5 Mon Sep 17 00:00:00 2001
From: BratSinot <bratsinot@gmail.com>
Date: Mon, 31 May 2021 13:55:08 +0300
Subject: [PATCH 1/3] WebSocket is now generic in graphql_subscription_upgrade
 functions.

---
 integrations/warp/src/subscription.rs | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/integrations/warp/src/subscription.rs b/integrations/warp/src/subscription.rs
index f6bb35df..fc12bf35 100644
--- a/integrations/warp/src/subscription.rs
+++ b/integrations/warp/src/subscription.rs
@@ -5,7 +5,6 @@ use async_graphql::http::{WebSocketProtocols, WsMessage};
 use async_graphql::{Data, ObjectType, Result, Schema, SubscriptionType};
 use futures_util::{future, StreamExt};
 use warp::filters::ws;
-use warp::ws::WebSocket;
 use warp::{Filter, Rejection, Reply};
 
 /// GraphQL subscription filter
@@ -165,14 +164,15 @@ pub fn graphql_protocol() -> impl Filter<Extract = (WebSocketProtocols,), Error
 ///     warp::serve(filter).run(([0, 0, 0, 0], 8000)).await;
 /// });
 /// ```
-pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription>(
-    websocket: WebSocket,
+pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription, S>(
+    websocket: S,
     protocol: WebSocketProtocols,
     schema: Schema<Query, Mutation, Subscription>,
 ) where
     Query: ObjectType + 'static,
     Mutation: ObjectType + 'static,
     Subscription: SubscriptionType + 'static,
+    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>> + futures_util::Sink<warp::ws::Message>,
 {
     graphql_subscription_upgrade_with_data(websocket, protocol, schema, |_| async {
         Ok(Default::default())
@@ -183,8 +183,8 @@ pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription>(
 /// Handle the WebSocket subscription.
 ///
 /// Specifies that a function converts the init payload to data.
-pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscription, F, R>(
-    websocket: WebSocket,
+pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscription, F, R, S>(
+    websocket: S,
     protocol: WebSocketProtocols,
     schema: Schema<Query, Mutation, Subscription>,
     initializer: F,
@@ -194,6 +194,7 @@ pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscriptio
     Subscription: SubscriptionType + 'static,
     F: FnOnce(serde_json::Value) -> R + Clone + Send + 'static,
     R: Future<Output = Result<Data>> + Send + 'static,
+    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>> + futures_util::Sink<warp::ws::Message>,
 {
     let (ws_sender, ws_receiver) = websocket.split();
     let _ = async_graphql::http::WebSocket::with_data(

From ede6c95e73932b1385e03e0e6bdf1150b650e726 Mon Sep 17 00:00:00 2001
From: BratSinot <bratsinot@gmail.com>
Date: Mon, 31 May 2021 13:59:10 +0300
Subject: [PATCH 2/3] cargo fmt

---
 integrations/warp/src/subscription.rs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/integrations/warp/src/subscription.rs b/integrations/warp/src/subscription.rs
index fc12bf35..d5f00bf8 100644
--- a/integrations/warp/src/subscription.rs
+++ b/integrations/warp/src/subscription.rs
@@ -172,7 +172,8 @@ pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription, S>(
     Query: ObjectType + 'static,
     Mutation: ObjectType + 'static,
     Subscription: SubscriptionType + 'static,
-    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>> + futures_util::Sink<warp::ws::Message>,
+    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>>
+        + futures_util::Sink<warp::ws::Message>,
 {
     graphql_subscription_upgrade_with_data(websocket, protocol, schema, |_| async {
         Ok(Default::default())
@@ -194,7 +195,8 @@ pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscriptio
     Subscription: SubscriptionType + 'static,
     F: FnOnce(serde_json::Value) -> R + Clone + Send + 'static,
     R: Future<Output = Result<Data>> + Send + 'static,
-    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>> + futures_util::Sink<warp::ws::Message>,
+    S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>>
+        + futures_util::Sink<warp::ws::Message>,
 {
     let (ws_sender, ws_receiver) = websocket.split();
     let _ = async_graphql::http::WebSocket::with_data(

From 484a04b5db3670417ffd33fe92b16ff4ecab2fcd Mon Sep 17 00:00:00 2001
From: BratSinot <bratsinot@gmail.com>
Date: Mon, 31 May 2021 14:53:00 +0300
Subject: [PATCH 3/3] Removed Copy trait from initializer in
 graphql_subscription_with_data.

---
 integrations/warp/src/subscription.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/integrations/warp/src/subscription.rs b/integrations/warp/src/subscription.rs
index d5f00bf8..ef23dc0a 100644
--- a/integrations/warp/src/subscription.rs
+++ b/integrations/warp/src/subscription.rs
@@ -193,7 +193,7 @@ pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscriptio
     Query: ObjectType + 'static,
     Mutation: ObjectType + 'static,
     Subscription: SubscriptionType + 'static,
-    F: FnOnce(serde_json::Value) -> R + Clone + Send + 'static,
+    F: FnOnce(serde_json::Value) -> R + Send + 'static,
     R: Future<Output = Result<Data>> + Send + 'static,
     S: futures_util::Stream<Item = Result<warp::ws::Message, warp::Error>>
         + futures_util::Sink<warp::ws::Message>,