我有一个有点复杂的技术堆栈。我正在利用netflix dgs提供graphql服务。幕后是一堆从各种服务发送和接收数据的jms组件。除了graphql订阅之外，我的一切都在工作。
具体来说，我要做的是为来自activemq主题的消息创建一个graphql订阅。
所以我有一个subscriptiondatafetcher，如下所示：

@DgsComponent
public class SurveyResultsSubscriptionDataFetcher {

    private final Publisher<SurveyResult> surveyResultsReactiveSource;

    @Autowired
    public SurveyResultsSubscriptionDataFetcher(Publisher<SurveyResult> surveyResultsReactiveSource) {
        this.surveyResultsReactiveSource = surveyResultsReactiveSource;
    }

    @DgsData(parentType = DgsConstants.SUBSCRIPTION.TYPE_NAME, field = DgsConstants.SUBSCRIPTION.SurveyResultStream)
    public Publisher<SurveyResult> surveyResults() {
        return surveyResultsReactiveSource;
    }
}

在我的spring配置中，我使用以下spring集成流：

@Bean
@Scope(BeanDefinition.SCOPE_PROTOTYPE)
public Publisher<SurveyResult> surveyResultsReactiveSource() {
    SurveyResultMessageConverter converter = new SurveyResultMessageConverter();

    return Flux.from(
            IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory()).destination(surveyDestination))
                    .log(LoggingHandler.Level.DEBUG)
                    .log()
                    .toReactivePublisher())
            .map((message) -> converter.fromMessage(message, SurveyResult.class));
}

我要说几句话：
我有一个单独的房间 @JmsListener 也就是说收到这些离题的信息
即使建立了web套接字连接，我也看不到多个使用者。
如果我将mongo-reactive-spring数据存储库连接到这个graphql订阅，那么客户机将接收数据。
当我将客户端连接到订阅时，会看到以下日志：

PublishSubscribeChannel      : Channel 'unknown.channel.name' has 1 subscriber(s).
DgsWebSocketHandler          : Subscription started for 1

我怀疑在建立web套接字连接时消息侦听器容器没有被激活。我应该“激活”通道适配器吗？我错过了什么？
技术堆栈：

// spring boots - version 2.4.3
    implementation "org.springframework.boot:spring-boot-starter-web"
    implementation "org.springframework.boot:spring-boot-starter-activemq"
    implementation "org.springframework.boot:spring-boot-starter-data-mongodb-reactive"
    implementation "org.springframework.boot:spring-boot-starter-integration"
    implementation 'org.springframework.boot:spring-boot-starter-security'

    // spring integration
    implementation group: 'org.springframework.integration', name: 'spring-integration-jms', version: '5.4.4'

    // dgs
    implementation "com.netflix.graphql.dgs:graphql-dgs-spring-boot-starter:3.10.2"
    implementation 'com.netflix.graphql.dgs:graphql-dgs-subscriptions-websockets-autoconfigure:3.10.2'

更新1：
不管它值多少钱，如果我更新订阅到以下内容，我会在客户端得到结果。

@DgsData(parentType = DgsConstants.SUBSCRIPTION.TYPE_NAME, field = DgsConstants.SUBSCRIPTION.SurveyResultStream)
    public Publisher<SurveyResult> surveyResults() {
        // repository is a ReactiveMongoRepository
        return repository.findAll();
    }

更新2：
这是最终确定的bean，以防它根据公认的解决方案帮助某人。我需要一个主题的听众，而不是一个队列。

@Bean
    public Publisher<Message<SurveyResult>> surveyResultsReactiveSource() {
        SurveyResultMessageConverter converter = new SurveyResultMessageConverter();

        return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(
                Jms.container(connectionFactory(), surveyDestination).pubSubDomain(true))
                .jmsMessageConverter(converter))
                .toReactivePublisher();
    }