无法通过 SSE 向所有连接发送通知

Question

我正在尝试通过服务器发送的事件通知所有用户事件，但不知道为什么，但它不起作用。如果我使用队列，我可以通知单个用户，但如果我将配置更改为使用主题，它什么也不做。你在下面看到的我在jmsTopicTemplate中使用队列并注释掉template.setPubSubDomain(true(，然后它正在发送给单个用户

@Bean
public Publisher<Message<LocationData>> jmsReactiveSource(ConnectionFactory connectionFactory) {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(connectionFactory)
.destination(JMSConfig.Location_TOPIC).
jmsMessageConverter(messageConverter()))
//.channel(new PublishSubscribeChannel(executor()))                 
//              .channel(MessageChannels.flux())
.channel(MessageChannels.queue())
.toReactivePublisher();
}

您可以在 https://github.com/haiderali22/spring-tracking-jms-sse-mongo-app 查看代码

Answer 1

好吧，不幸的是，我已经检查了您的项目，它仍然足够大，可以正确消化。 但是我从集成的角度来看应该是这样的：

return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(connectionFactory)
.destination(JMSConfig.Location_TOPIC).
jmsMessageConverter(messageConverter())
.autoStartup(false)
.id("jmsMessageDrivenChannelAdapter"))
.toReactivePublisher();

toReactivePublisher()已经自行注入通道。中间不需要任何其他。autoStartup(false)用于推迟订阅到最终Flux。这样，您就不会从 JMS 拉取消息，直到最终订阅发生在Flux。

稍后将在LocationService中使用的.id("jmsMessageDrivenChannelAdapter"))：

public Flux<LocationData> watch() {
return Flux.from(jmsReactiveSource)
.map(Message::getPayload)
.doOnSubscribe(s -> jmsMessageDrivenChannelAdapter.start());
}

这样，您将不会开始从JMS中提取，直到真正的订阅发生到Flux。

JMS 主题与此 SSE 主题无关。

如果你能让你的项目更简单，我会再试一次。 不过我对龙目岛并不熟悉...

更新

使用当前的解决方案，您需要将其设置为这样：

.channel(MessageChannels.flux())
.toReactivePublisher();

问题是，在消息驱动之后，常规.toReactivePublisher()我们只能获得最终发布者的一个订阅者。

要使其成为 pub-sub，您绝对需要在两者之间放置一个FluxMessageChannel。这样，你所有的SEE都会被发送到所有的JavaScript订阅者。