DGS Subscription using Spring Integration on an ActiveMQ Topic

[predhme]

I have the question fairly well documented on Stackoverflow, but I wanted try to here as well. Essentially I have a DgsComponent that exposes a subscription:

@DgsComponent
public class SurveyResultsSubscriptionDataFetcher {

    // a ReactiveMongoRepository from Spring Data Mongo to prove the subscription works
    private final SurveyResultRepository repository;

    // injected bean - shown later
    private final Publisher<SurveyResult> surveyResultsReactiveSource;

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

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

Inside my Spring Configuration I have the Publisher<SurveyResult> defined. Essentially it is a Spring Integration Flow. It is defined as the following:

    @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));
    }

I will note that I have a different @JmsListener (non spring integration variant) that is receiving the messages off the topic and tossing them in the database. After I establish a web socket connection with my client, I would expect to see an additional consumer on the topic in the ActiveMQ console, but I do not.

I know that the subscription is properly working as if I swap the surveyResultsReactiveSource for the repository, I get a stream of results on the client. I am wondering if there is something I should be doing in DGS land to "activate" the source. I would suppose whatever subscriber DGS supplies would do the trick, but perhaps I am missing something. Any thoughts?

P.S I absolutely love DGS, thank you developers for all the hard work!!

[predhme]

The stackoverflow link I provided in the question has the answer. The problem was with my Spring Integration Flow configuration and nothing to do with DGS.

[srinivasankavitha]

Thanks for the update.

[predhme]

If anyone is interested, I have since updated the data fetcher to use the Reactive Mongo Repository Tailable Cursor feature to provide live updates. Much simpler solution. I just return that Flux in my DGS Component and magic happens.