我正在使用Sping Boot v3.0.2与websockets，并面临graphQL订阅的问题。任务非常简单-当后端实体更改时通知客户端。所以，我有EntityChangeNotifier -

@Service
public class EntityChangeNotifier {

    private final Collection<Listener<MyEntity>> listeners = new ArrayList<>();

    @Override
    public void notifyChange(MyEntity entity) {
        listeners.forEach(listener -> listener.onEntityChange(entity));
    }

    @Override
    public void registerListener(Listener<T> listener) {
        listeners.add(listener);
    }

    interface Listener<T> {
        void onEntityChange(T entity);
    }

}

服务等级-

@Service
public class Service {

    private final MyRepository repository;

    private final EntityChangeNotifier entityChangeNotifier;

    @Autowired
    public Service(MyRepository repository, EntityChangeNotifier entityChangeNotifier) {
        this.repository = repository;
        this.entityChangeNotifier = entityChangeNotifier;
    }

    @Transactional
    public long saveEntity(MyEntity entity) {
        MyEntity saved = repository.save(entity);
        entityChangeNotifier.notifyChange(entity);
        return saved.getId();
    }

和GraphQLMap-

@Controller
public class Mappings {

    private final EntityChangeNotifier entityChangeNotifier;
    private final Service service;

    @Autowired
    public Mappings(Service service, EntityChangeNotifier entityChangeNotifier) {
        this.service = service;
        this.entityChangeNotifier = entityChangeNotifier;
    }

    @SubscriptionMapping
    public Publisher<MyEntity> changed() {
        return Flux.create(fluxSink -> entityChangeNotifier.registerListener(fluxSink::next));
    }

    @MutationMapping
    public Long create() {
        return service.saveEntity(new MyEntity(0, "testName"));
    }
}

GraphQL schema -

type Mutation {
    create: Int!
}

type Subscription {
    changed: MyEntity
}

type MyEntity {
    id
    name
}

代码是非常基本的，但问题是它不工作。我通过graphiQL测试，保存实体后，订阅只是结束没有任何结果。据我所知，由于某种原因，Flume被标记为取消。你能提示我什么是错的吗？顺便说一句-当我使用冷发布器时，一切正常。但这样的代码，显然，不安慰我-

@SubscriptionMapping
    public Publisher<MyEntity> test() {
        Random random = new Random();
        return Flux.interval(Duration.ofSeconds(5))
                .map(i -> new MyEntity(0, "name" + random.nextInt()));
    }