Add @stream directive
This commit is contained in:
parent
70448c1d63
commit
4868cf242b
|
@ -496,6 +496,11 @@ impl<'a, T> ContextBase<'a, T> {
|
||||||
pub fn is_defer(&self, directives: &[Positioned<Directive>]) -> bool {
|
pub fn is_defer(&self, directives: &[Positioned<Directive>]) -> bool {
|
||||||
directives.iter().any(|d| d.name.node == "defer")
|
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>> {
|
impl<'a> ContextBase<'a, &'a Positioned<SelectionSet>> {
|
||||||
|
|
|
@ -152,7 +152,7 @@ pub use subscription::{
|
||||||
};
|
};
|
||||||
pub use types::{
|
pub use types::{
|
||||||
Connection, Cursor, DataSource, Deferred, EmptyEdgeFields, EmptyMutation, EmptySubscription,
|
Connection, Cursor, DataSource, Deferred, EmptyEdgeFields, EmptyMutation, EmptySubscription,
|
||||||
PageInfo, QueryOperation, Upload,
|
PageInfo, QueryOperation, Streamed, Upload,
|
||||||
};
|
};
|
||||||
pub use validation::ValidationMode;
|
pub use validation::ValidationMode;
|
||||||
|
|
||||||
|
|
|
@ -81,10 +81,12 @@ impl QueryResponse {
|
||||||
}
|
}
|
||||||
serde_json::Value::Number(idx) => {
|
serde_json::Value::Number(idx) => {
|
||||||
if let serde_json::Value::Array(array) = p {
|
if let serde_json::Value::Array(array) = p {
|
||||||
if let Some(next) = array.get_mut(idx.as_i64().unwrap() as usize) {
|
let idx = idx.as_i64().unwrap() as usize;
|
||||||
p = next;
|
while array.len() <= idx {
|
||||||
continue;
|
array.push(serde_json::Value::Null);
|
||||||
}
|
}
|
||||||
|
p = array.get_mut(idx as usize).unwrap();
|
||||||
|
continue;
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
|
@ -235,6 +235,13 @@ where
|
||||||
args: Default::default(),
|
args: Default::default(),
|
||||||
});
|
});
|
||||||
|
|
||||||
|
registry.add_directive(MetaDirective {
|
||||||
|
name: "stream",
|
||||||
|
description: None,
|
||||||
|
locations: vec![__DirectiveLocation::FIELD],
|
||||||
|
args: Default::default(),
|
||||||
|
});
|
||||||
|
|
||||||
// register scalars
|
// register scalars
|
||||||
bool::create_type_info(&mut registry);
|
bool::create_type_info(&mut registry);
|
||||||
i32::create_type_info(&mut registry);
|
i32::create_type_info(&mut registry);
|
||||||
|
|
|
@ -8,7 +8,7 @@ use std::sync::atomic::AtomicUsize;
|
||||||
|
|
||||||
/// Deferred type
|
/// 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);
|
pub struct Deferred<T: Type + Send + Sync + Clone + 'static>(T);
|
||||||
|
|
||||||
impl<T: Type + Send + Sync + Clone + 'static> From<T> for Deferred<T> {
|
impl<T: Type + Send + Sync + Clone + 'static> From<T> for Deferred<T> {
|
||||||
|
|
|
@ -6,6 +6,7 @@ mod r#enum;
|
||||||
mod list;
|
mod list;
|
||||||
mod optional;
|
mod optional;
|
||||||
mod query_root;
|
mod query_root;
|
||||||
|
mod streamed;
|
||||||
mod upload;
|
mod upload;
|
||||||
|
|
||||||
pub use connection::{Connection, Cursor, DataSource, EmptyEdgeFields, PageInfo, QueryOperation};
|
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 empty_subscription::EmptySubscription;
|
||||||
pub use query_root::QueryRoot;
|
pub use query_root::QueryRoot;
|
||||||
pub use r#enum::{EnumItem, EnumType};
|
pub use r#enum::{EnumItem, EnumType};
|
||||||
|
pub use streamed::Streamed;
|
||||||
pub use upload::Upload;
|
pub use upload::Upload;
|
||||||
|
|
102
src/types/streamed.rs
Normal file
102
src/types/streamed.rs
Normal file
|
@ -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
|
||||||
|
}
|
||||||
|
}
|
|
@ -92,3 +92,63 @@ pub async fn test_defer() {
|
||||||
|
|
||||||
assert!(stream.next().await.is_none());
|
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());
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in New Issue
Block a user