diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 31603093..cbe389e0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -198,6 +198,9 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count private final AtomicInteger containerInstance = new AtomicInteger(); + private final Map consumerArgs = new HashMap<>(); + + private String exchange = DEFAULT_EXCHANGE; private String routingKey = DEFAULT_ROUTING_KEY; @@ -936,6 +939,33 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count .sum(); } + /** + * When using receive methods with a non-zero timeout, a + * {@link com.rabbitmq.client.Consumer} is created to receive the message. Use this + * property to add arguments to the consumer (e.g. {@code x-priority}). + * @param arg the argument name to pass into the {@code basicConsume} operation. + * @param value the argument value to pass into the {@code basicConsume} operation. + * @since 2.4.8 + * @see #removeConsumerArg(String) + */ + public void addConsumerArg(String arg, Object value) { + this.consumerArgs.put(arg, value); + } + + /** + * When using receive methods with a non-zero timeout, a + * {@link com.rabbitmq.client.Consumer} is created to receive the message. Use this + * method to remove an argument from those passed into the {@code basicConsume} + * operation. + * @param arg the argument name. + * @return the previous value. + * @since 2.4.8 + * @see #addConsumerArg(String, Object) + */ + public Object removeConsumerArg(String arg) { + return this.consumerArgs.remove(arg); + } + @Override public void start() { doStart(); @@ -2743,7 +2773,7 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count } }; - channel.basicConsume(queueName, consumer); + channel.basicConsume(queueName, false, this.consumerArgs, consumer); if (!latch.await(timeoutMillis, TimeUnit.MILLISECONDS)) { if (channel instanceof ChannelProxy) { ((ChannelProxy) channel).getTargetChannel().close(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 6dd054be..647e931b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -335,8 +335,10 @@ public class RabbitTemplateIntegrationTests { } @Override - public String basicConsume(String queue, Consumer callback) throws IOException { - return super.basicConsume(queue, new MockConsumer(callback)); + public String basicConsume(String queue, boolean autoAck, Map args, Consumer callback) + throws IOException { + + return super.basicConsume(queue, autoAck, args, new MockConsumer(callback)); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java index 0c36a162..ace861f5 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java @@ -25,6 +25,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; @@ -49,6 +50,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.springframework.amqp.AmqpAuthenticationException; @@ -658,6 +660,37 @@ public class RabbitTemplateTests { ConnectionFactoryUtils.enableAfterCompletionFailureCapture(false); } + @Test + void consumerArgs() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection); + given(mockConnection.isOpen()).willReturn(true); + given(mockConnection.createChannel()).willReturn(mockChannel); + willAnswer(inv -> { + Consumer consumer = inv.getArgument(3); + consumer.handleConsumeOk("tag"); + return null; + }).given(mockChannel).basicConsume(any(), anyBoolean(), anyMap(), any()); + + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); + connectionFactory.setExecutor(mock(ExecutorService.class)); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + assertThat(template.receive("foo", 1)).isNull(); + @SuppressWarnings("unchecked") + ArgumentCaptor> argsCaptor = ArgumentCaptor.forClass(Map.class); + verify(mockChannel).basicConsume(eq("foo"), eq(false), argsCaptor.capture(), any()); + assertThat(argsCaptor.getValue()).isEmpty(); + template.addConsumerArg("x-priority", 10); + assertThat(template.receive("foo", 1)).isNull(); + assertThat(argsCaptor.getValue()).containsEntry("x-priority", 10); + assertThat(template.removeConsumerArg("x-priority")).isEqualTo(10); + assertThat(template.receive("foo", 1)).isNull(); + assertThat(argsCaptor.getValue()).isEmpty(); + } + @SuppressWarnings("serial") private class TestTransactionManager extends AbstractPlatformTransactionManager { diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index d2f495e7..90062b4d 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1885,11 +1885,14 @@ By default, if no message is available, `null` is returned immediately. There is no blocking. Starting with version 1.5, you can set a `receiveTimeout`, in milliseconds, and the receive methods block for up to that long, waiting for a message. A value less than zero means block indefinitely (or at least until the connection to the broker is lost). -Version 1.6 introduced variants of the `receive` methods that let the timeout be passed in on each call. +Version 1.6 introduced variants of the `receive` methods that allows the timeout be passed in on each call. CAUTION: Since the receive operation creates a new `QueueingConsumer` for each message, this technique is not really appropriate for high-volume environments. Consider using an asynchronous consumer or a `receiveTimeout` of zero for those use cases. +Starting with version 2.4.8, when using a non-zero timeout, you can specify arguments passed into the `basicConsume` method used to associate the consumer with the channel. +For example: `template.addConsumerArg("x-priority", 10)`. + There are four simple `receive` methods available. As with the `Exchange` on the sending side, there is a method that requires that a default queue property has been set directly on the template itself, and there is a method that accepts a queue parameter at runtime.