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

Question

A sample project is located here: https://github.com/codependent/spring5-playground

I would like to bridge a message received from a JMS queue into a Reactive Controller that would publish the messages as an event stream.

I don't want the messages to be replayed, that is, if a message arrives and there isn't any subscriber I don't want them to be sent later when any subsbribes, so I am using an 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);
    }
}

This is my 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();
        }
    }
}

And finally my Rest Controller:

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

To test its behaviour I have added this controller method to simulate inserting in the queue:

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

Right after starting the application I load http://localhost:8080/accounts/1/alerts/live2 and the browser waits for data.

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()

Then I publish some messages 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

But none get to the browser. This eventyally finishes with a 500 error (no log).

After some manual retries it starts receiving data...

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)

... but many other times it doesn't get any.

Answer 1

It was resolved in Spring 5 M3:

• https://jira.spring.io/browse/SPR-14803
• https://jira.spring.io/browse/SPR-14772