Add @stream directive

src/context.rs

@@ -496,6 +496,11 @@ impl<'a, T> ContextBase<'a, T> {
     pub fn is_defer(&self, directives: &[Positioned<Directive>]) -> bool {
         directives.iter().any(|d| d.name.node == "defer")
     }
+
+    #[doc(hidden)]
+    pub fn is_stream(&self, directives: &[Positioned<Directive>]) -> bool {
+        directives.iter().any(|d| d.name.node == "stream")
+    }
 }
 
 impl<'a> ContextBase<'a, &'a Positioned<SelectionSet>> {

src/lib.rs

@@ -152,7 +152,7 @@ pub use subscription::{
 };
 pub use types::{
     Connection, Cursor, DataSource, Deferred, EmptyEdgeFields, EmptyMutation, EmptySubscription,
-    PageInfo, QueryOperation, Upload,
+    PageInfo, QueryOperation, Streamed, Upload,
 };
 pub use validation::ValidationMode;
 

src/query.rs

@@ -81,10 +81,12 @@ impl QueryResponse {
                 }
                 serde_json::Value::Number(idx) => {
                     if let serde_json::Value::Array(array) = p {
-                        if let Some(next) = array.get_mut(idx.as_i64().unwrap() as usize) {
-                            p = next;
-                            continue;
+                        let idx = idx.as_i64().unwrap() as usize;
+                        while array.len() <= idx {
+                            array.push(serde_json::Value::Null);
                         }
+                        p = array.get_mut(idx as usize).unwrap();
+                        continue;
                     }
                     return;
                 }

src/schema.rs

@@ -235,6 +235,13 @@ where
             args: Default::default(),
         });
 
+        registry.add_directive(MetaDirective {
+            name: "stream",
+            description: None,
+            locations: vec![__DirectiveLocation::FIELD],
+            args: Default::default(),
+        });
+
         // register scalars
         bool::create_type_info(&mut registry);
         i32::create_type_info(&mut registry);

src/types/deferred.rs

@@ -8,7 +8,7 @@ use std::sync::atomic::AtomicUsize;
 
 /// Deferred type
 ///
-/// Allows to defer the type of results returned, Only takes effect when the @defer directive exists on the field.
+/// Allows to defer the type of results returned, only takes effect when the @defer directive exists on the field.
 pub struct Deferred<T: Type + Send + Sync + Clone + 'static>(T);
 
 impl<T: Type + Send + Sync + Clone + 'static> From<T> for Deferred<T> {

src/types/mod.rs

@@ -6,6 +6,7 @@ mod r#enum;
 mod list;
 mod optional;
 mod query_root;
+mod streamed;
 mod upload;
 
 pub use connection::{Connection, Cursor, DataSource, EmptyEdgeFields, PageInfo, QueryOperation};
@@ -14,4 +15,5 @@ pub use empty_mutation::EmptyMutation;
 pub use empty_subscription::EmptySubscription;
 pub use query_root::QueryRoot;
 pub use r#enum::{EnumItem, EnumType};
+pub use streamed::Streamed;
 pub use upload::Upload;

src/types/streamed.rs

@@ -0,0 +1,102 @@
+use crate::context::DeferList;
+use crate::registry::Registry;
+use crate::{ContextSelectionSet, OutputValueType, Positioned, QueryResponse, Result, Type};
+use async_graphql_parser::query::Field;
+use itertools::Itertools;
+use std::borrow::Cow;
+use std::sync::atomic::AtomicUsize;
+use std::sync::Arc;
+
+/// Streamed type
+///
+/// Similar to Deferred, but you can defer every item of the list type, only takes effect when the @stream directive exists on the field.
+pub struct Streamed<T: Type + Send + Sync + Clone + 'static>(Vec<T>);
+
+impl<T: Type + Send + Sync + Clone + 'static> From<Vec<T>> for Streamed<T> {
+    fn from(value: Vec<T>) -> Self {
+        Self(value)
+    }
+}
+
+impl<T: Type + Send + Sync + Clone + 'static> Type for Streamed<T> {
+    fn type_name() -> Cow<'static, str> {
+        Vec::<T>::type_name()
+    }
+
+    fn create_type_info(registry: &mut Registry) -> String {
+        Vec::<T>::create_type_info(registry)
+    }
+}
+
+#[async_trait::async_trait]
+impl<T: OutputValueType + Send + Sync + Clone + 'static> OutputValueType for Streamed<T> {
+    async fn resolve(
+        &self,
+        ctx: &ContextSelectionSet<'_>,
+        field: &Positioned<Field>,
+    ) -> Result<serde_json::Value> {
+        if let Some(defer_list) = ctx.defer_list {
+            if ctx.is_stream(&field.directives) {
+                let list = self.0.clone();
+                let mut field = field.clone();
+
+                // remove @stream directive
+                if let Some((idx, _)) = field
+                    .node
+                    .directives
+                    .iter()
+                    .find_position(|d| d.name.as_str() == "stream")
+                {
+                    field.node.directives.remove(idx);
+                }
+
+                let field = Arc::new(field);
+
+                let path_prefix = ctx
+                    .path_node
+                    .as_ref()
+                    .map(|path| path.to_json())
+                    .unwrap_or_default();
+
+                for (idx, item) in list.into_iter().enumerate() {
+                    let path_prefix = {
+                        let mut path_prefix = path_prefix.clone();
+                        path_prefix.push(serde_json::Value::Number(idx.into()));
+                        path_prefix
+                    };
+                    let field = field.clone();
+                    let schema_env = ctx.schema_env.clone();
+                    let query_env = ctx.query_env.clone();
+
+                    defer_list.append(async move {
+                        let inc_resolve_id = AtomicUsize::default();
+                        let defer_list = DeferList {
+                            path_prefix: path_prefix.clone(),
+                            futures: Default::default(),
+                        };
+                        let ctx = query_env.create_context(
+                            &schema_env,
+                            None,
+                            &field.selection_set,
+                            &inc_resolve_id,
+                            Some(&defer_list),
+                        );
+                        let data = item.resolve(&ctx, &field).await?;
+
+                        Ok((
+                            QueryResponse {
+                                path: Some(path_prefix),
+                                data,
+                                extensions: None,
+                                cache_control: Default::default(),
+                            },
+                            defer_list,
+                        ))
+                    });
+                }
+                return Ok(serde_json::Value::Array(Vec::new()));
+            }
+        }
+        OutputValueType::resolve(&self.0, ctx, field).await
+    }
+}

tests/defer.rs

@@ -92,3 +92,63 @@ pub async fn test_defer() {
 
     assert!(stream.next().await.is_none());
 }
+
+#[async_std::test]
+pub async fn test_stream() {
+    #[SimpleObject]
+    #[derive(Clone)]
+    struct MyObj {
+        value: i32,
+    }
+
+    struct Query;
+
+    #[Object]
+    impl Query {
+        async fn objs(&self) -> Streamed<MyObj> {
+            Streamed::from(vec![
+                MyObj { value: 1 },
+                MyObj { value: 2 },
+                MyObj { value: 3 },
+                MyObj { value: 4 },
+                MyObj { value: 5 },
+            ])
+        }
+    }
+
+    let schema = Schema::new(Query, EmptyMutation, EmptySubscription);
+    let query = r#"{
+        objs @stream { value }
+    }"#;
+    assert_eq!(
+        schema.execute(&query).await.unwrap().data,
+        serde_json::json!({
+            "objs": [
+                { "value": 1 },
+                { "value": 2 },
+                { "value": 3 },
+                { "value": 4 },
+                { "value": 5 },
+            ]
+        })
+    );
+
+    let mut stream = schema.execute_stream(&query);
+    assert_eq!(
+        stream.next().await.unwrap().unwrap().data,
+        serde_json::json!({
+            "objs": [],
+        })
+    );
+
+    for i in 0..5 {
+        let next_resp = stream.next().await.unwrap().unwrap();
+        assert_eq!(
+            next_resp.path,
+            Some(vec![serde_json::json!("objs"), i.into()])
+        );
+        assert_eq!(next_resp.data, serde_json::json!({ "value": i + 1 }));
+    }
+
+    assert!(stream.next().await.is_none());
+}