diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java index 5f7556fb..90dfe728 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java @@ -92,11 +92,30 @@ public interface AsyncAmqpTemplate { throw new UnsupportedOperationException(); } - default CompletableFuture receiveAndConvert(ParameterizedTypeReference type) { + default CompletableFuture receiveAndConvert(@Nullable ParameterizedTypeReference type) { throw new UnsupportedOperationException(); } - default CompletableFuture receiveAndConvert(String queueName, ParameterizedTypeReference type) { + default CompletableFuture receiveAndConvert(String queueName, @Nullable ParameterizedTypeReference type) { + throw new UnsupportedOperationException(); + } + + default CompletableFuture receiveAndReply(ReceiveAndReplyCallback callback) { + throw new UnsupportedOperationException(); + } + + /** + * Perform a server-side RPC functionality. + * The request message must have a {@code replyTo} property. + * The request {@code messageId} property is used for correlation. + * The callback might not produce a reply with the meaning nothing to answer. + * @param queueName the queue to consume request. + * @param callback an application callback to handle request and produce reply. + * @return the completion status: true if no errors and reply has been produced. + * @param the request body type. + * @param the response body type + */ + default CompletableFuture receiveAndReply(String queueName, ReceiveAndReplyCallback callback) { throw new UnsupportedOperationException(); } @@ -240,8 +259,9 @@ public interface AsyncAmqpTemplate { * @param the expected result type. * @return the {@link CompletableFuture}. */ - CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, - ParameterizedTypeReference responseType); + CompletableFuture convertSendAndReceiveAsType(Object object, + @Nullable MessagePostProcessor messagePostProcessor, + @Nullable ParameterizedTypeReference responseType); /** * Convert the object to a message and send it to the default exchange with the diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java index 0972747d..cb3e8a4c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java @@ -464,7 +464,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa @Override public RabbitConverterFuture convertSendAndReceiveAsType(Object object, - MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { + @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object, messagePostProcessor, responseType); diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java index 8df58e39..559f744b 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java @@ -17,25 +17,28 @@ package org.springframework.amqp.rabbitmq.client; import java.time.Duration; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; import com.rabbitmq.client.amqp.Consumer; import com.rabbitmq.client.amqp.Environment; import com.rabbitmq.client.amqp.Publisher; import com.rabbitmq.client.amqp.Resource; +import com.rabbitmq.client.amqp.RpcClient; import org.jspecify.annotations.Nullable; -import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpIllegalStateException; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.AsyncAmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.ReceiveAndReplyCallback; -import org.springframework.amqp.core.ReplyToAddressCallback; +import org.springframework.amqp.core.ReceiveAndReplyMessageCallback; import org.springframework.amqp.rabbit.core.AmqpNackReceivedException; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; @@ -43,7 +46,9 @@ import org.springframework.amqp.support.converter.SmartMessageConverter; import org.springframework.amqp.utils.JavaUtils; import org.springframework.beans.factory.DisposableBean; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.log.LogAccessor; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * The {@link AmqpTemplate} for RabbitMQ AMQP 1.0 protocol support. @@ -55,6 +60,8 @@ import org.springframework.util.Assert; */ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { + private static final LogAccessor LOG = new LogAccessor(RabbitAmqpAdmin.class); + private final AmqpConnectionFactory connectionFactory; private final Lock instanceLock = new ReentrantLock(); @@ -71,10 +78,14 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { private @Nullable String defaultReceiveQueue; + private @Nullable String defaultReplyToQueue; + private Resource.StateListener @Nullable [] stateListeners; private Duration publishTimeout = Duration.ofSeconds(60); + private Duration completionTimeout = Duration.ofSeconds(60); + public RabbitAmqpTemplate(AmqpConnectionFactory connectionFactory) { this.connectionFactory = connectionFactory; } @@ -87,6 +98,19 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { this.publishTimeout = timeout; } + /** + * Set a duration for {@link CompletableFuture#orTimeout(long, TimeUnit)} on returns. + * There is no {@link CompletableFuture} API like {@code onTimeout()} requested + * from the {@link CompletableFuture#get(long, TimeUnit)}, + * but used in operations AMQP resources have to be closed eventually independently + * of the {@link CompletableFuture} fulfilment. + * Defaults to 1 minute. + * @param completionTimeout duration for future completions. + */ + public void setCompletionTimeout(Duration completionTimeout) { + this.completionTimeout = completionTimeout; + } + /** * Set a default exchange for publishing. * Cannot be real default AMQP exchange. @@ -116,6 +140,22 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { this.defaultQueue = queue; } + /** + * The name of the default queue to receive messages from when none is specified explicitly. + * @param queue the default queue name to use for receive operation. + */ + public void setReceiveQueue(String queue) { + this.defaultReceiveQueue = queue; + } + + /** + * The name of the default queue to receive replies from when none is specified explicitly. + * @param queue the default queue name to use for send-n-receive operation. + */ + public void setReplyToQueue(String queue) { + this.defaultReplyToQueue = queue; + } + /** * Set a converter for {@link #convertAndSend(Object)} operations. * @param messageConverter the converter. @@ -124,14 +164,6 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { this.messageConverter = messageConverter; } - /** - * The name of the default queue to receive messages from when none is specified explicitly. - * @param queue the default queue name to use for receive operation. - */ - public void setDefaultReceiveQueue(String queue) { - this.defaultReceiveQueue = queue; - } - private String getRequiredQueue() throws IllegalStateException { String name = this.defaultReceiveQueue; Assert.state(name != null, "No 'queue' specified. Check configuration of this 'RabbitAmqpTemplate'."); @@ -178,7 +210,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { @Override public CompletableFuture send(Message message) { Assert.state(this.defaultExchange != null || this.defaultQueue != null, - "For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided"); + "For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); return doSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message); } @@ -201,16 +233,8 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { private CompletableFuture doSend(@Nullable String exchange, @Nullable String routingKey, @Nullable String queue, Message message) { - com.rabbitmq.client.amqp.Message amqpMessage = getPublisher().message(); - com.rabbitmq.client.amqp.Message.MessageAddressBuilder address = amqpMessage.toAddress(); - JavaUtils.INSTANCE - .acceptIfNotNull(exchange, address::exchange) - .acceptIfNotNull(routingKey, address::key) - .acceptIfNotNull(queue, address::queue); - - amqpMessage = address.message(); - - RabbitAmqpUtils.toAmqpMessage(message, amqpMessage); + com.rabbitmq.client.amqp.Message amqpMessage = + toAmqpMessage(exchange, routingKey, queue, message, getPublisher()::message); CompletableFuture publishResult = new CompletableFuture<>(); @@ -235,7 +259,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { @Override public CompletableFuture convertAndSend(Object message) { Assert.state(this.defaultExchange != null || this.defaultQueue != null, - "For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided"); + "For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); return doConvertAndSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message, null); } @@ -273,10 +297,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { private CompletableFuture doConvertAndSend(@Nullable String exchange, @Nullable String routingKey, @Nullable String queue, Object data, @Nullable MessagePostProcessor messagePostProcessor) { - Message message = - data instanceof Message - ? (Message) data - : this.messageConverter.toMessage(data, new MessageProperties()); + Message message = convertToMessageIfNecessary(data); if (messagePostProcessor != null) { message = messagePostProcessor.postProcessMessage(message); } @@ -288,6 +309,13 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { return receive(getRequiredQueue()); } + /** + * Request a head message from the provided queue. + * A returned {@link CompletableFuture} timeouts after {@link #setCompletionTimeout(Duration)}. + * @param queueName the queue to consume message from. + * @return the future with a received message. + * @see #setCompletionTimeout(Duration) + */ @SuppressWarnings("try") @Override public CompletableFuture receive(String queueName) { @@ -306,47 +334,54 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { .build(); return messageFuture - .orTimeout(1, TimeUnit.MINUTES) + .orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS) .whenComplete((message, exception) -> consumer.close()); } @Override public CompletableFuture receiveAndConvert() { - return receiveAndConvert(getRequiredQueue()); + return receiveAndConvert((ParameterizedTypeReference) null); } @Override public CompletableFuture receiveAndConvert(String queueName) { - return receive(queueName) - .thenApply(this.messageConverter::fromMessage); + return receiveAndConvert(queueName, null); } /** - * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body + * Receive a message from {@link #setReceiveQueue(String)} and convert its body * to the expected type. * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}. * @param type the type to covert received result. * @return the CompletableFuture with a result. */ @Override - public CompletableFuture receiveAndConvert(ParameterizedTypeReference type) { + public CompletableFuture receiveAndConvert(@Nullable ParameterizedTypeReference type) { return receiveAndConvert(getRequiredQueue(), type); } /** - * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body + * Receive a message from {@link #setReceiveQueue(String)} and convert its body * to the expected type. * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}. * @param queueName the queue to consume message from. * @param type the type to covert received result. * @return the CompletableFuture with a result. */ - @SuppressWarnings("unchecked") @Override - public CompletableFuture receiveAndConvert(String queueName, ParameterizedTypeReference type) { - SmartMessageConverter smartMessageConverter = getRequiredSmartMessageConverter(); + public CompletableFuture receiveAndConvert(String queueName, @Nullable ParameterizedTypeReference type) { return receive(queueName) - .thenApply((message) -> (T) smartMessageConverter.fromMessage(message, type)); + .thenApply((message) -> convertReply(message, type)); + } + + @SuppressWarnings("unchecked") + private T convertReply(Message message, @Nullable ParameterizedTypeReference type) { + if (type != null) { + return (T) getRequiredSmartMessageConverter().fromMessage(message, type); + } + else { + return (T) this.messageConverter.fromMessage(message); + } } private SmartMessageConverter getRequiredSmartMessageConverter() throws IllegalStateException { @@ -355,103 +390,280 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean { return (SmartMessageConverter) this.messageConverter; } - public boolean receiveAndReply(ReceiveAndReplyCallback callback) throws AmqpException { - throw new UnsupportedOperationException(); + @Override + public CompletableFuture receiveAndReply(ReceiveAndReplyCallback callback) { + return receiveAndReply(getRequiredQueue(), callback); } - public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback) throws AmqpException { - throw new UnsupportedOperationException(); + @Override + @SuppressWarnings("try") + public CompletableFuture receiveAndReply(String queueName, ReceiveAndReplyCallback callback) { + CompletableFuture rpcFuture = new CompletableFuture<>(); + + Consumer.MessageHandler consumerHandler = + (context, message) -> { + Message requestMessage = RabbitAmqpUtils.fromAmqpMessage(message, null); + try { + Object messageId = message.messageId(); + Assert.notNull(messageId, + "The 'message-id' property has to be set on request. Used for reply correlation."); + String replyTo = message.replyTo(); + Assert.hasText(replyTo, + "The 'reply-to' property has to be set on request. Used for reply publishing."); + Message reply = handleRequestAndProduceReply(requestMessage, callback); + if (reply == null) { + LOG.info(() -> "No reply for request: " + requestMessage); + context.accept(); + rpcFuture.complete(false); + } + else { + com.rabbitmq.client.amqp.Message replyMessage = getPublisher().message(); + RabbitAmqpUtils.toAmqpMessage(reply, replyMessage); + replyMessage.correlationId(messageId); + replyMessage.to(replyTo); + getPublisher().publish(replyMessage, (ctx) -> { + }); + context.accept(); + rpcFuture.complete(true); + } + } + catch (Exception ex) { + context.discard(); + rpcFuture.completeExceptionally( + new AmqpIllegalStateException("Failed to process RPC request: " + requestMessage, ex)); + } + }; + + Consumer consumer = + this.connectionFactory.getConnection() + .consumerBuilder() + .queue(queueName) + .initialCredits(1) + .priority(10) + .messageHandler(consumerHandler) + .build(); + + return rpcFuture + .orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS) + .whenComplete((message, exception) -> consumer.close()); } - public boolean receiveAndReply(ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException { - throw new UnsupportedOperationException(); + @SuppressWarnings("unchecked") + private @Nullable Message handleRequestAndProduceReply(Message requestMessage, + ReceiveAndReplyCallback callback) { + + Object receive = requestMessage; + if (!(ReceiveAndReplyMessageCallback.class.isAssignableFrom(callback.getClass()))) { + receive = this.messageConverter.fromMessage(requestMessage); + } + + S reply; + try { + reply = callback.handle((R) receive); + } + catch (ClassCastException ex) { + StackTraceElement[] trace = ex.getStackTrace(); + if (trace[0].getMethodName().equals("handle") + && Objects.equals(trace[1].getFileName(), "RabbitAmqpTemplate.java")) { + + throw new IllegalArgumentException("ReceiveAndReplyCallback '" + callback + + "' can't handle received object '" + receive + "'", ex); + } + else { + throw ex; + } + } + + if (reply != null) { + return convertToMessageIfNecessary(reply); + } + return null; } - public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException { - throw new UnsupportedOperationException(); - } - - public boolean receiveAndReply(ReceiveAndReplyCallback callback, ReplyToAddressCallback replyToAddressCallback) throws AmqpException { - throw new UnsupportedOperationException(); - } - - public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback, ReplyToAddressCallback replyToAddressCallback) throws AmqpException { - throw new UnsupportedOperationException(); + private Message convertToMessageIfNecessary(Object data) { + if (data instanceof Message msg) { + return msg; + } + else { + return this.messageConverter.toMessage(data, new MessageProperties()); + } } @Override public CompletableFuture sendAndReceive(Message message) { - throw new UnsupportedOperationException(); + Assert.state(this.defaultExchange != null || this.defaultQueue != null, + "For send-n-receive with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); + return doSendAndReceive(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message); } @Override - public CompletableFuture sendAndReceive(String routingKey, Message message) { - throw new UnsupportedOperationException(); + public CompletableFuture sendAndReceive(String exchange, @Nullable String routingKey, Message message) { + return doSendAndReceive(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message); } @Override - public CompletableFuture sendAndReceive(String exchange, String routingKey, Message message) { - throw new UnsupportedOperationException(); + public CompletableFuture sendAndReceive(String queue, Message message) { + return doSendAndReceive(null, null, queue, message); + } + + @SuppressWarnings("try") + private CompletableFuture doSendAndReceive(@Nullable String exchange, @Nullable String routingKey, + @Nullable String queue, Message message) { + + MessageProperties messageProperties = message.getMessageProperties(); + String messageId = messageProperties.getMessageId(); + String correlationId = messageProperties.getCorrelationId(); + String replyTo = messageProperties.getReplyTo(); + + // HTTP over AMQP 1.0 extension specification, 5.1: + // To associate a response with a request, the correlation-id value of the response properties + // MUST be set to the message-id value of the request properties. + // So, this supplier will override request message-id, respectively. + // Otherwise, the RpcClient generates correlation-id internally. + Supplier correlationIdSupplier = null; + if (StringUtils.hasText(correlationId)) { + correlationIdSupplier = () -> correlationId; + } + else if (StringUtils.hasText(messageId)) { + correlationIdSupplier = () -> messageId; + } + + // The default reply-to queue, or the one supplied in the message. + // Otherwise, the RpcClient generates one as exclusive and auto-delete. + String replyToQueue = this.defaultReplyToQueue; + if (StringUtils.hasText(replyTo)) { + replyToQueue = replyTo; + } + + RpcClient rpcClient = + this.connectionFactory.getConnection() + .rpcClientBuilder() + .requestTimeout(this.publishTimeout) + .correlationIdSupplier(correlationIdSupplier) + .replyToQueue(replyToQueue) + .build(); + + com.rabbitmq.client.amqp.Message amqpMessage = + toAmqpMessage(exchange, routingKey, queue, message, rpcClient::message); + + return rpcClient.publish(amqpMessage) + .thenApply((reply) -> RabbitAmqpUtils.fromAmqpMessage(reply, null)) + .orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS) + .whenComplete((replyMessage, exception) -> rpcClient.close()); } @Override public CompletableFuture convertSendAndReceive(Object object) { - throw new UnsupportedOperationException(); + return convertSendAndReceiveAsType(object, null, null); } @Override - public CompletableFuture convertSendAndReceive(String routingKey, Object object) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceive(String queue, Object object) { + return convertSendAndReceiveAsType(queue, object, null, null); } @Override - public CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceive(String exchange, @Nullable String routingKey, Object object) { + return convertSendAndReceiveAsType(exchange, routingKey, object, null, null); } @Override public CompletableFuture convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor) { - throw new UnsupportedOperationException(); + return convertSendAndReceiveAsType(object, messagePostProcessor, null); } @Override - public CompletableFuture convertSendAndReceive(String routingKey, Object object, MessagePostProcessor messagePostProcessor) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceive(String queue, Object object, + MessagePostProcessor messagePostProcessor) { + + return convertSendAndReceiveAsType(queue, object, messagePostProcessor, null); } @Override - public CompletableFuture convertSendAndReceive(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceive(String exchange, @Nullable String routingKey, + Object object, @Nullable MessagePostProcessor messagePostProcessor) { + + return convertSendAndReceiveAsType(exchange, routingKey, object, messagePostProcessor, null); } @Override - public CompletableFuture convertSendAndReceiveAsType(Object object, ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(Object object, + ParameterizedTypeReference responseType) { + + return convertSendAndReceiveAsType(object, null, responseType); } @Override - public CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(String queue, Object object, + ParameterizedTypeReference responseType) { + + return convertSendAndReceiveAsType(queue, object, null, responseType); } @Override - public CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(String exchange, @Nullable String routingKey, + Object object, ParameterizedTypeReference responseType) { + + return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType); } @Override - public CompletableFuture convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(Object object, + @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { + + Assert.state(this.defaultExchange != null || this.defaultQueue != null, + "For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); + + return doConvertSendAndReceive(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, object, + messagePostProcessor, responseType); } @Override - public CompletableFuture convertSendAndReceiveAsType(String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(String queue, Object object, + @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { + + return doConvertSendAndReceive(null, null, queue, object, messagePostProcessor, responseType); } @Override - public CompletableFuture convertSendAndReceiveAsType(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference responseType) { - throw new UnsupportedOperationException(); + public CompletableFuture convertSendAndReceiveAsType(String exchange, @Nullable String routingKey, + Object object, @Nullable MessagePostProcessor messagePostProcessor, + @Nullable ParameterizedTypeReference responseType) { + + return doConvertSendAndReceive(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, + object, messagePostProcessor, responseType); + } + + private CompletableFuture doConvertSendAndReceive(@Nullable String exchange, @Nullable String routingKey, + @Nullable String queue, Object data, @Nullable MessagePostProcessor messagePostProcessor, + @Nullable ParameterizedTypeReference responseType) { + + Message message = convertToMessageIfNecessary(data); + if (messagePostProcessor != null) { + message = messagePostProcessor.postProcessMessage(message); + } + return doSendAndReceive(exchange, routingKey, queue, message) + .thenApply((reply) -> convertReply(reply, responseType)); + } + + private static com.rabbitmq.client.amqp.Message toAmqpMessage(@Nullable String exchange, + @Nullable String routingKey, @Nullable String queue, Message message, + Supplier amqpMessageSupplier) { + + com.rabbitmq.client.amqp.Message.MessageAddressBuilder address = + amqpMessageSupplier.get() + .toAddress(); + + JavaUtils.INSTANCE + .acceptIfNotNull(exchange, address::exchange) + .acceptIfNotNull(routingKey, address::key) + .acceptIfNotNull(queue, address::queue); + + com.rabbitmq.client.amqp.Message amqpMessage = address.message(); + + RabbitAmqpUtils.toAmqpMessage(message, amqpMessage); + + return amqpMessage; } } diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java index 340b82d9..44e1df59 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpUtils.java @@ -53,7 +53,8 @@ public final class RabbitAmqpUtils { .acceptIfNotNull(amqpMessage.contentEncoding(), messageProperties::setContentEncoding) .acceptIfNotNull(amqpMessage.absoluteExpiryTime(), (exp) -> messageProperties.setExpiration(Long.toString(exp))) - .acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time))); + .acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time))) + .acceptIfNotNull(amqpMessage.replyTo(), messageProperties::setReplyTo); amqpMessage.forEachProperty(messageProperties::setHeader); diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java index afd021cd..7cb211c6 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java @@ -16,20 +16,31 @@ package org.springframework.amqp.rabbitmq.client; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.amqp.AmqpIllegalStateException; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageBuilder; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.test.context.ContextConfiguration; +import org.springframework.util.MimeTypeUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; @@ -58,11 +69,13 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase { void illegalStateOnNoDefaults() { assertThatIllegalStateException() .isThrownBy(() -> this.template.send(new Message(new byte[0]))) - .withMessage("For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided"); + .withMessage( + "For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); assertThatIllegalStateException() .isThrownBy(() -> this.template.convertAndSend(new byte[0])) - .withMessage("For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided"); + .withMessage( + "For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided"); } @Test @@ -81,7 +94,7 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase { @Test void defaultQueues() { this.rabbitAmqpTemplate.setQueue("q1"); - this.rabbitAmqpTemplate.setDefaultReceiveQueue("q1"); + this.rabbitAmqpTemplate.setReceiveQueue("q1"); assertThat(this.rabbitAmqpTemplate.convertAndSend("test2")) .succeedsWithin(Duration.ofSeconds(10)); @@ -91,6 +104,65 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase { .isEqualTo("test2"); } + @Test + void verifyRpc() { + String testRequest = "rpc-request"; + String testReply = "rpc-reply"; + + CompletableFuture rpcClientResult = this.template.convertSendAndReceive("e1", "k1", testRequest); + + AtomicReference receivedRequest = new AtomicReference<>(); + CompletableFuture rpcServerResult = + this.rabbitAmqpTemplate.receiveAndReply("q1", + payload -> { + receivedRequest.set(payload); + return testReply; + }); + + assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(true); + assertThat(rpcClientResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(testReply); + assertThat(receivedRequest.get()).isEqualTo(testRequest); + + this.template.send("q1", + MessageBuilder.withBody("non-rpc-request".getBytes(StandardCharsets.UTF_8)) + .setMessageId(UUID.randomUUID().toString()) + .setContentType(MimeTypeUtils.TEXT_PLAIN_VALUE) + .build()); + + rpcServerResult = this.rabbitAmqpTemplate.receiveAndReply("q1", payload -> "reply-attempt"); + + assertThat(rpcServerResult).failsWithin(Duration.ofSeconds(10)) + .withThrowableOfType(ExecutionException.class) + .withCauseInstanceOf(AmqpIllegalStateException.class) + .withRootCauseInstanceOf(IllegalArgumentException.class) + .withMessageContaining("Failed to process RPC request: (Body:'non-rpc-request'") + .withStackTraceContaining("The 'reply-to' property has to be set on request. Used for reply publishing."); + + rpcClientResult = this.template.convertSendAndReceive("q1", testRequest); + rpcServerResult = this.rabbitAmqpTemplate.receiveAndReply("q1", payload -> null); + + assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(false); + assertThat(rpcClientResult).failsWithin(Duration.ofSeconds(2)) + .withThrowableThat() + .isInstanceOf(TimeoutException.class); + + this.template.convertSendAndReceive("q1", new byte[0]); + + rpcServerResult = this.rabbitAmqpTemplate.receiveAndReply("q1", payload -> payload); + assertThat(rpcServerResult).failsWithin(Duration.ofSeconds(10)) + .withThrowableOfType(ExecutionException.class) + .withCauseInstanceOf(AmqpIllegalStateException.class) + .withRootCauseInstanceOf(ClassCastException.class) + .withMessageContaining("Failed to process RPC request: (Body:'[B") + .withStackTraceContaining("class [B cannot be cast to class java.lang.String"); + + assertThat(this.template.receiveAndConvert("dlq1")).succeedsWithin(10, TimeUnit.SECONDS) + .isEqualTo("non-rpc-request"); + + assertThat(this.template.receiveAndConvert("dlq1")).succeedsWithin(10, TimeUnit.SECONDS) + .isEqualTo(new byte[0]); + } + @Configuration static class Config { @@ -101,7 +173,7 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase { @Bean Queue q1() { - return new Queue("q1"); + return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build(); } @Bean diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java index 5d580e84..d4ccbe04 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java @@ -23,10 +23,13 @@ import java.util.stream.Stream; import com.rabbitmq.client.amqp.Environment; import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Declarable; import org.springframework.amqp.core.Declarables; import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.junit.AbstractTestContainerTests; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.Lifecycle; @@ -94,6 +97,21 @@ public abstract class RabbitAmqpTestBase extends AbstractTestContainerTests { return new RabbitAmqpTemplate(connectionFactory); } + @Bean + TopicExchange dlx1() { + return new TopicExchange("dlx1"); + } + + @Bean + Queue dlq1() { + return new Queue("dlq1"); + } + + @Bean + Binding dlq1Binding() { + return BindingBuilder.bind(dlq1()).to(dlx1()).with("#"); + } + volatile boolean running; @Override diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java index 9a375034..bbada753 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java @@ -30,11 +30,8 @@ import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; import org.springframework.amqp.core.AmqpAcknowledgment; -import org.springframework.amqp.core.Binding; -import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; -import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.annotation.EnableRabbit; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor; @@ -149,21 +146,6 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { @EnableRabbit static class Config { - @Bean - TopicExchange dlx1() { - return new TopicExchange("dlx1"); - } - - @Bean - Queue dlq1() { - return new Queue("dlq1"); - } - - @Bean - Binding dlq1Binding() { - return BindingBuilder.bind(dlq1()).to(dlx1()).with("#"); - } - @Bean Queue q1() { return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build(); diff --git a/spring-rabbitmq-client/src/test/resources/log4j2-test.xml b/spring-rabbitmq-client/src/test/resources/log4j2-test.xml index a4931c98..752714e3 100644 --- a/spring-rabbitmq-client/src/test/resources/log4j2-test.xml +++ b/spring-rabbitmq-client/src/test/resources/log4j2-test.xml @@ -7,6 +7,7 @@ +