Fix potential security issues with subscription connections.

src/lib.rs

@@ -151,8 +151,7 @@ pub use registry::CacheControl;
 pub use scalars::{Any, Json, ID};
 pub use schema::{Schema, SchemaBuilder, SchemaEnv};
 pub use subscription::{
-    SimpleBroker, SubscriptionStream, SubscriptionStreams, SubscriptionTransport,
-    WebSocketTransport,
+    SimpleBroker, SubscriptionStreams, SubscriptionTransport, WebSocketTransport,
 };
 pub use types::{
     Connection, Cursor, DataSource, Deferred, EmptyEdgeFields, EmptyMutation, EmptySubscription,

src/schema.rs

@@ -9,7 +9,7 @@ use crate::types::QueryRoot;
 use crate::validation::{check_rules, CheckResult, ValidationMode};
 use crate::{
     CacheControl, Error, ObjectType, Pos, QueryEnv, QueryError, QueryResponse, Result,
-    SubscriptionStream, SubscriptionType, Type, Variables, ID,
+    SubscriptionType, Type, Variables, ID,
 };
 use async_graphql_parser::query::{Document, OperationType};
 use bytes::Bytes;
@@ -400,7 +400,7 @@ where
         transport: T,
     ) -> (
         mpsc::UnboundedSender<Bytes>,
-        SubscriptionStream<Query, Mutation, Subscription, T>,
+        impl Stream<Item = Bytes> + Unpin,
     ) {
         create_connection(self.clone(), transport)
     }

src/subscription/connection.rs

@@ -57,7 +57,7 @@ pub fn create_connection<Query, Mutation, Subscription, T: SubscriptionTransport
     transport: T,
 ) -> (
     mpsc::UnboundedSender<Bytes>,
-    SubscriptionStream<Query, Mutation, Subscription, T>,
+    impl Stream<Item = Bytes> + Unpin,
 )
 where
     Query: ObjectType + Sync + Send + 'static,
@@ -67,7 +67,7 @@ where
     let (tx_bytes, rx_bytes) = mpsc::unbounded();
     (
         tx_bytes,
-        SubscriptionStream {
+        Box::pin(SubscriptionStream {
             schema,
             transport,
             streams: SubscriptionStreams {
@@ -76,7 +76,7 @@ where
             rx_bytes,
             handle_request_fut: None,
             waker: AtomicWaker::new(),
-        },
+        }),
     )
 }
 
@@ -90,7 +90,7 @@ type HandleRequestBoxFut<T> = Pin<
 
 #[allow(missing_docs)]
 #[allow(clippy::type_complexity)]
-pub struct SubscriptionStream<Query, Mutation, Subscription, T: SubscriptionTransport> {
+struct SubscriptionStream<Query, Mutation, Subscription, T: SubscriptionTransport> {
     schema: Schema<Query, Mutation, Subscription>,
     transport: T,
     streams: SubscriptionStreams,

src/subscription/mod.rs

@@ -3,9 +3,7 @@ mod simple_broker;
 mod subscription_type;
 mod ws_transport;
 
-pub use connection::{
-    create_connection, SubscriptionStream, SubscriptionStreams, SubscriptionTransport,
-};
+pub use connection::{create_connection, SubscriptionStreams, SubscriptionTransport};
 pub use simple_broker::SimpleBroker;
 pub use subscription_type::{create_subscription_stream, SubscriptionType};
 pub use ws_transport::WebSocketTransport;