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.
