graphql-java-如何在Spring Boot中使用订阅

Question

在一个项目中，我将graphql-java和spring boot与postgreSQL数据库一起使用。现在，我想使用3.0.0版中发布的订阅功能。不幸的是，有关使用订阅功能的信息不是很成熟。

如何通过订阅实现实时功能graphql-java？

Answer 1

从最新的graphql-java版本开始，完全支持订阅。该DataFetcher用于订阅必须返回org.reactivestreams.Publisher，并graphql-java会接管结果映射查询功能的保健。

对该功能进行了很好的记录，并且在正式版本中有使用Web套接字的完整示例。

对于特定于Spring的实现，我无法想象使用Spring自己的事件机制（这里也是一个不错的教程）来支撑Publisher。

每次有传入订阅时，使用应用程序上下文创建并注册一个新的侦听器：context.addApplicationListener(listener)它将发布到正确的Publisher。例如DataFetcher：

// Somehow create a publisher, probably using Spring's Reactor project. Or RxJava.
Publisher<ResultObject> publisher = ...; 
//The listener reacts on application events and pushes new values through the publisher
ApplicationListener listener = createListener(publisher);
context.addApplicationListener(listener);
return publisher;

当Web套接字断开连接或以某种方式知道事件流已完成时，必须确保删除侦听器。

请注意，我实际上还没有尝试过任何一种方法，我只是在大声地思考。

另一种选择是直接用反应器（带或不带弹簧WebFlux）。有使用电抗器和WebSocket的一个样本（通过GraphQL SPQR春天引导初学者）在这里。

您创建一个Publisher像这样的：

//This is really just a thread-safe wrapper around Map<String, Set<FluxSink<Task>>>
private final ConcurrentMultiRegistry<String, FluxSink<Task>> subscribers = new ConcurrentMultiRegistry<>();

@GraphQLSubscription
public Publisher<Task> taskStatusChanged(String taskId) {
    return Flux.create(subscriber -> subscribers.add(taskId, subscriber.onDispose(() -> subscribers.remove(taskId, subscriber))), FluxSink.OverflowStrategy.LATEST);
}

然后从其他地方像这样的（可能是一个相关的突变或反应存储）推新值：

subscribers.get(taskId).forEach(subscriber -> subscriber.next(task));

例如

@GraphQLMutation
public Task updateTask(@GraphQLNonNull String taskId, @GraphQLNonNull Status status) {
    Task task = repo.byId(taskId); //find the task
    task.setStatus(status); //update the task
    repo.save(task); //persist the task
    //Notify all the subscribers following this task
    subscribers.get(taskId).forEach(subscriber -> subscriber.next(task));
    return task;
}

随着SPQR春季首发，这是所有需要让你的阿波罗兼容订阅实现。