clients can use multipart too

Author: enisdenjo

bin/router/src/pipeline/header.rs

@@ -17,6 +17,7 @@ lazy_static! {
     pub static ref APPLICATION_GRAPHQL_RESPONSE_JSON: HeaderValue =
         HeaderValue::from_static(&APPLICATION_GRAPHQL_RESPONSE_JSON_STR);
     pub static ref TEXT_EVENT_STREAM: &'static str = "text/event-stream";
+    pub static ref MULTIPART_MIXED: &'static str = r#"multipart/mixed;subscriptionSpec="1.0""#;
 }
 
 pub trait RequestAccepts {

bin/router/src/pipeline/mod.rs

@@ -1,5 +1,6 @@
 use std::{sync::Arc, time::Instant};
 
+use futures_util::Stream;
 use hive_router_plan_executor::execution::{
     client_request_details::{ClientRequestDetails, JwtRequestDetails, OperationDetails},
     plan::QueryPlanExecutionResult,
@@ -24,8 +25,8 @@ use crate::{
         execution_request::get_execution_request,
         header::{
             RequestAccepts, APPLICATION_GRAPHQL_RESPONSE_JSON,
-            APPLICATION_GRAPHQL_RESPONSE_JSON_STR, APPLICATION_JSON, TEXT_EVENT_STREAM,
-            TEXT_HTML_CONTENT_TYPE,
+            APPLICATION_GRAPHQL_RESPONSE_JSON_STR, APPLICATION_JSON, MULTIPART_MIXED,
+            TEXT_EVENT_STREAM, TEXT_HTML_CONTENT_TYPE,
         },
         normalize::{normalize_request_with_cache, GraphQLNormalizationPayload},
         parser::parse_operation_with_cache,
@@ -108,13 +109,32 @@ pub async fn graphql_request_handler(
                 .body(response_bytes)
         }
         Ok(QueryPlanExecutionResult::Stream(response)) => {
-            use crate::pipeline::sse;
+            use crate::pipeline::{header::TEXT_EVENT_STREAM, multipart_subscribe, sse};
 
-            let response_content_type = http::HeaderValue::from_static("text/event-stream");
-            let body = Box::pin(sse::create_stream(
-                response.body,
-                std::time::Duration::from_secs(10),
-            ));
+            // TODO: respect order of Accept header
+            let (response_content_type, body): (
+                http::HeaderValue,
+                std::pin::Pin<
+                    Box<dyn Stream<Item = Result<ntex::util::Bytes, std::io::Error>> + Send>,
+                >,
+            ) = if req.accepts_content_type(*TEXT_EVENT_STREAM) {
+                (
+                    http::HeaderValue::from_static("text/event-stream"),
+                    Box::pin(sse::create_stream(
+                        response.body,
+                        std::time::Duration::from_secs(10),
+                    )),
+                )
+            } else {
+                // NOTE: client accept headers should have been validated earlier
+                (
+                    http::HeaderValue::from_static("multipart/mixed;boundary=graphql"),
+                    Box::pin(multipart_subscribe::create_stream(
+                        response.body,
+                        std::time::Duration::from_secs(10),
+                    )),
+                )
+            };
 
             let mut response_builder = web::HttpResponse::Ok();
             for (header_name, header_value) in response.headers {
@@ -162,7 +182,10 @@ pub async fn execute_pipeline(
         Some(OperationKind::Subscription)
     );
 
-    if is_subscription && !req.accepts_content_type(*TEXT_EVENT_STREAM) {
+    if is_subscription
+        && !req.accepts_content_type(*MULTIPART_MIXED)
+        && !req.accepts_content_type(*TEXT_EVENT_STREAM)
+    {
         return Err(
             req.new_pipeline_error(PipelineErrorVariant::SubscriptionNotSupportedOverTransport)
         );

e2e/src/subscriptions.rs

@@ -20,7 +20,7 @@ mod subscription_e2e_tests {
     }
 
     #[ntex::test]
-    async fn subscription_no_entity_resolution_sse() {
+    async fn subscription_no_entity_resolution_sse_subgraph() {
         let _subgraphs_server = SubgraphsServer::start_with_subscriptions_protocol(
             subgraphs::SubscriptionProtocol::SseOnly,
         )
@@ -107,7 +107,7 @@ mod subscription_e2e_tests {
     }
 
     #[ntex::test]
-    async fn subscription_no_entity_resolution_multipart() {
+    async fn subscription_no_entity_resolution_multipart_subgraph() {
         let _subgraphs_server = SubgraphsServer::start_with_subscriptions_protocol(
             subgraphs::SubscriptionProtocol::MultipartOnly,
         )
@@ -278,6 +278,106 @@ mod subscription_e2e_tests {
         );
     }
 
+    #[ntex::test]
+    async fn subscription_yes_entity_resolution_multipart_client() {
+        let _subgraphs_server = SubgraphsServer::start().await;
+
+        let router = init_router_from_config_inline(&format!(
+            r#"
+            supergraph:
+                source: file
+                path: supergraph.graphql
+            "#
+        ))
+        .await
+        .unwrap();
+
+        wait_for_readiness(&router.app).await;
+
+        let req = init_graphql_request(
+            r#"
+            subscription {
+                reviewAdded(intervalInMs: 0) {
+                    id
+                    product {
+                        name
+                    }
+                }
+            }
+            "#,
+            None,
+        )
+        .header(
+            http::header::ACCEPT,
+            // as per https://www.apollographql.com/docs/graphos/routing/operations/subscriptions/multipart-protocol#executing-a-subscription
+            r#"multipart/mixed;subscriptionSpec="1.0", application/json"#,
+        )
+        .to_request();
+
+        let res = test::call_service(&router.app, req).await;
+
+        assert!(res.status().is_success(), "Expected 200 OK");
+
+        let content_type_header = get_content_type_header(&res);
+
+        let body = test::read_body(res).await;
+        let body_str = std::str::from_utf8(&body).unwrap();
+
+        assert_snapshot!(body_str, @r#"
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"1","product":{"name":"Table"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"2","product":{"name":"Table"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"3","product":{"name":"Table"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"4","product":{"name":"Table"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"5","product":{"name":"Couch"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"6","product":{"name":"Couch"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"7","product":{"name":"Couch"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"8","product":{"name":"Couch"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"9","product":{"name":"Glass"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"10","product":{"name":"Chair"}}}}}
+        --graphql
+        Content-Type: application/json
+
+        {"payload":{"data":{"reviewAdded":{"id":"11","product":{"name":"Chair"}}}}}
+        --graphql--
+        "#);
+
+        // we check this at the end because the body will hold clues to why the test fails
+        assert_eq!(
+            content_type_header, "multipart/mixed;boundary=graphql",
+            "Expected Content-Type to be multipart/mixed;boundary=graphql"
+        );
+    }
+
     #[ntex::test]
     async fn subscription_entity_resolution_with_requires() {
         let _subgraphs_server = SubgraphsServer::start().await;

lib/executor/src/executors/http.rs

@@ -363,7 +363,7 @@ impl SubgraphExecutor for HTTPSubgraphExecutor {
         headers.insert(
             http::header::ACCEPT,
             HeaderValue::from_static(
-                r#"multipart/mixed;boundary="graphql";subscriptionSpec="1.0", text/event-stream"#,
+                r#"multipart/mixed;subscriptionSpec="1.0", text/event-stream"#,
             ),
         );
         *req.headers_mut() = headers;