async-graphql 2.0.0-alpha.9

A GraphQL server library implemented in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

use crate::parser::types::{Selection, TypeCondition};
use crate::{Context, ContextSelectionSet, Result, Type};
use futures::{Stream, StreamExt};
use std::pin::Pin;

/// Represents a GraphQL subscription object
pub trait SubscriptionType: Type {
    /// This function returns true of type `EmptySubscription` only.
    #[doc(hidden)]
    fn is_empty() -> bool {
        false
    }

    #[doc(hidden)]
    fn create_field_stream<'a>(
        &'a self,
        ctx: &'a Context<'a>,
    ) -> Pin<Box<dyn Stream<Item = Result<serde_json::Value>> + Send + 'a>>;
}

pub(crate) fn collect_subscription_streams<'a, T: SubscriptionType + Send + Sync + 'static>(
    ctx: &ContextSelectionSet<'a>,
    root: &'a T,
    streams: &mut Vec<Pin<Box<dyn Stream<Item = Result<serde_json::Value>> + Send + 'a>>>,
) -> Result<()> {
    for selection in &ctx.item.node.items {
        if ctx.is_skip(selection.node.directives())? {
            continue;
        }
        match &selection.node {
            Selection::Field(field) => streams.push(Box::pin({
                let ctx = ctx.clone();
                async_stream::stream! {
                    let ctx = ctx.with_field(field);
                    let mut stream = root.create_field_stream(&ctx);
                    while let Some(item) = stream.next().await {
                        yield item;
                    }
                }
            })),
            Selection::FragmentSpread(fragment_spread) => {
                if let Some(fragment) = ctx
                    .query_env
                    .document
                    .fragments
                    .get(&fragment_spread.node.fragment_name.node)
                {
                    collect_subscription_streams(
                        &ctx.with_selection_set(&fragment.node.selection_set),
                        root,
                        streams,
                    )?;
                }
            }
            Selection::InlineFragment(inline_fragment) => {
                if let Some(TypeCondition { on: name }) = inline_fragment
                    .node
                    .type_condition
                    .as_ref()
                    .map(|v| &v.node)
                {
                    if name.node.as_str() == T::type_name() {
                        collect_subscription_streams(
                            &ctx.with_selection_set(&inline_fragment.node.selection_set),
                            root,
                            streams,
                        )?;
                    }
                } else {
                    collect_subscription_streams(
                        &ctx.with_selection_set(&inline_fragment.node.selection_set),
                        root,
                        streams,
                    )?;
                }
            }
        }
    }
    Ok(())
}

impl<T: SubscriptionType + Send + Sync> SubscriptionType for &T {
    fn create_field_stream<'a>(
        &'a self,
        ctx: &'a Context<'a>,
    ) -> Pin<Box<dyn Stream<Item = Result<serde_json::Value>> + Send + 'a>> {
        T::create_field_stream(*self, ctx)
    }
}