How can I convert the following code using the Spring framework?
ConnectionFactory factory = new ConnectionFactory();
factory.setExceptionHandler(new BrokerExceptionHandler(logger, instance));
public final class BrokerExceptionHandler extends StrictExceptionHandler {
@Override
public void handleReturnListenerException(Channel channel, Throwable exception) {
logger.log(Level.SEVERE, "ReturnListenerException detected: ReturnListener.handleReturn", exception);
this.publishAlert(exception, "ReturnListener.handleReturn");
logger.log(Level.SEVERE, "Close application", exception);
System.exit(-1);
}
....
}
Basically I need to specify a custom exception handler if a rabbitMQ exception occurs and then stop the application
How can I publish a rabbitMq message every time that there is an exception?
EDIT
I modified my configuration class in this way:
@Bean
SimpleMessageListenerContainer containerPredict(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerPredictAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setDefaultRequeueRejected(false);
container.setErrorHandler(new BrokerExceptionHandler());
container.setQueueNames(getQueueName());
container.setMessageListener(listenerAdapter);
return container;
}
and this is my BrokerExceptionHandler class
public class BrokerExceptionHandler implements ErrorHandler {
private final Logger logger = Logger.getLogger(getClass().getSimpleName());
@Autowired
private Helper helper;
@Override
public void handleError(Throwable t) {
logger.log(Level.SEVERE, "Exception Detected. Publishing error alert");
String message = "Exception detected. Message: " + t.getMessage());
// Notify the error to the System sending a new RabbitMq message
System.out.println("---> Before convertAndSend");
rabbitTemplate.convertAndSend(exchange, routing, message);
System.out.println("---> After convertAndSend");
}
}
I can see the log Exception Detected. Publishing error alert and ---> Before convertAdnSend in the console, but the new alert is not published and the log ---> After convertAndSend doesn't appear in the console.
Here it is the log:
2018-10-17 09:32:02.849 ERROR 1506 --- [tainer****-1] BrokerExceptionHandler : Exception Detected. Publishing error alert
---> Before convertAndSend
2018-10-17 09:32:02.853 INFO 1506 --- [tainer****-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@4f5b08d: tags=[{amq.ctag-yUcUmg5BCo20ucG1wJZoWA=myechange}], channel=Cached Rabbit Channel: AMQChannel(amqp://admin@XXX.XXX.XXX.XXX:5672/testbed_simulators,1), conn: Proxy@3964d79 Shared Rabbit Connection: SimpleConnection@61f39bb [delegate=amqp://admin@XXX.XXX.XXX.XXX:5672/testbed_simulators, localPort= 51528], acknowledgeMode=AUTO local queue size=0
2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (myexchange) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (myexchange) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
EDIT
Debugging I see that before to send the new message, the following code is called:
File: SimpleMessageListenerContainer.class line 1212
if (!isActive(this.consumer) || aborted) {
.....
}
else {
---> logger.info("Restarting " + this.consumer);
restart(this.consumer);
}
EDIT 2
Example code: http://github.com/fabry00/spring-boot-rabbitmq
It depends on how you are doing your configuration; if you are using Spring Boot's auto-configured connection factory...
@Bean
public InitializingBean connectionFactoryConfigurer(CachingConnectionFactory ccf) {
return () -> ccf.getRabbitConnectionFactory().setExceptionHandler(...);
}
If you are wiring up your own beans (e.g. via a RabbitConnectionFactoryBean) then set it directly.
EDIT
You are throwing a NullPointerException in your error handler...
2018-10-17 11:51:58.733 DEBUG 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Consumer raised exception, processing can restart if the connection factory supports it
java.lang.NullPointerException: null
at com.test.BrokerExceptionHandler.handleError(BrokerExceptionHandler.java:27) ~[main/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1243) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1488) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1318) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]
2018-10-17 11:51:58.734 INFO 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@1aabf50d: tags=[{amq.ctag-VxxHKiMsWI_w8DIooAsySA=myapp.mydomain.KPIS}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@b88a7d6 Shared Rabbit Connection: SimpleConnection@25dc64a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55662], acknowledgeMode=AUTO local queue size=0
To turn on DEBUG logging, add
logging.level.org.springframework.amqp=debug
to your application.properties.
this.helper is null because the error handler is not a Spring Bean - @Autowired only works if Spring manages the object; you are using new BrokerExceptionHandler().
EDIT2
I added these 2 beans
@Bean
public BrokerExceptionHandler errorHandler() {
return new BrokerExceptionHandler();
}
@Bean
public MessageConverter json() { // Boot auto-configures in template
return new Jackson2JsonMessageConverter();
}
and now...
---> Before publishing Alert event
--- ALERT
2018-10-17 12:14:45.304 INFO 43359 --- [containerKpis-1] Helper : publishAlert
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.a.r.c.CachingConnectionFactory : Creating cached Rabbit Channel from AMQChannel(amqp://guest@127.0.0.1:5672/,3)
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate : Executing callback RabbitTemplate$$Lambda$638/975724213 on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56220]
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate : Publishing message (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=146, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [myevent.ALERT], routingKey = [/]
--- ALERT 2
---> After publishing Alert event
2018-10-17 12:14:45.323 DEBUG 43359 --- [pool-1-thread-6] o.s.a.r.listener.BlockingQueueConsumer : Storing delivery for consumerTag: 'amq.ctag-eYbzZ09pCw3cjdtSprlZMQ' with deliveryTag: '1' in Consumer@4b790d86: tags=[{amq.ctag-eYbzZ09pCw3cjdtSprlZMQ=myapp.myevent.ALERT}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56220], acknowledgeMode=AUTO local queue size=0
2018-10-17 12:14:45.324 DEBUG 43359 --- [ontainerReset-1] o.s.a.r.listener.BlockingQueueConsumer : Received message: (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=myevent.ALERT, receivedRoutingKey=/, deliveryTag=1, consumerTag=amq.ctag-eYbzZ09pCw3cjdtSprlZMQ, consumerQueue=myapp.myevent.ALERT])
2018-10-17 12:14:45.324 INFO 43359 --- [ontainerReset-1] Application : ---> kpisAlert RECEIVED
2018-10-17 12:14:45.325 ERROR 43359 --- [ontainerReset-1] Application : ---> Message: Exception detected. Message: Listener method 'kpisEvent' threw exception
2018-10-17 12:14:45.326 DEBUG 43359 --- [containerKpis-1] o.s.a.r.listener.BlockingQueueConsumer : Rejecting messages (requeue=false)
EDIT3
Or, if you prefer Gson...
@Bean
public MessageConverter json() {
Gson gson = new GsonBuilder().create();
return new MessageConverter() {
@Override
public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
return new Message(gson.toJson(object).getBytes(), messageProperties);
}
@Override
public Object fromMessage(Message message) throws MessageConversionException {
throw new UnsupportedOperationException();
}
};
}
EDIT4
I changed the current version of your app as follows:
@Bean
public MessageConverter jsonConverter() {
Gson gson = new GsonBuilder().create();
EventKpisCollected collected = new EventKpisCollected();
return new MessageConverter() {
@Override
public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
System.out.println("toMessage");
return new Message(gson.toJson(object).getBytes(), messageProperties);
}
@Override
public Object fromMessage(Message message) throws MessageConversionException {
System.out.println("fromMessage");
return collected.decode(new String(message.getBody()));
}
};
}
...
@Bean
SimpleMessageListenerContainer containerKpis(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerKpisAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setDefaultRequeueRejected(false);
container.setErrorHandler(errorHandler());
container.setQueueNames(getQueueKpis());
container.setMessageListener(listenerKpisAdapter);
return container;
}
@Bean
SimpleMessageListenerContainer containerReset(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerAlertAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setDefaultRequeueRejected(false);
container.setErrorHandler(errorHandler());
container.setQueueNames(getQueueAlert());
container.setMessageListener(listenerAlertAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerKpisAdapter(Application receiver) {
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "kpisEvent");
messageListenerAdapter.setMessageConverter(jsonConverter());
return messageListenerAdapter;
}
@Bean
MessageListenerAdapter listenerAlertAdapter(Application receiver) {
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "alertEvent");
// messageListenerAdapter.setMessageConverter(jsonConverter()); converter only handles events.
return messageListenerAdapter;
}
and
fromMessage
2018-10-19 13:46:53.734 INFO 10725 --- [containerKpis-1] Application : ---> kpisEvent RECEIVED
2018-10-19 13:46:53.734 INFO 10725 --- [containerKpis-1] Application : ---> kpisEvent DECODED, windowId: 1522751098000-1522752198000
With the event decoding done by the framework (just for the event currently - you will need a second converter for the alers).