基于AXUM的GraphQL订阅功能异常，求可参考的实现示例

一、先修正Playground的WebSocket端点配置 #

你当前的Playground未指定订阅用的WebSocket端点，这大概率是订阅无法工作的直接原因。修改Playground生成代码，添加订阅端点配置：

#[allow(clippy::unused_async)]
async fn graphql_playground() -> impl IntoResponse {
    Html(playground_source(
        GraphQLPlaygroundConfig::new("/graphql")
            .subscription_endpoint("/ws"), // 明确指定WebSocket订阅地址
    ))
}

二、完整的订阅实现示例 #

1. 最小化可运行示例 #

以下是包含查询、订阅、Playground的完整基础实现：

use async_graphql::{Schema, Object, Subscription, EmptyMutation, EmptyQuery};
use async_graphql_axum::{GraphQLSubscription, GraphQLRequest, GraphQLResponse};
use axum::{Router, Extension, response::Html};
use async_graphql_playground::playground_source;

// 定义查询根
struct Query;

#[Object]
impl Query {
    async fn hello(&self) -> &str {
        "Hello World!"
    }
}

// 定义订阅根
struct SubscriptionRoot;

#[Subscription]
impl SubscriptionRoot {
    // 一个简单的计数订阅，每秒返回递增数值
    async fn count(&self, #[graphql(default = 1)] step: i32) -> impl async_stream::Stream<Item = i32> {
        async_stream::stream! {
            let mut count = 0;
            loop {
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                count += step;
                yield count;
            }
        }
    }
}

// GraphQL请求处理函数
async fn graphql_handler(
    schema: Extension<Schema<Query, EmptyMutation, SubscriptionRoot>>,
    req: GraphQLRequest
) -> GraphQLResponse {
    schema.execute(req.into_inner()).await.into()
}

#[tokio::main]
async fn main() {
    // 构建Schema
    let schema = Schema::build(Query, EmptyMutation, SubscriptionRoot).finish();

    // 配置Axum路由
    let app = Router::new()
        .route("/graphql", axum::routing::post(graphql_handler))
        .route("/playground", axum::routing::get(graphql_playground))
        .route_service("/ws", GraphQLSubscription::new(schema.clone()))
        .layer(Extension(schema));

    // 启动服务器
    axum::Server::bind(&"0.0.0.0:8000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

2. 带上下文的订阅实现 #

如果需要在订阅中使用数据库连接、认证信息等上下文，可参考以下实现：

use sqlx::PgPool;

// 自定义上下文结构体
struct AppContext {
    db_pool: PgPool,
}

// 订阅中使用上下文示例
#[Subscription]
impl SubscriptionRoot {
    async fn user_updates(
        &self,
        ctx: &async_graphql::Context<'_>,
        user_id: i32
    ) -> impl async_stream::Stream<Item = User> {
        // 从上下文中取出数据库连接池
        let pool = ctx.data_unchecked::<AppContext>().db_pool.clone();
        
        async_stream::stream! {
            // 这里可以替换为真实的数据库变更监听逻辑（比如PostgreSQL LISTEN/NOTIFY）
            loop {
                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                // 模拟从数据库获取更新后的用户数据
                let user = User { id: user_id, name: format!("Updated User {}", user_id) };
                yield user;
            }
        }
    }
}

// 构建Schema时注入上下文
let db_pool = PgPool::connect("postgres://user:pass@localhost/db").await.unwrap();
let schema = Schema::build(Query, EmptyMutation, SubscriptionRoot)
    .data(AppContext { db_pool })
    .finish();

三、其他实用实现思路 #

• 基于外部事件源的订阅：可以将Redis Pub/Sub、Kafka消息、文件系统变化等异步事件源，通过async_stream包装成订阅流
• 路由路径一致性：确保GraphQLSubscription的路由端点（比如/ws）和Playground配置的subscription_endpoint完全匹配，避免路径错误
• 跨域处理：如果Playground和服务端存在跨域，需要为Axum添加CORS中间件，允许WebSocket请求的跨域访问

内容的提问来源于stack exchange，提问作者hbkim