GraphQL Subscriptions, Federation, and Protocol Comparison

A subscription is a GraphQL operation that returns a stream of results over time. Federation builds one graph from schemas owned by different services. This lesson builds an order-status subscription, two federated subgraphs, and tests. It continues from DataLoader, errors and limits.

Project setup

The project is in memory and needs no database. tokio-stream with the sync feature wraps a broadcast channel as a stream.

Cargo.toml

TOML

[package]
name = "orders-events"
version = "0.1.0"
edition = "2021"

[dependencies]
async-graphql = "7.0"
async-graphql-axum = "7.0"
axum = "0.8"
futures-util = "0.3"
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "time", "sync"] }
tokio-stream = { version = "0.1", features = ["sync"] }

Example 1: A subscription

A subscription is a third root type. The client sends it once, and the server sends one result each time an event happens. Each result has the shape of a query result.

GRAPHQL

subscription {
  orderStatusChanged(orderId: "7") {
    status
  }
}

A subscription needs a transport where the server can send at any time, so it uses WebSocket (a persistent two-way connection over an upgraded HTTP request) with the graphql-transport-ws sub-protocol. Authentication and backpressure on the socket are covered in WebSockets and SSE.

A subscription resolver is an async method that returns a stream, and each item becomes one result. Events come from a tokio::sync::broadcast channel. A mutation sends an event after the change is stored, and each subscriber filters for its order.

src/lib.rs

Rust

pub mod schema;

src/schema.rs

Rust

use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};

use async_graphql::{Context, Enum, Error, Object, Result, Schema, SimpleObject, Subscription, ID};
use futures_util::{Stream, StreamExt};
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;

#[derive(Enum, Copy, Clone, Eq, PartialEq, Debug)]
pub enum OrderStatus {
    Placed,
    Paid,
    Shipped,
}

#[derive(SimpleObject, Clone, Debug)]
pub struct Order {
    pub id: ID,
    pub status: OrderStatus,
    #[graphql(deprecation = "Use totalCents")]
    pub total: f64,
    pub total_cents: i32,
}

pub struct Store {
    orders: Mutex<HashMap<String, Order>>,
}

impl Store {
    pub fn seeded() -> Self {
        let mut orders = HashMap::new();
        orders.insert(
            "7".to_string(),
            Order {
                id: ID::from("7"),
                status: OrderStatus::Placed,
                total: 12.5,
                total_cents: 1250,
            },
        );
        Store {
            orders: Mutex::new(orders),
        }
    }
}

pub struct QueryRoot;

#[Object]
impl QueryRoot {
    async fn order(&self, ctx: &Context<'_>, id: ID) -> Result<Option<Order>> {
        let store = ctx.data::<Arc<Store>>()?;
        let orders = store.orders.lock().unwrap();
        Ok(orders.get(id.as_str()).cloned())
    }
}

pub struct MutationRoot;

#[Object]
impl MutationRoot {
    async fn advance_order(&self, ctx: &Context<'_>, id: ID) -> Result<Order> {
        let store = ctx.data::<Arc<Store>>()?;
        let updated = {
            let mut orders = store.orders.lock().unwrap();
            let order = orders
                .get_mut(id.as_str())
                .ok_or_else(|| Error::new(format!("order {} not found", id.as_str())))?;
            order.status = match order.status {
                OrderStatus::Placed => OrderStatus::Paid,
                OrderStatus::Paid => OrderStatus::Shipped,
                OrderStatus::Shipped => return Err(Error::new("order already shipped")),
            };
            order.clone()
        };
        let events = ctx.data::<broadcast::Sender<Order>>()?;
        let _ = events.send(updated.clone());
        Ok(updated)
    }
}

pub struct SubscriptionRoot;

#[Subscription]
impl SubscriptionRoot {
    async fn order_status_changed(
        &self,
        ctx: &Context<'_>,
        order_id: ID,
    ) -> impl Stream<Item = Order> {
        let receiver = ctx.data_unchecked::<broadcast::Sender<Order>>().subscribe();
        BroadcastStream::new(receiver).filter_map(move |item| {
            let wanted = order_id.clone();
            async move {
                match item {
                    Ok(order) if order.id == wanted => Some(order),
                    _ => None,
                }
            }
        })
    }
}

pub type OrdersSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;

pub fn build_schema() -> OrdersSchema {
    let (events, _) = broadcast::channel::<Order>(64);
    Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
        .data(Arc::new(Store::seeded()))
        .data(events)
        .finish()
}

send returns an error when nobody is subscribed. Ignoring it is correct.
BroadcastStream yields an error item when a slow subscriber misses messages. The filter drops those errors, so a slow client silently skips events. That suits a status that a later one supersedes. A message that must arrive needs a durable source and a resume position (see SSE event ids in WebSockets and SSE).
Send events only after the change is stored. The transactional outbox from SQL transactions is the reliable form of this rule.
A per-process channel reaches only clients on the same instance. Several instances need a shared source such as Redis pub/sub or a broker topic (messaging and Kafka).

src/main.rs

GraphQL is a Tower service for queries and mutations. GraphQLSubscription upgrades the request and speaks the WebSocket protocol.

Rust

use async_graphql_axum::{GraphQL, GraphQLSubscription};
use axum::{routing::post_service, Router};
use orders_events::schema::build_schema;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = build_schema();

    let app = Router::new()
        .route("/graphql", post_service(GraphQL::new(schema.clone())))
        .route_service("/ws", GraphQLSubscription::new(schema));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Browsers cannot set an Authorization header on a WebSocket. The protocol's connection_init message carries a payload where the client puts its token, and the server validates it once per connection.

Example 2: Testing subscriptions and the schema

schema.execute runs a query without HTTP, and schema.execute_stream returns a stream of results for a subscription. The subscription resolver runs on the first poll, so the test polls once (with a short timeout that elapses) before sending the mutation.

tests/subscription.rs

Rust

use std::time::Duration;

use async_graphql::Request;
use futures_util::StreamExt;
use orders_events::schema::build_schema;

#[tokio::test]
async fn subscriber_receives_status_change() {
    let schema = build_schema();
    let mut stream = Box::pin(schema.execute_stream(Request::new(
        r#"subscription { orderStatusChanged(orderId: "7") { status } }"#,
    )));

    let first_poll = tokio::time::timeout(Duration::from_millis(50), stream.next()).await;
    assert!(first_poll.is_err());

    let response = schema
        .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
        .await;
    assert!(response.errors.is_empty());

    let event = stream.next().await.unwrap();
    assert_eq!(
        event.data.into_json().unwrap(),
        serde_json::json!({ "orderStatusChanged": { "status": "PAID" } })
    );
}

#[tokio::test]
async fn shipped_order_cannot_advance_again() {
    let schema = build_schema();
    for _ in 0..2 {
        let response = schema
            .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
            .await;
        assert!(response.errors.is_empty());
    }

    let response = schema
        .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
        .await;
    assert_eq!(response.errors[0].message, "order already shipped");
}

Two further tests apply: nested list queries run a bounded number of SQL statements (count them in a test database to catch N+1), and a resolver rejects a user who does not own the object. Database-backed tests follow service testing.

Example 3: Schema evolution and an SDL check

GraphQL APIs usually stay on one schema version and grow additively. Clients name the fields they use, so adding a field never breaks an existing query.

Change	Breaking for existing clients
Add a field, type, or optional argument	No
Add a required argument	Yes
Remove a field or change its type	Yes
Make a non-null output nullable	Yes, clients may not handle `null`
Add an enum value	Can break exhaustive matches

Removal follows a process: mark the field deprecated (the total field above uses #[graphql(deprecation = "Use totalCents")]), watch field-level usage until no client requests it, then remove it. A committed SDL file and a test that compares it with the generated SDL make every schema change a visible, deliberate diff. A schema diff tool in CI can report breaking changes. This is the GraphQL form of the reserved field numbers in gRPC basics.

src/bin/print_sdl.rs

Rust

use orders_events::schema::build_schema;

fn main() {
    println!("{}", build_schema().sdl());
}

Generate the committed file with cargo run --bin print_sdl > schema.graphql.

tests/sdl.rs

Rust

use orders_events::schema::build_schema;

#[test]
fn committed_sdl_matches_schema() {
    let committed = include_str!("../schema.graphql");
    assert_eq!(build_schema().sdl().trim(), committed.trim());
}

Example 4: Two federated subgraphs

With several teams, one GraphQL server becomes a codebase everyone edits. Federation lets each service own the schema for its data. Each service is a subgraph, and a gateway combines them into a supergraph that clients query as one graph. A type that several subgraphs contribute fields to is an entity, identified by a key.

Catalog owns Product. Reviews adds a reviews field to Product. In async-graphql, enable_federation() adds the _entities and _service fields that gateways use, and #[graphql(entity)] marks a resolver that builds an entity from its key (the resolver's arguments).

src/bin/catalog_subgraph.rs

Rust

use async_graphql::{EmptyMutation, EmptySubscription, Object, Schema, SimpleObject, ID};
use async_graphql_axum::GraphQL;
use axum::{routing::post_service, Router};

#[derive(SimpleObject, Clone)]
struct Product {
    id: ID,
    name: String,
    price_cents: i32,
}

fn find(id: &str) -> Option<Product> {
    match id {
        "1" => Some(Product {
            id: ID::from("1"),
            name: "Notebook".into(),
            price_cents: 450,
        }),
        _ => None,
    }
}

struct Query;

#[Object]
impl Query {
    async fn product(&self, id: ID) -> Option<Product> {
        find(id.as_str())
    }

    #[graphql(entity)]
    async fn find_product_by_id(&self, id: ID) -> Option<Product> {
        find(id.as_str())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = Schema::build(Query, EmptyMutation, EmptySubscription)
        .enable_federation()
        .finish();

    let app = Router::new().route("/graphql", post_service(GraphQL::new(schema)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:4001").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

src/bin/reviews_subgraph.rs

The Reviews subgraph does not know product names. Its Product holds only the key, and its entity resolver builds a Product from the key alone.

Rust

use async_graphql::{
    ComplexObject, EmptyMutation, EmptySubscription, Object, Schema, SimpleObject, ID,
};
use async_graphql_axum::GraphQL;
use axum::{routing::post_service, Router};

#[derive(SimpleObject)]
struct Review {
    id: ID,
    rating: i32,
}

#[derive(SimpleObject)]
#[graphql(complex)]
struct Product {
    id: ID,
}

#[ComplexObject]
impl Product {
    async fn reviews(&self) -> Vec<Review> {
        match self.id.as_str() {
            "1" => vec![
                Review {
                    id: ID::from("10"),
                    rating: 5,
                },
                Review {
                    id: ID::from("11"),
                    rating: 4,
                },
            ],
            _ => vec![],
        }
    }
}

struct Query;

#[Object]
impl Query {
    #[graphql(entity)]
    async fn find_product_by_id(&self, id: ID) -> Product {
        Product { id }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = Schema::build(Query, EmptyMutation, EmptySubscription)
        .enable_federation()
        .finish();

    let app = Router::new().route("/graphql", post_service(GraphQL::new(schema)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:4002").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

A client of the supergraph sees one Product with name, priceCents, and reviews. For product(id: "1") { name reviews { rating } }, the gateway calls Catalog for product, takes the returned key, and calls Reviews through _entities for the reviews. For lists, it sends all keys to the owning subgraph in one call, which keeps the N+1 rule at the service boundary.

One subgraph owns each field, and every subgraph that declares a key can resolve it.
Subgraphs keep their own databases. Federation joins graphs, not tables (decomposition).
The gateway validates, plans, executes, and merges. It is also where client authentication, depth and complexity limits, and rate limits apply (API gateway).
Composition is checked in CI, so an incompatible subgraph change fails before deploy.
The cost is an extra hop for cross-service fields, a planning step that fans out, and a composition process to run. A modular monolith with one GraphQL server avoids that, and federation pays off when teams must release schema parts independently.

The composition tool and router configuration depend on the federation version. The SDL that enable_federation() emits must be accepted by the chosen composition tool, so check that pairing in the async-graphql documentation and the tool's documentation.

Example 5: GraphQL, REST, and gRPC

Criterion	REST	GraphQL	gRPC
Client picks fields	No	Yes	No
Contract	OpenAPI, optional	Schema, required	`.proto`, required
HTTP caching	Works for GET	Weak, requests are POST	Not available
Streaming	SSE, WebSocket	Subscriptions	Native
Browsers	Native	Native	Needs gRPC-Web
Cost control	Per endpoint	Depth and complexity limits	Per method
Errors	HTTP status	`errors` array, status usually 200	Status codes

GraphQL fits clients that need different slices of the same data. gRPC fits internal typed calls and streams. REST fits public resource APIs with HTTP caching. A GraphQL edge over gRPC services combines them: resolvers call gRPC clients, and DataLoaders batch those calls.

Complete example

The order service with tests, plus the two subgraphs.

Text

orders-events/
  Cargo.toml
  schema.graphql            (generated, see Run)
  src/lib.rs
  src/schema.rs
  src/main.rs
  src/bin/print_sdl.rs
  src/bin/catalog_subgraph.rs
  src/bin/reviews_subgraph.rs
  tests/subscription.rs
  tests/sdl.rs

Cargo.toml

TOML

[package]
name = "orders-events"
version = "0.1.0"
edition = "2021"

[dependencies]
async-graphql = "7.0"
async-graphql-axum = "7.0"
axum = "0.8"
futures-util = "0.3"
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "time", "sync"] }
tokio-stream = { version = "0.1", features = ["sync"] }

src/lib.rs

Rust

pub mod schema;

src/schema.rs

Rust

use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};

use async_graphql::{Context, Enum, Error, Object, Result, Schema, SimpleObject, Subscription, ID};
use futures_util::{Stream, StreamExt};
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;

#[derive(Enum, Copy, Clone, Eq, PartialEq, Debug)]
pub enum OrderStatus {
    Placed,
    Paid,
    Shipped,
}

#[derive(SimpleObject, Clone, Debug)]
pub struct Order {
    pub id: ID,
    pub status: OrderStatus,
    #[graphql(deprecation = "Use totalCents")]
    pub total: f64,
    pub total_cents: i32,
}

pub struct Store {
    orders: Mutex<HashMap<String, Order>>,
}

impl Store {
    pub fn seeded() -> Self {
        let mut orders = HashMap::new();
        orders.insert(
            "7".to_string(),
            Order {
                id: ID::from("7"),
                status: OrderStatus::Placed,
                total: 12.5,
                total_cents: 1250,
            },
        );
        Store {
            orders: Mutex::new(orders),
        }
    }
}

pub struct QueryRoot;

#[Object]
impl QueryRoot {
    async fn order(&self, ctx: &Context<'_>, id: ID) -> Result<Option<Order>> {
        let store = ctx.data::<Arc<Store>>()?;
        let orders = store.orders.lock().unwrap();
        Ok(orders.get(id.as_str()).cloned())
    }
}

pub struct MutationRoot;

#[Object]
impl MutationRoot {
    async fn advance_order(&self, ctx: &Context<'_>, id: ID) -> Result<Order> {
        let store = ctx.data::<Arc<Store>>()?;
        let updated = {
            let mut orders = store.orders.lock().unwrap();
            let order = orders
                .get_mut(id.as_str())
                .ok_or_else(|| Error::new(format!("order {} not found", id.as_str())))?;
            order.status = match order.status {
                OrderStatus::Placed => OrderStatus::Paid,
                OrderStatus::Paid => OrderStatus::Shipped,
                OrderStatus::Shipped => return Err(Error::new("order already shipped")),
            };
            order.clone()
        };
        let events = ctx.data::<broadcast::Sender<Order>>()?;
        let _ = events.send(updated.clone());
        Ok(updated)
    }
}

pub struct SubscriptionRoot;

#[Subscription]
impl SubscriptionRoot {
    async fn order_status_changed(
        &self,
        ctx: &Context<'_>,
        order_id: ID,
    ) -> impl Stream<Item = Order> {
        let receiver = ctx.data_unchecked::<broadcast::Sender<Order>>().subscribe();
        BroadcastStream::new(receiver).filter_map(move |item| {
            let wanted = order_id.clone();
            async move {
                match item {
                    Ok(order) if order.id == wanted => Some(order),
                    _ => None,
                }
            }
        })
    }
}

pub type OrdersSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;

pub fn build_schema() -> OrdersSchema {
    let (events, _) = broadcast::channel::<Order>(64);
    Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
        .data(Arc::new(Store::seeded()))
        .data(events)
        .finish()
}

src/main.rs

Rust

use async_graphql_axum::{GraphQL, GraphQLSubscription};
use axum::{routing::post_service, Router};
use orders_events::schema::build_schema;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = build_schema();

    let app = Router::new()
        .route("/graphql", post_service(GraphQL::new(schema.clone())))
        .route_service("/ws", GraphQLSubscription::new(schema));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

src/bin/print_sdl.rs

Rust

use orders_events::schema::build_schema;

fn main() {
    println!("{}", build_schema().sdl());
}

src/bin/catalog_subgraph.rs

Rust

use async_graphql::{EmptyMutation, EmptySubscription, Object, Schema, SimpleObject, ID};
use async_graphql_axum::GraphQL;
use axum::{routing::post_service, Router};

#[derive(SimpleObject, Clone)]
struct Product {
    id: ID,
    name: String,
    price_cents: i32,
}

fn find(id: &str) -> Option<Product> {
    match id {
        "1" => Some(Product {
            id: ID::from("1"),
            name: "Notebook".into(),
            price_cents: 450,
        }),
        _ => None,
    }
}

struct Query;

#[Object]
impl Query {
    async fn product(&self, id: ID) -> Option<Product> {
        find(id.as_str())
    }

    #[graphql(entity)]
    async fn find_product_by_id(&self, id: ID) -> Option<Product> {
        find(id.as_str())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = Schema::build(Query, EmptyMutation, EmptySubscription)
        .enable_federation()
        .finish();

    let app = Router::new().route("/graphql", post_service(GraphQL::new(schema)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:4001").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

src/bin/reviews_subgraph.rs

Rust

use async_graphql::{
    ComplexObject, EmptyMutation, EmptySubscription, Object, Schema, SimpleObject, ID,
};
use async_graphql_axum::GraphQL;
use axum::{routing::post_service, Router};

#[derive(SimpleObject)]
struct Review {
    id: ID,
    rating: i32,
}

#[derive(SimpleObject)]
#[graphql(complex)]
struct Product {
    id: ID,
}

#[ComplexObject]
impl Product {
    async fn reviews(&self) -> Vec<Review> {
        match self.id.as_str() {
            "1" => vec![
                Review {
                    id: ID::from("10"),
                    rating: 5,
                },
                Review {
                    id: ID::from("11"),
                    rating: 4,
                },
            ],
            _ => vec![],
        }
    }
}

struct Query;

#[Object]
impl Query {
    #[graphql(entity)]
    async fn find_product_by_id(&self, id: ID) -> Product {
        Product { id }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let schema = Schema::build(Query, EmptyMutation, EmptySubscription)
        .enable_federation()
        .finish();

    let app = Router::new().route("/graphql", post_service(GraphQL::new(schema)));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:4002").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

tests/subscription.rs

Rust

use std::time::Duration;

use async_graphql::Request;
use futures_util::StreamExt;
use orders_events::schema::build_schema;

#[tokio::test]
async fn subscriber_receives_status_change() {
    let schema = build_schema();
    let mut stream = Box::pin(schema.execute_stream(Request::new(
        r#"subscription { orderStatusChanged(orderId: "7") { status } }"#,
    )));

    let first_poll = tokio::time::timeout(Duration::from_millis(50), stream.next()).await;
    assert!(first_poll.is_err());

    let response = schema
        .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
        .await;
    assert!(response.errors.is_empty());

    let event = stream.next().await.unwrap();
    assert_eq!(
        event.data.into_json().unwrap(),
        serde_json::json!({ "orderStatusChanged": { "status": "PAID" } })
    );
}

#[tokio::test]
async fn shipped_order_cannot_advance_again() {
    let schema = build_schema();
    for _ in 0..2 {
        let response = schema
            .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
            .await;
        assert!(response.errors.is_empty());
    }

    let response = schema
        .execute(r#"mutation { advanceOrder(id: "7") { status } }"#)
        .await;
    assert_eq!(response.errors[0].message, "order already shipped");
}

tests/sdl.rs

Rust

use orders_events::schema::build_schema;

#[test]
fn committed_sdl_matches_schema() {
    let committed = include_str!("../schema.graphql");
    assert_eq!(build_schema().sdl().trim(), committed.trim());
}

Run

schema.graphql is generated from the schema, so it is not listed above. Create it first, then run the tests:

Terminal

cargo run --bin print_sdl > schema.graphql
cargo test

Start the order service with cargo run --bin orders-events. Requests:

Terminal

curl -s localhost:8080/graphql -H 'content-type: application/json' \
  -d '{"query":"{ order(id: \"7\") { id status totalCents } }"}'

Output

{"data":{"order":{"id":"7","status":"PLACED","totalCents":1250}}}

Open a subscription with a WebSocket client such as websocat (typed lines are the client messages, the others are the replies):

Terminal

websocat --protocol graphql-transport-ws ws://127.0.0.1:8080/ws

Text

{"type":"connection_init"}
{"type":"connection_ack"}
{"id":"1","type":"subscribe","payload":{"query":"subscription { orderStatusChanged(orderId: \"7\") { status } }"}}

In another terminal, advance the order:

Terminal

curl -s localhost:8080/graphql -H 'content-type: application/json' \
  -d '{"query":"mutation { advanceOrder(id: \"7\") { status } }"}'

Output

{"data":{"advanceOrder":{"status":"PAID"}}}

The WebSocket client receives:

Text

{"id":"1","type":"next","payload":{"data":{"orderStatusChanged":{"status":"PAID"}}}}

Two more advances give SHIPPED, and then an error:

Output

{"data":null,"errors":[{"message":"order already shipped","locations":[{"line":1,"column":12}],"path":["advanceOrder"]}]}

Start the subgraphs with cargo run --bin catalog_subgraph and cargo run --bin reviews_subgraph, then query them directly:

Terminal

curl -s localhost:4001/graphql -H 'content-type: application/json' \
  -d '{"query":"{ product(id: \"1\") { name priceCents } }"}'

Output

{"data":{"product":{"name":"Notebook","priceCents":450}}}

Terminal

curl -s localhost:4002/graphql -H 'content-type: application/json' \
  -d '{"query":"query($r: [_Any!]!) { _entities(representations: $r) { ... on Product { reviews { rating } } } }","variables":{"r":[{"__typename":"Product","id":"1"}]}}'

Output

{"data":{"_entities":[{"reviews":[{"rating":5},{"rating":4}]}]}}

Summary

A subscription adds a third root type and delivers results as a stream over WebSocket.
async-graphql subscription methods return impl Stream, and GraphQLSubscription serves them from Axum.
Several instances need a shared event source. Send events after the change is stored.
Federation splits one graph into subgraphs that share entities by key, and a gateway composes and plans.
Evolve schemas additively, deprecate before removing, and compare the generated SDL with a committed file.
schema.execute and schema.execute_stream test queries and subscriptions without HTTP.
The three protocols serve different callers, and systems often use more than one.