第三方 GraphQL 客户端应如何与 Quarkus Smallrye GraphQL 订阅一起使用？

Question

我是 Quarkus + GraphQL 的新手。

我在https://github.com/tigerinus/quakus-graphql-demo 构建了一个 GraphQL 服务，仅用于学习目的

import java.util.Collection;

import javax.inject.Inject;

import org.eclipse.microprofile.graphql.GraphQLApi;
import org.eclipse.microprofile.graphql.Mutation;
import org.eclipse.microprofile.graphql.Query;

import com.wangxiaohu.quarkus.graphql.demo.model.Person;
import com.wangxiaohu.quarkus.graphql.demo.service.PersonService;

import io.quarkus.logging.Log;
import io.smallrye.graphql.api.Subscription;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.operators.multi.processors.BroadcastProcessor;

@GraphQLApi
public class PersonResource {

    private final BroadcastProcessor<Person> _personBroadcastProcessor;

    public PersonResource() {
        _personBroadcastProcessor = BroadcastProcessor.create();
    }

    @Inject
    PersonService _personService;

    @Query("getAllPeople")
    public Collection<Person> getAllPeople() {
        return _personService.getAllPeople();
    }

    @Query("getPersonById")
    public Person getPerson(int id) {
        return _personService.getPerson(id);
    }

    @Mutation("createPerson")
    public Person createPerson(String firstName, String lastName) {
        Person person = _personService.createPerson(firstName, lastName);

        Log.info("signaling the person created...");
        _personBroadcastProcessor.onNext(person);
        Log.info("signaled the person created.");

        return person;
    }

    @Subscription("personCreated")
    public Multi<Person> subscribeToPersonCreation() {
        Log.info("subscribeToPersonCreation");
        return _personBroadcastProcessor;
    }
}

该服务允许添加人员、获取所有人以及订阅人员创建。

我还用 Python 构建了一个小测试代码，以订阅 https://github.com/tigerinus/quakus-graphql-demo/tree/master/test/python 的人员创建

from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport

if __name__ == '__main__':
    transport = WebsocketsTransport(
        url="ws://localhost:8080/graphql",
        subprotocols=[WebsocketsTransport.GRAPHQLWS_SUBPROTOCOL]
    )

    client = Client(transport=transport, fetch_schema_from_transport=True)

    query = gql(
        '''
        subscription subscribeToPersonCreation {
            personCreated{
                id
                firstName
                lastName
            }
        }
        '''
    )

    for result in client.subscribe(query):
        print(result)

但是，由于某种原因，添加新用户不会触发订阅。

我在_personBroadcastProcessor.onNext(person)方法设置了断点，然后我看到subscribers是空的

有人可以告诉我我在这里缺少什么吗？

谢谢！

更新

我还尝试在 nodejs 中编写一个 GraphQL 客户端来订阅，但在创建新的 person 记录时也没有收到任何内容：

const ws = require('ws');
const Crypto = require('crypto');
const { createClient } = require('graphql-ws');

const client = createClient({
  url: "ws://localhost:8080/graphql",
  webSocketImpl: ws,
  generateID: () =>
    ([1e7] + -1e3 + -4e3 + -8e3 + -1e11).replace(/[018]/g, (c) =>
      (c ^ (Crypto.randomBytes(1)[0] & (15 >> (c / 4)))).toString(16),
    ),
});

(async () => {
  const onNext = (value) => {
    console.log(value);
  }

  await new Promise((resolve, reject) => {
    unsubscribe = client.subscribe(
      {
        query: `subscription subscribeToPersonCreation {
          personCreated {
            id
            firstName
            lastName
          }
        }`
      },
      {
        next: onNext,
        error: reject,
        complete: resolve,
      }
    );
  });
})();

在https://github.com/tigerinus/quakus-graphql-demo/tree/master/test/nodejs查看实际代码

Answer 1

它最终成为对如何使用 BroadcastProcessor 的误解。详情请见https://github.com/quarkusio/quarkus/discussions/25967。