2020-04-06 05:49:39 +00:00
|
|
|
use crate::context::Environment;
|
2020-05-15 02:38:48 +00:00
|
|
|
use crate::parser::query::{Selection, TypeCondition};
|
2020-04-06 05:49:39 +00:00
|
|
|
use crate::{Context, ContextSelectionSet, ObjectType, Result, Schema, Type};
|
2020-04-07 06:30:46 +00:00
|
|
|
use futures::{Future, Stream};
|
2020-04-06 05:49:39 +00:00
|
|
|
use std::pin::Pin;
|
|
|
|
use std::sync::Arc;
|
2020-03-29 12:02:52 +00:00
|
|
|
|
|
|
|
/// Represents a GraphQL subscription object
|
|
|
|
#[async_trait::async_trait]
|
|
|
|
pub trait SubscriptionType: Type {
|
|
|
|
/// This function returns true of type `EmptySubscription` only
|
|
|
|
#[doc(hidden)]
|
|
|
|
fn is_empty() -> bool {
|
|
|
|
false
|
|
|
|
}
|
|
|
|
|
|
|
|
#[doc(hidden)]
|
2020-04-07 06:30:46 +00:00
|
|
|
async fn create_field_stream<Query, Mutation>(
|
2020-03-29 12:02:52 +00:00
|
|
|
&self,
|
2020-05-12 08:27:06 +00:00
|
|
|
idx: usize,
|
2020-04-06 05:49:39 +00:00
|
|
|
ctx: &Context<'_>,
|
|
|
|
schema: &Schema<Query, Mutation, Self>,
|
|
|
|
environment: Arc<Environment>,
|
2020-05-03 08:02:46 +00:00
|
|
|
) -> Result<Pin<Box<dyn Stream<Item = Result<serde_json::Value>> + Send>>>
|
2020-04-06 05:49:39 +00:00
|
|
|
where
|
|
|
|
Query: ObjectType + Send + Sync + 'static,
|
|
|
|
Mutation: ObjectType + Send + Sync + 'static,
|
|
|
|
Self: Send + Sync + 'static + Sized;
|
|
|
|
}
|
|
|
|
|
2020-04-07 06:30:46 +00:00
|
|
|
type BoxCreateStreamFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
|
|
|
|
|
|
|
|
pub fn create_subscription_stream<'a, Query, Mutation, Subscription>(
|
|
|
|
schema: &'a Schema<Query, Mutation, Subscription>,
|
2020-04-06 05:49:39 +00:00
|
|
|
environment: Arc<Environment>,
|
2020-04-07 06:30:46 +00:00
|
|
|
ctx: &'a ContextSelectionSet<'_>,
|
2020-05-03 08:02:46 +00:00
|
|
|
streams: &'a mut Vec<Pin<Box<dyn Stream<Item = Result<serde_json::Value>> + Send>>>,
|
2020-04-07 06:30:46 +00:00
|
|
|
) -> BoxCreateStreamFuture<'a>
|
2020-04-06 05:49:39 +00:00
|
|
|
where
|
|
|
|
Query: ObjectType + Send + Sync + 'static,
|
|
|
|
Mutation: ObjectType + Send + Sync + 'static,
|
|
|
|
Subscription: SubscriptionType + Send + Sync + 'static + Sized,
|
|
|
|
{
|
2020-04-07 06:30:46 +00:00
|
|
|
Box::pin(async move {
|
2020-05-12 08:27:06 +00:00
|
|
|
for (idx, selection) in ctx.items.iter().enumerate() {
|
2020-05-09 09:55:04 +00:00
|
|
|
match &selection.node {
|
2020-04-07 06:30:46 +00:00
|
|
|
Selection::Field(field) => {
|
|
|
|
if ctx.is_skip(&field.directives)? {
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
streams.push(
|
|
|
|
schema
|
|
|
|
.0
|
|
|
|
.subscription
|
|
|
|
.create_field_stream(
|
2020-05-12 08:27:06 +00:00
|
|
|
idx,
|
2020-04-07 06:30:46 +00:00
|
|
|
&ctx.with_field(field),
|
|
|
|
schema,
|
|
|
|
environment.clone(),
|
|
|
|
)
|
|
|
|
.await?,
|
|
|
|
)
|
2020-04-06 05:49:39 +00:00
|
|
|
}
|
2020-04-07 06:30:46 +00:00
|
|
|
Selection::FragmentSpread(fragment_spread) => {
|
|
|
|
if ctx.is_skip(&fragment_spread.directives)? {
|
|
|
|
continue;
|
|
|
|
}
|
2020-04-06 05:49:39 +00:00
|
|
|
|
2020-05-12 08:27:06 +00:00
|
|
|
if let Some(fragment) = ctx
|
|
|
|
.document
|
|
|
|
.fragments()
|
2020-05-16 13:14:26 +00:00
|
|
|
.get(fragment_spread.fragment_name.as_str())
|
2020-04-07 06:30:46 +00:00
|
|
|
{
|
|
|
|
create_subscription_stream(
|
|
|
|
schema,
|
|
|
|
environment.clone(),
|
|
|
|
&ctx.with_selection_set(&fragment.selection_set),
|
|
|
|
streams,
|
|
|
|
)
|
|
|
|
.await?;
|
|
|
|
}
|
2020-04-06 05:49:39 +00:00
|
|
|
}
|
2020-04-07 06:30:46 +00:00
|
|
|
Selection::InlineFragment(inline_fragment) => {
|
|
|
|
if ctx.is_skip(&inline_fragment.directives)? {
|
|
|
|
continue;
|
|
|
|
}
|
2020-04-06 05:49:39 +00:00
|
|
|
|
2020-05-09 09:55:04 +00:00
|
|
|
if let Some(TypeCondition::On(name)) =
|
|
|
|
inline_fragment.type_condition.as_ref().map(|v| &v.node)
|
|
|
|
{
|
2020-05-12 08:27:06 +00:00
|
|
|
if name.node == Subscription::type_name() {
|
2020-04-07 06:30:46 +00:00
|
|
|
create_subscription_stream(
|
|
|
|
schema,
|
|
|
|
environment.clone(),
|
|
|
|
&ctx.with_selection_set(&inline_fragment.selection_set),
|
|
|
|
streams,
|
|
|
|
)
|
|
|
|
.await?;
|
|
|
|
}
|
|
|
|
} else {
|
2020-04-06 05:49:39 +00:00
|
|
|
create_subscription_stream(
|
|
|
|
schema,
|
|
|
|
environment.clone(),
|
|
|
|
&ctx.with_selection_set(&inline_fragment.selection_set),
|
|
|
|
streams,
|
2020-04-07 06:30:46 +00:00
|
|
|
)
|
|
|
|
.await?;
|
2020-04-06 05:49:39 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2020-04-07 06:30:46 +00:00
|
|
|
Ok(())
|
|
|
|
})
|
2020-03-29 12:02:52 +00:00
|
|
|
}
|