Spring 5 Web Reactive - 热发布 - 如何使用 EmitterProcessor 将 MessageListener 桥接到事件流

Spring 5 Web Reactive - Hot Publishing - How to use EmitterProcessor to bridge a MessageListener to an event stream

示例项目位于此处:https://github.com/codependent/spring5-playground

我想将从 JMS 队列接收到的消息桥接到反应式控制器中,后者将消息发布为事件流。

我不希望消息被重播,也就是说,如果消息到达并且没有任何订阅者,我不希望它们稍后在任何订阅时发送,所以我正在使用 EmitterProcessor :

@Component
public class AlertEmitterProcessor {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private EmitterProcessor<Alert> processor;

    public AlertEmitterProcessor(){
        processor = EmitterProcessor.<Alert>create();
        processor.connect();
    }

    public EmitterProcessor<Alert> getProcessor() {
        return processor;
    }

    public void onNext(Alert alert){
        logger.info("onNext [{}]", alert);
        processor.onNext(alert);
    }

    public void onComplete(){
        logger.info("onComplete");
        processor.onComplete();
    }

    public void onError(Throwable t){
        logger.error("onError", t);
        processor.onError(t);
    }
}

这是我的MessageListener:

@Component
public class AlertMessageListener implements MessageListener{

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired 
    private AlertEmitterProcessor alertProcessor;

    @Autowired
    private MappingJackson2HttpMessageConverter jacksonMessageConverter;

    @Override
    public void onMessage(Message message) {
        logger.info("Message received: [{}]", message);
        TextMessage tm = (TextMessage)message;
        try {
            Alert alert = jacksonMessageConverter.getObjectMapper().readValue(tm.getText(), Alert.class);
            alertProcessor.onNext(alert);
        } catch (IOException | JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

最后是我的休息控制器:

@Autowired
private AlertEmitterProcessor alertTopicProcessor;

@Autowired 
private AlertMessageListener messageListener;

@Autowired
private MappingJackson2HttpMessageConverter jacksonMessageConverter;

@GetMapping(value="/accounts/{id}/alerts/live2", produces="text/event-stream")
public Flux<Alert> getAccountAlertsStreaming2(@PathVariable Integer id) {
    return alertTopicProcessor.getProcessor()
        .log().filter( a -> a.getAccountId().equals(id) );
}

为了测试它的行为,我添加了这个控制器方法来模拟插入队列:

@GetMapping(value="/mock/accounts/{id}/alerts/put", produces="text/event-stream")
public void putAlert(@PathVariable Integer id) throws JsonProcessingException {
    Alert alert = new Alert(id, (long)Math.round(Math.random()*10), "Message");
    String alertStr = jacksonMessageConverter.getObjectMapper().writeValueAsString(alert);
    TextMessage tm = new MockTextMessage(alertStr);
    messageListener.onMessage(tm);
}

启动应用程序后,我立即加载 http://localhost:8080/accounts/1/alerts/live2,浏览器等待数据。

2016-10-03 13:43:38.755 DEBUG 12800 --- [nio-8080-exec-1] o.s.web.reactive.DispatcherHandler       : Processing GET request for [http://localhost:8080/accounts/1/alerts/live2]
2016-10-03 13:43:38.770 DEBUG 12800 --- [nio-8080-exec-1] s.w.r.r.m.a.RequestMappingHandlerMapping : Looking up handler method for path /accounts/1/alerts/live2
2016-10-03 13:43:38.778 DEBUG 12800 --- [nio-8080-exec-1] s.w.r.r.m.a.RequestMappingHandlerMapping : Returning handler method [public reactor.core.publisher.Flux<com.codependent.spring5.playground.reactive.dto.Alert> com.codependent.spring5.playground.reactive.web.AccountsRestController.getAccountAlertsStreaming2(java.lang.Integer)]
2016-10-03 13:43:38.779 DEBUG 12800 --- [nio-8080-exec-1] o.s.b.f.s.DefaultListableBeanFactory     : Returning cached instance of singleton bean 'accountsRestController'
2016-10-03 13:43:38.800  INFO 12800 --- [nio-8080-exec-1] reactor.unresolved                       : onSubscribe(reactor.core.publisher.FluxPeek$PeekSubscriber@54d4fb36)
2016-10-03 13:43:38.802  INFO 12800 --- [nio-8080-exec-1] reactor.unresolved                       : request(unbounded)
2016-10-03 13:43:38.803  INFO 12800 --- [nio-8080-exec-1] reactor.unresolved                       : onNext(1)
2016-10-03 13:43:38.822  INFO 12800 --- [nio-8080-exec-1] reactor.Flux.EmitterProcessor.2          : onSubscribe(reactor.core.publisher.EmitterProcessor$EmitterSubscriber@227405f2)
2016-10-03 13:43:38.822  INFO 12800 --- [nio-8080-exec-1] reactor.Flux.EmitterProcessor.2          : request(1)
2016-10-03 13:43:38.823  INFO 12800 --- [nio-8080-exec-1] reactor.unresolved                       : onComplete()

然后我发布一些消息http://localhost:8080/mock/accounts/1/alerts/put

2016-10-03 13:43:43.063 DEBUG 12800 --- [nio-8080-exec-2] o.s.web.reactive.DispatcherHandler       : Processing GET request for [http://localhost:8080/mock/accounts/1/alerts/put]
2016-10-03 13:43:43.063 DEBUG 12800 --- [nio-8080-exec-2] s.w.r.r.m.a.RequestMappingHandlerMapping : Looking up handler method for path /mock/accounts/1/alerts/put
2016-10-03 13:43:43.068 DEBUG 12800 --- [nio-8080-exec-2] s.w.r.r.m.a.RequestMappingHandlerMapping : Returning handler method [public void com.codependent.spring5.playground.reactive.web.AccountsRestController.putAlert(java.lang.Integer) throws com.fasterxml.jackson.core.JsonProcessingException]
2016-10-03 13:43:43.069 DEBUG 12800 --- [nio-8080-exec-2] o.s.b.f.s.DefaultListableBeanFactory     : Returning cached instance of singleton bean 'accountsRestController'
2016-10-03 13:43:43.071  INFO 12800 --- [nio-8080-exec-2] reactor.unresolved                       : onSubscribe(reactor.core.publisher.FluxPeek$PeekSubscriber@2ba7d09c)
2016-10-03 13:43:43.071  INFO 12800 --- [nio-8080-exec-2] reactor.unresolved                       : request(unbounded)
2016-10-03 13:43:43.072  INFO 12800 --- [nio-8080-exec-2] reactor.unresolved                       : onNext(1)
2016-10-03 13:43:43.112  INFO 12800 --- [nio-8080-exec-2] c.c.s.p.r.message.AlertMessageListener   : Message received: [com.codependent.spring5.playground.reactive.message.MockTextMessage@37262c9e]
2016-10-03 13:43:43.145  INFO 12800 --- [nio-8080-exec-2] c.c.s.p.r.message.AlertEmitterProcessor  : onNext [Alert [alertId=3, message=Message, accountId=1]]
2016-10-03 13:43:43.146  INFO 12800 --- [nio-8080-exec-2] reactor.Flux.EmitterProcessor.2          : onNext(Alert [alertId=3, message=Message, accountId=1])
2016-10-03 13:43:43.177  INFO 12800 --- [nio-8080-exec-2] reactor.unresolved                       : onComplete()
2016-10-03 13:43:43.177 DEBUG 12800 --- [nio-8080-exec-2] o.s.h.s.r.ServletHttpHandlerAdapter      : Successfully completed request

但是 none 进入浏览器。这最终以 500 错误(无日志)结束。

经过一些手动重试后,它开始接收数据...

2016-10-03 13:45:07.726 DEBUG 12800 --- [nio-8080-exec-8] s.w.r.r.m.a.RequestMappingHandlerMapping : Returning handler method [public reactor.core.publisher.Flux<com.codependent.spring5.playground.reactive.dto.Alert> com.codependent.spring5.playground.reactive.web.AccountsRestController.getAccountAlertsStreaming2(java.lang.Integer)]
2016-10-03 13:45:07.726 DEBUG 12800 --- [nio-8080-exec-8] o.s.b.f.s.DefaultListableBeanFactory     : Returning cached instance of singleton bean 'accountsRestController'
2016-10-03 13:45:07.727  INFO 12800 --- [nio-8080-exec-8] reactor.unresolved                       : onSubscribe(reactor.core.publisher.FluxPeek$PeekSubscriber@909f06f)
2016-10-03 13:45:07.727  INFO 12800 --- [nio-8080-exec-8] reactor.unresolved                       : request(unbounded)
2016-10-03 13:45:07.727  INFO 12800 --- [nio-8080-exec-8] reactor.unresolved                       : onNext(1)
2016-10-03 13:45:07.729  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : onSubscribe(reactor.core.publisher.EmitterProcessor$EmitterSubscriber@7ce1f3e)
2016-10-03 13:45:07.729  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : request(1)
2016-10-03 13:45:07.729  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : onNext(Alert [alertId=4, message=Message, accountId=1])
2016-10-03 13:45:07.730  INFO 12800 --- [nio-8080-exec-8] reactor.unresolved                       : onComplete()
2016-10-03 13:45:07.747  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : request(1)
2016-10-03 13:45:07.747  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : onNext(Alert [alertId=0, message=Message, accountId=1])
2016-10-03 13:45:07.748  INFO 12800 --- [nio-8080-exec-8] reactor.Flux.EmitterProcessor.9          : request(1)

...但很多时候它什么也得不到。

已在 Spring 5 M3:

中解决