graphql-java -如何在spring启动时使用订阅？

Question

在一个项目中，我使用graphql-java和spring引导和postgreSQL数据库。现在，我想使用版本3.0.0中发布的订阅功能。遗憾的是，关于辅助函数应用的信息还不是很成熟。

如何使用带有订阅的实现的实时功能？

Answer 1

在最近的graphql版本中，完全支持订阅。订阅的DataFetcher必须返回一个org.reactivestreams.Publisher，graphql将负责将查询函数映射到结果上。

这个特性是很好的记录在案，还有一个完整的例子使用官方回购中可用的网络套接字。

如果您已经设置了一个反应性数据源(例如，带有反应性驱动程序的Mongo，或者R2DBC支持的任何东西)，那么您就都准备好了。只需使用@Tailable，Spring数据就会为您提供一个Flux (它实现了Publisher)，并且没有其他需要做的事情。

至于更手动的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;
}

有了Spring，这就是为您提供与阿波罗兼容的订阅实现所需的全部内容。

Answer 2

我也遇到了同样的问题，我在库中加入了spring引导。但是，我发现graphql似乎只支持模式级的“订阅”，它不对此特性执行任何跨国支持。这意味着你可能需要自己去实现它。

请参阅https://github.com/graphql-java/graphql-java/blob/master/docs/schema.rst#subscription-support

Answer 3

记录:下面是另一个非常好的、紧凑的示例，它实现了GraphQLs的基本特性查询、突变和订阅：https://github.com/npalm/blog-graphql-spring-service