anna/async-graphql
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #530 from BratSinot/ws_stream_to_generic

Browse Source 
WebSocket is now generic in graphql_subscription_upgrade functions.
This commit is contained in:
Sunli 2021-05-31 20:47:11 +08:00 committed by GitHub
commit a06d418b8e
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
integrations/warp/src/subscription.rs
View File

@ -5,7 +5,6 @@ use async_graphql::http::{WebSocketProtocols, WsMessage};
use async_graphql::{Data, ObjectType, Result, Schema, SubscriptionType}; use async_graphql::{Data, ObjectType, Result, Schema, SubscriptionType};
use futures_util::{future, StreamExt}; use futures_util::{future, StreamExt};
use warp::filters::ws; use warp::filters::ws;
use warp::ws::WebSocket;
use warp::{Filter, Rejection, Reply}; use warp::{Filter, Rejection, Reply};
/// GraphQL subscription filter /// GraphQL subscription filter
@ -165,14 +164,16 @@ pub fn graphql_protocol() -> impl Filter<Extract = (WebSocketProtocols,), Error
/// warp::serve(filter).run(([0, 0, 0, 0], 8000)).await; /// warp::serve(filter).run(([0, 0, 0, 0], 8000)).await;
/// }); /// });
/// ``` /// ```
pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription>( pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription, S>(
websocket: WebSocket, websocket: S,
protocol: WebSocketProtocols, protocol: WebSocketProtocols,
schema: Schema<Query, Mutation, Subscription>, schema: Schema<Query, Mutation, Subscription>,
) where ) where
Query: ObjectType + 'static, Query: ObjectType + 'static,
Mutation: ObjectType + 'static, Mutation: ObjectType + 'static,
Subscription: SubscriptionType + '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 { graphql_subscription_upgrade_with_data(websocket, protocol, schema, |_| async {
Ok(Default::default()) Ok(Default::default())
@ -183,8 +184,8 @@ pub async fn graphql_subscription_upgrade<Query, Mutation, Subscription>(
/// Handle the WebSocket subscription. /// Handle the WebSocket subscription.
/// ///
/// Specifies that a function converts the init payload to data. /// Specifies that a function converts the init payload to data.
pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscription, F, R>( pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscription, F, R, S>(
websocket: WebSocket, websocket: S,
protocol: WebSocketProtocols, protocol: WebSocketProtocols,
schema: Schema<Query, Mutation, Subscription>, schema: Schema<Query, Mutation, Subscription>,
initializer: F, initializer: F,
@ -192,8 +193,10 @@ pub async fn graphql_subscription_upgrade_with_data<Query, Mutation, Subscriptio
Query: ObjectType + 'static, Query: ObjectType + 'static,
Mutation: ObjectType + 'static, Mutation: ObjectType + 'static,
Subscription: SubscriptionType + '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, 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 (ws_sender, ws_receiver) = websocket.split();
let _ = async_graphql::http::WebSocket::with_data( let _ = async_graphql::http::WebSocket::with_data(