support the incremental response field

NEXT_CHANGELOG.md

@@ -484,6 +484,13 @@ Introspection queries will now see the `@defer` directive if it was activated in
 
 By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1557
 
+### Support the incremental response field ([PR #1551](https://github.com/apollographql/router/pull/1551))
+
+Recent changes in the `@defer` specification now mandate that the deferred responses are transmitted
+as an array in the new `incremental` field of the JSON response.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1551
+
 ## 🛠 Maintenance
 
 ### Display licenses.html diff in CI if the check failed ([#1524](https://github.com/apollographql/router/issues/1524))

apollo-router/src/error.rs

@@ -125,13 +125,8 @@ impl FetchError {
     /// Convert the error to an appropriate response.
     pub fn to_response(&self) -> Response {
         Response {
-            label: Default::default(),
-            data: Default::default(),
-            path: Default::default(),
             errors: vec![self.to_graphql_error(None)],
-            extensions: Default::default(),
-            subselection: Default::default(),
-            has_next: Default::default(),
+            ..Response::default()
         }
     }
 }

apollo-router/src/response.rs

@@ -42,12 +42,16 @@ pub struct Response {
 
     #[serde(skip_serializing)]
     pub subselection: Option<String>,
+
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
+    pub incremental: Vec<IncrementalResponse>,
 }
 
 #[buildstructor::buildstructor]
 impl Response {
     /// Constructor
     #[builder(visibility = "pub")]
+    #[allow(clippy::too_many_arguments)]
     fn new(
         label: Option<String>,
         data: Option<Value>,
@@ -56,6 +60,7 @@ impl Response {
         extensions: Map<ByteString, Value>,
         subselection: Option<String>,
         has_next: Option<bool>,
+        incremental: Vec<IncrementalResponse>,
     ) -> Self {
         Self {
             label,
@@ -65,10 +70,11 @@ impl Response {
             extensions,
             subselection,
             has_next,
+            incremental,
         }
     }
 
-    /// If path is None, this is a primary query.
+    /// If path is None, this is a primary response.
     pub fn is_primary(&self) -> bool {
         self.path.is_none()
     }
@@ -128,6 +134,24 @@ impl Response {
                 service: service_name.to_string(),
                 reason: err.to_string(),
             })?;
+        let incremental =
+            extract_key_value_from_object!(object, "incremental", Value::Array(a) => a).map_err(
+                |err| FetchError::SubrequestMalformedResponse {
+                    service: service_name.to_string(),
+                    reason: err.to_string(),
+                },
+            )?;
+        let incremental: Vec<IncrementalResponse> = match incremental {
+            Some(v) => v
+                .into_iter()
+                .map(serde_json_bytes::from_value)
+                .collect::<Result<Vec<IncrementalResponse>, _>>()
+                .map_err(|err| FetchError::SubrequestMalformedResponse {
+                    service: service_name.to_string(),
+                    reason: err.to_string(),
+                })?,
+            None => vec![],
+        };
 
         Ok(Response {
             label,
@@ -137,10 +161,62 @@ impl Response {
             extensions,
             subselection: None,
             has_next,
+            incremental,
         })
     }
 }
 
+/// A graphql incremental response.
+/// Used with `@defer`
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct IncrementalResponse {
+    /// The label that was passed to the defer or stream directive for this patch.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub label: Option<String>,
+
+    /// The response data.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub data: Option<Value>,
+
+    /// The path that the data should be merged at.
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub path: Option<Path>,
+
+    /// The optional graphql errors encountered.
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
+    pub errors: Vec<Error>,
+
+    /// The optional graphql extensions.
+    #[serde(skip_serializing_if = "Object::is_empty", default)]
+    pub extensions: Object,
+}
+
+#[buildstructor::buildstructor]
+impl IncrementalResponse {
+    /// Constructor
+    #[builder(visibility = "pub")]
+    fn new(
+        label: Option<String>,
+        data: Option<Value>,
+        path: Option<Path>,
+        errors: Vec<Error>,
+        extensions: Map<ByteString, Value>,
+    ) -> Self {
+        Self {
+            label,
+            data,
+            path,
+            errors,
+            extensions,
+        }
+    }
+
+    /// append_errors default the errors `path` with the one provided.
+    pub fn append_errors(&mut self, errors: &mut Vec<Error>) {
+        self.errors.append(errors)
+    }
+}
 #[cfg(test)]
 mod tests {
     use serde_json::json;

apollo-router/src/services/supergraph_service.rs

@@ -5,8 +5,6 @@ use std::task::Poll;
 
 use futures::future::ready;
 use futures::future::BoxFuture;
-use futures::future::Either;
-use futures::stream;
 use futures::stream::once;
 use futures::stream::BoxStream;
 use futures::stream::StreamExt;
@@ -41,6 +39,7 @@ use crate::plugin::DynPlugin;
 use crate::plugin::Handler;
 use crate::query_planner::BridgeQueryPlanner;
 use crate::query_planner::CachingQueryPlanner;
+use crate::response::IncrementalResponse;
 use crate::router_factory::SupergraphServiceFactory;
 use crate::services::layers::apq::APQLayer;
 use crate::services::layers::ensure_query_presence::EnsureQueryPresence;
@@ -183,7 +182,7 @@ where
                             response: http::Response::from_parts(
                                 parts,
                                 response_stream
-                                    .flat_map(move |mut response: Response| {
+                                    .map(move |mut response: Response| {
                                         tracing::debug_span!("format_response").in_scope(|| {
                                             query.format_response(
                                                 &mut response,
@@ -199,7 +198,7 @@ where
                                                     response.has_next = Some(true);
                                                 }
 
-                                                Either::Left(once(ready(response)))
+                                                response
                                             }
                                             // if the deferred response specified a path, we must extract the
                                             //values matched by that path and create a separate response for
@@ -221,21 +220,27 @@ where
                                                     },
                                                 );
 
-                                                Either::Right(stream::iter(
-                                                    sub_responses.into_iter().map(
-                                                        move |(path, data)| Response {
-                                                            label: response.label.clone(),
-                                                            data: Some(data),
-                                                            path: Some(path),
-                                                            errors: response.errors.clone(),
-                                                            extensions: response.extensions.clone(),
-                                                            has_next: Some(true),
-                                                            subselection: response
-                                                                .subselection
-                                                                .clone(),
-                                                        },
-                                                    ),
-                                                ))
+                                                Response::builder()
+                                                    .has_next(true)
+                                                    .incremental(
+                                                        sub_responses
+                                                            .into_iter()
+                                                            .map(move |(path, data)| {
+                                                                IncrementalResponse::builder()
+                                                                    .and_label(
+                                                                        response.label.clone(),
+                                                                    )
+                                                                    .data(data)
+                                                                    .path(path)
+                                                                    .errors(response.errors.clone())
+                                                                    .extensions(
+                                                                        response.extensions.clone(),
+                                                                    )
+                                                                    .build()
+                                                            })
+                                                            .collect(),
+                                                    )
+                                                    .build()
                                             }
                                         }
                                     })

apollo-router/tests/integration_tests.rs

@@ -659,7 +659,7 @@ async fn defer_path() {
             r#"{
             me {
                 id
-                ...@defer {
+                ...@defer(label: "name") {
                     name
                 }
             }
@@ -678,26 +678,10 @@ async fn defer_path() {
     let mut stream = router.oneshot(request.into()).await.unwrap();
 
     let first = stream.next_response().await.unwrap();
-    println!("first: {:?}", first);
-    assert_eq!(
-        first.data.unwrap(),
-        serde_json_bytes::json! {{
-            "me":{
-                "id":"1"
-            }
-        }}
-    );
-    assert_eq!(first.path, None);
+    insta::assert_json_snapshot!(first);
 
     let second = stream.next_response().await.unwrap();
-    println!("second: {:?}", second);
-    assert_eq!(
-        second.data.unwrap(),
-        serde_json_bytes::json! {{
-            "name": "Ada Lovelace"
-        }}
-    );
-    assert_eq!(second.path.unwrap().to_string(), "/me");
+    insta::assert_json_snapshot!(second);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -720,7 +704,7 @@ async fn defer_path_in_array() {
                         id
                         author {
                             id
-                            ... @defer {
+                            ... @defer(label: "author name") {
                             name
                             }
                         }
@@ -741,51 +725,10 @@ async fn defer_path_in_array() {
     let mut stream = router.oneshot(request.into()).await.unwrap();
 
     let first = stream.next_response().await.unwrap();
-    println!("first: {:?}", first);
-    assert_eq!(
-        first.data.unwrap(),
-        serde_json_bytes::json! {{
-            "me":{
-                "reviews":[
-                    {
-                        "id": "1",
-                        "author":
-                        {
-                            "id": "1"
-                        }
-                    },
-                    {
-                        "id": "2",
-                        "author":
-                        {
-                            "id": "1"
-                        }
-                    }
-                ]
-            }
-        }}
-    );
-    assert_eq!(first.path, None);
+    insta::assert_json_snapshot!(first);
 
     let second = stream.next_response().await.unwrap();
-    println!("second: {:?}", second);
-    assert_eq!(
-        second.data.unwrap(),
-        serde_json_bytes::json! {{
-            "name": "Ada Lovelace"
-        }}
-    );
-    assert_eq!(second.path.unwrap().to_string(), "/me/reviews/0/author");
-
-    let third = stream.next_response().await.unwrap();
-    println!("third: {:?}", third);
-    assert_eq!(
-        third.data.unwrap(),
-        serde_json_bytes::json! {{
-            "name": "Ada Lovelace"
-        }}
-    );
-    assert_eq!(third.path.unwrap().to_string(), "/me/reviews/1/author");
+    insta::assert_json_snapshot!(second);
 }
 
 async fn query_node(

apollo-router/tests/snapshots/integration_tests__defer_path-2.snap

@@ -0,0 +1,19 @@
+---
+source: apollo-router/tests/integration_tests.rs
+assertion_line: 679
+expression: second
+---
+{
+  "hasNext": true,
+  "incremental": [
+    {
+      "label": "name",
+      "data": {
+        "name": "Ada Lovelace"
+      },
+      "path": [
+        "me"
+      ]
+    }
+  ]
+}

apollo-router/tests/snapshots/integration_tests__defer_path.snap

@@ -0,0 +1,13 @@
+---
+source: apollo-router/tests/integration_tests.rs
+assertion_line: 686
+expression: first
+---
+{
+  "data": {
+    "me": {
+      "id": "1"
+    }
+  },
+  "hasNext": true
+}