spring-projects/spring-amqp

Spring AMQP - support for Spring programming model with AMQP, especially but not limited to RabbitMQ

Artem Bilan

@artembilan

I think so

but you don't need to handle it yourself

let me check the test-case

Scott White

@kibbled

awesome, thanks!

Artem Bilan

@artembilan

Yes, it is there:

@Bean
        public SimpleRabbitListenerContainerFactory dontRequeueFactory() {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(rabbitConnectionFactory());
            factory.setMismatchedQueuesFatal(true);
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            factory.setMessageConverter(converter());
            factory.setDefaultRequeueRejected(false);
            return factory;
        }

So, since Mono might be processed in the separate from the listener container thread, we can't rely on the AUTO and really deffer ack'ing up to the success of that mono process

Scott White

@kibbled

  rabbitmq:
    listener:
      # Using simple listener type because RabbitMQ Reactive framework requires it
      type: simple
      simple:
        acknowledge-mode: manual
        default-requeue-rejected: false
        prefetch: 1

so config can remain the same?

Artem Bilan

@artembilan

Looks like

But your code should not use those basicAck/basicReject

However I believe you can do anything else in those callbacks

Scott White

@kibbled

@artembilan yea that’s fine by me, less code plus I can remove the channel & deliveryTag arguments to my function

thanks

Artem Bilan

@artembilan

Yep! Let's hope it works as you expected!

Sorry for confusing you in the beginning :smile:

Scott White

@kibbled

yea no problem, it was a bit of work getting reactor handler to work properly in the beginning, I guess we were an early adopter. :)

Scott White

@kibbled

@artembilan One more question, if a straightup Java exception is throw before the reactor sequence this would also be handled? Or should it be caught and returned as a Mono error?

Artem Bilan

@artembilan

No, it is going to be handled via returnOrThrow(amqpMessage, channel, message, e.getCause(), e);

We deal with Mono only when a normal return from the @RabbitListener method

Priyanshu Shukla

@priyanshus1

Hello Folks,
I wanna connect to RabbitMQ AMQP 1.0 using Spring JMS.
I created a ConnectionFactory but unable to configure it properly
getting this error

Caused by: java.lang.IllegalArgumentException: Target Connection or ConnectionFactory is required
    at org.springframework.jms.connection.SingleConnectionFactory.afterPropertiesSet(SingleConnectionFactory.java:234) ~[spring-jms-5.1.9.RELEASE.jar:5.1.9.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1837) ~[spring-beans-5.1.9.RELEASE.jar:5.1.9.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1774) ~[spring-beans-5.1.9.RELEASE.jar:5.1.9.RELEASE]
    ... 44 common frames omitted

Any hints what am I doing wrong ?

Thanks in advance

vanseverk

@vanseverk

@priyanshus1 kind of hard to without seeing your code :-)

Priyanshu Shukla

@priyanshus1

Hi @vanseverk
here is how my sample config looks like

@Configuration
@EnableJms
public class MQConfig {

    @Bean
    public MessageConverter jsonMessageConverter(ObjectMapper objectMapper) {
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
        messageConverter.setObjectMapper(objectMapper);
        return messageConverter;
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        return new CachingConnectionFactory();

    }

    @Bean
    JmsListenerContainerFactory<?> jmsListenerContainerFactory(ConnectionFactory connectionFactory, ConditionalRejectingErrorHandler errorHandler, MessageConverter jsonMessageConverter,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(jsonMessageConverter);
        factory.setErrorHandler(errorHandler);
        return factory;
    }

}

Priyanshu Shukla

@priyanshus1

@artembilan is QpidJms sufficient to work with RabbitMQ AMQP 1.0 ?

Artem Bilan

@artembilan

Have no idea, but looks like you are missing some bits in your configuration. Why is that plain new CachingConnectionFactory() without any target factory? Isn't that java.lang.IllegalArgumentException: Target Connection or ConnectionFactory is required enough for you to determine that something is missed?

Priyanshu Shukla

@priyanshus1

@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

I understand this may not be correct channel to ask this question but I am just seeking help from channel I know would reply :D

Artem Bilan

@artembilan

@priyanshus1 , I think you need to consult with RabbitMQ team about the best Java client for their AMQP 1.0 support: https://groups.google.com/forum/#!forum/rabbitmq-users

Priyanshu Shukla

@priyanshus1

@artembilan good point

thanks

:+1:

Gary Russell

@garyrussell

https://github.com/rabbitmq/rabbitmq-jms-client

Artem Bilan

@artembilan

I thought that one is for regular RabbitMQ AMQP 0.9 protocol and not for AMQP 1.0 ...

Gary Russell

@garyrussell

Right - but it's supported by the RabbitMQ team, unlike QPID.

Artem Bilan

@artembilan

The question was to enable AMQP 1.0 plugin on RabbitMQ and find some JMS client which could work with that

I'm just not sure that RabbitMQ JMS Client is AMQP 1.0 compliant...

Gary Russell

@garyrussell

It's not; the question should be why you would want to use an unsupported software stack Vs. a supported one. i.e. what is the compelling reason to use 1.0 ?

Artem Bilan

@artembilan

Oh! This is not the question, but more like instruction for action. So, good point, Gary! :smile:

Yuval Bonen

@yuvalprtn

Hi have this code

@Configuration
public class RabbitTemplateConfiguration {

  RabbitTemplateConfiguration(RabbitTemplate rabbitTemplate){
    rabbitTemplate.setAfterReceivePostProcessors(new AuthenticationAwareMessagePostProcessor());
  }
}

But for some reason the AuthenticationAwareMessagePostProcessor is never called

If I use setBeforePublishPostProcessors it is being called, but I need the "filter" on the receiving side

Using SpringBoot 2.1.8

Gary Russell

@garyrussell

@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' ...

Yuval Bonen

@yuvalprtn

Thanks @garyrussell we are working with @RabbitListener

Gary Russell

@garyrussell

Then you need to inject it into the listener container factory, not the template.

The template is not used by @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;
        });
    }

}

Yuval Bonen

@yuvalprtn

@garyrussell :heart: :wink:
Thanks, will give it a shot!

Yuval Bonen

@yuvalprtn

Works like a charm, thanks

Where communities thrive

People

Repo info

Activity