Spring AMQP - support for Spring programming model with AMQP, especially but not limited to RabbitMQ
@artembilan
Yeah sorry it was missing code, I actually created a targetFactory using Qpid and injected it CachingConnectioFactory everything seems to work when I publish a message, but when I try to receive I see a delay in listener and message is received and then @JmsLisneter crashes I get this QPID error in console
2019-09-04 11:49:48.224 WARN 21725 --- [-6777ebb5153a:1] o.a.qpid.jms.util.QpidJMSThreadFactory : Thread: QpidJMS Connection Executor: ID:5b1e1c76-8c04-4d68-8e89-6777ebb5153a:1 failed due to an uncaught exception: null
2019-09-04 11:49:48.224 INFO 21725 --- [-6777ebb5153a:1] org.apache.qpid.jms.JmsSession : A JMS MessageConsumer has been closed: JmsConsumerInfo: { ID:5b1e1c76-8c04-4d68-8e89-6777ebb5153a:1:4:1, destination = inbound1 }
2019-09-04 11:49:48.225 WARN 21725 --- [-6777ebb5153a:1] o.a.qpid.jms.util.QpidJMSThreadFactory : Thread: QpidJMS Connection Executor: ID:5b1e1c76-8c04-4d68-8e89-6777ebb5153a:1 failed due to an uncaught exception: null
2019-09-04 11:49:53.216 WARN 21725 --- [enerContainer-7] o.s.j.l.DefaultMessageListenerContainer : Setup of JMS message listener invoker failed for destination 'randomQueueName' - trying to recover. Cause: Remote did not respond to a drain request in time
2019-09-04 11:49:58.220 WARN 21725 --- [nerContainer-12] o.s.j.l.DefaultMessageListenerContainer : Setup of JMS message listener invoker failed for destination 'randomQueueName' - trying to recover. Cause: The MessageConsumer was closed due to an unrecoverable error.
2019-09-04 11:50:03.226 WARN 21725 --- [nerContainer-21] o.s.j.l.DefaultMessageListenerContainer : Setup of JMS message listener invoker failed for destination 'randomQueueName' - trying to recover. Cause: The MessageConsumer was closed due to an unrecoverable error.
2019-09-04 11:50:04.590 INFO 21725 --- [n(14)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2019-09-04 11:50:04.592 WARN 21725 --- [nerContainer-22] o.s.j.l.DefaultMessageListenerContainer : Setup of JMS message listener invoker failed for destination 'randomQueueName' - trying to recover. Cause: The MessageConsumer was closed due to an unrecoverable error.
2019-09-04 11:50:34.597 INFO 21725 --- [n(14)-127.0.0.1] o.s.c.support.DefaultLifecycleProcessor : Failed to shut down 1 bean with phase value 2147483647 within timeout of 30000: [org.springframework.jms.config.internalJmsListenerEndpointRegistry]
2019-09-04 11:50:34.602 INFO 21725 --- [n(14)-127.0.0.1] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor'Remember I am trying to use Qpid JMS AMQP 1.0 with Spring to connect rabbitmq over AMQP1.0 prorocol
This actually works for other MQs which support AMQP1.0 protocol
@Configuration
public class RabbitTemplateConfiguration {
RabbitTemplateConfiguration(RabbitTemplate rabbitTemplate){
rabbitTemplate.setAfterReceivePostProcessors(new AuthenticationAwareMessagePostProcessor());
}
}AuthenticationAwareMessagePostProcessor is never called
setBeforePublishPostProcessors it is being called, but I need the "filter" on the receiving side
@yuvalprtn Works fine for me...
@SpringBootApplication
public class Gitter53Application {
public static void main(String[] args) {
SpringApplication.run(Gitter53Application.class, args);
}
@Bean
public ApplicationRunner runner(RabbitTemplate template) {
return args -> {
template.convertAndSend("foo", "bar");
System.out.println(template.receive("foo", 10_000L));
};
}
}
@Configuration
class RabbitTemplateConfiguration {
RabbitTemplateConfiguration(RabbitTemplate rabbitTemplate) {
rabbitTemplate.setAfterReceivePostProcessors(msg -> {
System.out.println("In MPP: " + msg);
return msg;
});
}
}In MPP: (Body:'bar' ...(Body:'bar' ...
@RabbitListeners
@SpringBootApplication
public class Gitter53Application {
public static void main(String[] args) {
SpringApplication.run(Gitter53Application.class, args);
}
@Bean
public ApplicationRunner runner(RabbitTemplate template) {
return args -> {
template.convertAndSend("foo", "bar");
};
}
@RabbitListener(queues = "foo")
public void listen(String in) {
System.out.println(in);
}
}
@Configuration
class ContainerFactoryConfiguration {
ContainerFactoryConfiguration(AbstractRabbitListenerContainerFactory<?> factory) {
factory.setAfterReceivePostProcessors(msg -> {
System.out.println("In MPP: " + msg);
return msg;
});
}
}
WS-OUTPUT-BUFFER:RFH * 4MQSTR Ï´ ∑∞é¥û∞(Ω¥û|Ω™æµ¬æ∞(Ω¥û∞é¥û∑∑ ë∞|Ωû∞ÖΩæûêÛµÛµÈä§É†âÉ
Ñ<…§(â§+äàâäà†&<·∞ÖΩæû∞äΩû⁄fl‡Â›Âfi„fi∞äΩû∞Ñ•¥û§ÖÈ雵¥„fiŸé›µ¥„µfiŸfiŸfiŸfiŸfiŸfiŸfiŸfiŸ¥‡éfiflé›∂éé„„fi„⁄Ÿ∞Ñ•¥û∞Ö%¿û∞Ö
%¿û∞|_Ωû ∫∞ÛΩºû∞ÑÑâ§Ö∑¥æ±•fiû„⁄Â∞ÑÑâ§Öû∞∂/•%§∂ÇÛ•µΩéµ∑¥æ±•fiû⁄∞∂/•%§∂ÇÛ•µΩéµû∞µ>é?¥•>è∑¥æ±•fiû›∞µ>é?¥•>èû∞ÛΩ
ºû∑∑∑<?xml version="1.0" encoding="UTF-8"?>
@priyanshus1 Because Spring AMQP has had its own recovery mechanism that predates auto recovery by several years (since 2010); we tried to make it compatible with auto recovery but it was like whack-a-mole to kill the bugs; even if it's enabled now, we kill the recovery.
The listener container goes into recovery mode after a connection failure. If you are not seeing that, something is wrong; turn on DEBUG logging to figure out what's going on. If you can't figure it out, provide a simple example that exhibits the behaviour you see.
There should be no circumstances where recovery doesn't work. One exception to that is the use of auto-delete queues on an HA cluster, where the master-locator is not client-local (i.e. the queue is created on a different node to the one we are connected to and that node goes down). The container will stop itself after 15 seconds if it can't reconnect to its queue.
We changed the AnonymousQueue to use client-local by default in version 2.1.
If you are creating your own auto-delete queues, you must set the master-locator, either as an argument or a policy on the broker.