Removed Clone trait in Context definition in actix_subscriptions

Author: engylemure

juniper_actix/examples/actix_subscriptions.rs

@@ -3,22 +3,17 @@
 use actix_cors::Cors;
 use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer};
 use futures::Stream;
-use juniper::{DefaultScalarValue, EmptyMutation, FieldError, RootNode};
+use juniper::{
+    tests::{model::Database, schema::Query},
+    DefaultScalarValue, EmptyMutation, FieldError, RootNode,
+};
 use juniper_actix::{
     graphiql_handler as gqli_handler, graphql_handler, playground_handler as play_handler,
     subscriptions::{graphql_subscriptions as sub_handler, EmptySubscriptionHandler},
 };
 use juniper_subscriptions::Coordinator;
 use std::{pin::Pin, time::Duration};
 
-pub struct Query;
-
-#[juniper::graphql_object(Context = Database)]
-impl Query {
-    fn hello_world() -> &str {
-        "Hello World!"
-    }
-}
 type Schema = RootNode<'static, Query, EmptyMutation<Database>, Subscription>;
 type MyCoordinator = Coordinator<
     'static,
@@ -33,23 +28,13 @@ type StringStream = Pin<Box<dyn Stream<Item = Result<String, FieldError>> + Send
 
 struct Subscription;
 
-#[derive(Clone)]
-pub struct Database;
-
-impl juniper::Context for Database {}
-
-impl Database {
-    fn new() -> Self {
-        Self {}
-    }
-}
-
 #[juniper::graphql_subscription(Context = Database)]
 impl Subscription {
     async fn hello_world() -> StringStream {
         let mut counter = 0;
-        let stream = tokio::time::interval(Duration::from_secs(5)).map(move |_| {
+        let stream = tokio::time::interval(Duration::from_secs(1)).map(move |_| {
             counter += 1;
+
             if counter % 2 == 0 {
                 Ok(String::from("World!"))
             } else {
@@ -87,16 +72,8 @@ async fn graphql_subscriptions(
     req: HttpRequest,
 ) -> Result<HttpResponse, Error> {
     let context = Database::new();
-    unsafe {
-        sub_handler(
-            coordinator,
-            context,
-            stream,
-            req,
-            Some(EmptySubscriptionHandler::default()),
-        )
-    }
-    .await
+    let handler: Option<EmptySubscriptionHandler> = None;
+    unsafe { sub_handler(coordinator, context, stream, req, handler) }.await
 }
 
 #[actix_rt::main]

juniper_actix/src/subscriptions.rs

@@ -15,6 +15,7 @@ pub use juniper_subscriptions::ws_util::{
 };
 use juniper_subscriptions::Coordinator;
 use serde::Serialize;
+use std::ops::Deref;
 use std::{
     collections::HashMap,
     error::Error as StdError,
@@ -33,7 +34,7 @@ fn start<Query, Mutation, Subscription, Context, S, SubHandler, T, E>(
 where
     T: Stream<Item = Result<Bytes, PayloadError>> + 'static,
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -65,7 +66,7 @@ pub async unsafe fn graphql_subscriptions<
 ) -> Result<HttpResponse, Error>
 where
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -78,7 +79,7 @@ where
     start(
         GraphQLWSSession {
             coordinator: coordinator.into_inner(),
-            graphql_context: context,
+            graphql_context: Arc::new(context),
             map_req_id_to_spawn_handle: HashMap::new(),
             has_started: Arc::new(AtomicBool::new(false)),
             handler,
@@ -92,7 +93,7 @@ where
 struct GraphQLWSSession<Query, Mutation, Subscription, Context, S, SubHandler, E>
 where
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -104,7 +105,7 @@ where
 {
     pub map_req_id_to_spawn_handle: HashMap<String, SpawnHandle>,
     pub has_started: Arc<AtomicBool>,
-    pub graphql_context: Context,
+    pub graphql_context: Arc<Context>,
     pub coordinator: Arc<Coordinator<'static, Query, Mutation, Subscription, Context, S>>,
     pub handler: Option<SubHandler>,
     error_handler: std::marker::PhantomData<E>,
@@ -114,7 +115,7 @@ impl<Query, Mutation, Subscription, Context, S, SubHandler, E> Actor
     for GraphQLWSSession<Query, Mutation, Subscription, Context, S, SubHandler, E>
 where
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -134,7 +135,7 @@ impl<Query, Mutation, Subscription, Context, S, SubHandler, E>
     GraphQLWSSession<Query, Mutation, Subscription, Context, S, SubHandler, E>
 where
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -192,7 +193,7 @@ where
         result: (
             GraphQLRequest<S>,
             String,
-            Context,
+            Arc<Context>,
             Arc<Coordinator<'static, Query, Mutation, Subscription, Context, S>>,
         ),
         actor: &mut Self,
@@ -205,7 +206,7 @@ where
 
     async fn handle_subscription(
         req: GraphQLRequest<S>,
-        graphql_context: Context,
+        graphql_context: Arc<Context>,
         request_id: String,
         coord: Arc<Coordinator<'static, Query, Mutation, Subscription, Context, S>>,
         ctx: *mut ws::WebsocketContext<Self>,
@@ -255,7 +256,7 @@ impl<Query, Mutation, Subscription, Context, S, SubHandler, E>
     for GraphQLWSSession<Query, Mutation, Subscription, Context, S, SubHandler, E>
 where
     S: ScalarValue + Send + Sync + 'static,
-    Context: Clone + Send + Sync + 'static + std::marker::Unpin,
+    Context: Send + Sync + 'static + std::marker::Unpin,
     Query: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
     Query::TypeInfo: Send + Sync,
     Mutation: juniper::GraphQLTypeAsync<S, Context = Context> + Send + Sync + 'static,
@@ -289,7 +290,7 @@ where
                         if let Some(handler) = &self.handler {
                             let state = SubscriptionState::OnConnection(
                                 request.payload,
-                                &mut self.graphql_context,
+                                Arc::get_mut(&mut self.graphql_context).unwrap(),
                             );
                             let on_connect_result = handler.handle(state);
                             if let Err(_err) = on_connect_result {
@@ -312,7 +313,7 @@ where
                     }
                     GraphQLOverWebSocketMessage::Start if has_started_value => {
                         let coordinator = self.coordinator.clone();
-                        let mut context = self.graphql_context.clone();
+
                         let payload = request
                             .graphql_payload::<S>()
                             .expect("Could not deserialize payload");
@@ -323,9 +324,12 @@ where
                             payload.variables,
                         );
                         if let Some(handler) = &self.handler {
-                            let state = SubscriptionState::OnOperation(&mut context);
+                            let state = SubscriptionState::OnOperation(
+                                self.graphql_context.deref(),
+                            );
                             handler.handle(state).unwrap();
                         }
+                        let context = self.graphql_context.clone();
                         {
                             use std::collections::hash_map::Entry;
                             let req_id = request_id.clone();
@@ -347,8 +351,8 @@ where
                     GraphQLOverWebSocketMessage::Stop if has_started_value => {
                         let request_id = request.id.unwrap_or("1".to_owned());
                         if let Some(handler) = &self.handler {
-                            let state =
-                                SubscriptionState::OnOperationComplete(&self.graphql_context);
+                            let context = self.graphql_context.deref();
+                            let state = SubscriptionState::OnOperationComplete(context);
                             handler.handle(state).unwrap();
                         }
                         match self.map_req_id_to_spawn_handle.remove(&request_id) {
@@ -366,7 +370,8 @@ where
                     }
                     GraphQLOverWebSocketMessage::ConnectionTerminate => {
                         if let Some(handler) = &self.handler {
-                            let state = SubscriptionState::OnDisconnect(&self.graphql_context);
+                            let context = self.graphql_context.deref();
+                            let state = SubscriptionState::OnDisconnect(context);
                             handler.handle(state).unwrap();
                         }
                         ctx.stop();
@@ -376,7 +381,8 @@ where
             }
             ws::Message::Close(_) => {
                 if let Some(handler) = &self.handler {
-                    let state = SubscriptionState::OnDisconnect(&self.graphql_context);
+                    let context = self.graphql_context.deref();
+                    let state = SubscriptionState::OnDisconnect(context);
                     handler.handle(state).unwrap();
                 }
                 ctx.stop();
@@ -483,6 +489,9 @@ mod tests {
             String::from(
                 r#"{"id":"1","type":"start","payload":{"variables":{},"extensions":{},"operationName":"hello","query":"subscription hello {  helloWorld}"}}"#,
             ),
+            String::from(
+                r#"{"id":"2","type":"start","payload":{"variables":{},"extensions":{},"operationName":"hello","query":"subscription hello {  helloWorld}"}}"#,
+            ),
             String::from(r#"{"id":"1","type":"stop"}"#),
             String::from(r#"{"type":"connection_terminate"}"#),
         ];
@@ -496,6 +505,9 @@ mod tests {
             vec![Some(bytes::Bytes::from(
                 r#"{"type":"data","id":"1","payload":{"data":{"helloWorld":"Hello"}} }"#,
             ))],
+            vec![Some(bytes::Bytes::from(
+                r#"{"type":"data","id":"2","payload":{"data":{"helloWorld":"Hello"}} }"#,
+            ))],
             vec![Some(bytes::Bytes::from(
                 r#"{"type":"complete","id":"1","payload":null}"#,
             ))],

juniper_subscriptions/src/ws_util.rs

@@ -70,7 +70,7 @@ where
     OnConnection(Option<Value>, &'a mut Context),
     /// The Subscription is at the start of a operation after the GQL_START message is
     /// is received.
-    OnOperation(&'a mut Context),
+    OnOperation(&'a Context),
     /// The subscription is on the end of a operation before sending the GQL_COMPLETE
     /// message to the client.
     OnOperationComplete(&'a Context),
@@ -160,15 +160,15 @@ where
 pub mod tests {
     use super::*;
     use juniper::DefaultScalarValue;
-    use std::sync::Mutex;
+    use std::sync::atomic::{AtomicBool, Ordering};
 
     #[derive(Default)]
     struct Context {
         pub user_id: Option<String>,
         pub has_connected: bool,
-        pub has_operated: bool,
-        pub has_completed_operation: Mutex<bool>,
-        pub has_disconnected: Mutex<bool>,
+        pub has_operated: AtomicBool,
+        pub has_completed_operation: AtomicBool,
+        pub has_disconnected: AtomicBool,
     }
 
     #[derive(Deserialize)]
@@ -192,15 +192,13 @@ pub mod tests {
                     ctx.has_connected = true;
                 }
                 SubscriptionState::OnOperation(ctx) => {
-                    ctx.has_operated = true;
+                    ctx.has_operated.store(true, Ordering::Relaxed);
                 }
                 SubscriptionState::OnOperationComplete(ctx) => {
-                    let mut has_completed = ctx.has_completed_operation.lock().unwrap();
-                    *has_completed = true;
+                    ctx.has_completed_operation.store(true, Ordering::Relaxed);
                 }
                 SubscriptionState::OnDisconnect(ctx) => {
-                    let mut has_disconnected = ctx.has_disconnected.lock().unwrap();
-                    *has_disconnected = true;
+                    ctx.has_disconnected.store(true, Ordering::Relaxed);
                 }
             };
             Ok(())
@@ -261,7 +259,7 @@ pub mod tests {
         let type_value = serde_json::to_string(&GraphQLOverWebSocketMessage::Start).unwrap();
         let msg = format!(r#"{{"type":{}, "payload": {{}}, "id": "1" }}"#, type_value);
         assert!(implementation_example(&msg, &mut ctx));
-        assert!(ctx.has_operated);
+        assert!(ctx.has_operated.load(Ordering::Relaxed));
     }
 
     #[test]
@@ -270,8 +268,8 @@ pub mod tests {
         let type_value = serde_json::to_string(&GraphQLOverWebSocketMessage::Stop).unwrap();
         let msg = format!(r#"{{"type":{}, "payload": null, "id": "1" }}"#, type_value);
         assert!(implementation_example(&msg, &mut ctx));
-        let has_completed = ctx.has_completed_operation.lock().unwrap();
-        assert!(*has_completed);
+        let has_completed = ctx.has_completed_operation.load(Ordering::Relaxed);
+        assert!(has_completed);
     }
 
     #[test]
@@ -281,7 +279,7 @@ pub mod tests {
             serde_json::to_string(&GraphQLOverWebSocketMessage::ConnectionTerminate).unwrap();
         let msg = format!(r#"{{"type":{}, "payload": null, "id": "1" }}"#, type_value);
         assert!(implementation_example(&msg, &mut ctx));
-        let has_disconnected = ctx.has_disconnected.lock().unwrap();
-        assert!(*has_disconnected);
+        let has_disconnected = ctx.has_disconnected.load(Ordering::Relaxed);
+        assert!(has_disconnected);
     }
 }