Move some of RabbitAmqpTemplate method up to AsyncAmqpTemplate

* Improve Client module tests

Author: artembilan

spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java

@@ -27,11 +27,79 @@
  * receive operations using {@link CompletableFuture}s.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  *
  */
 public interface AsyncAmqpTemplate {
 
+	default CompletableFuture<Boolean> send(Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> send(String queue, Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> send(String exchange, @Nullable String routingKey, Message message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(Object message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(String queue, Object message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(String queue, Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message,
+			@Nullable MessagePostProcessor messagePostProcessor) {
+
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Message> receive() {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Message> receive(String queueName) {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Object> receiveAndConvert() {
+		throw new UnsupportedOperationException();
+	}
+
+	default CompletableFuture<Object> receiveAndConvert(String queueName) {
+		throw new UnsupportedOperationException();
+	}
+
+	default <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T> type) {
+		throw new UnsupportedOperationException();
+	}
+
+	default <T> CompletableFuture<T> receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) {
+		throw new UnsupportedOperationException();
+	}
+
 	/**
 	 * Send a message to the default exchange with the default routing key. If the message
 	 * contains a correlationId property, it must be unique.

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplate.java

@@ -156,7 +156,10 @@ public void destroy() {
 	 * @param message to publish
 	 * @return the {@link CompletableFuture} as an async result of the message publication.
 	 */
+	@Override
 	public CompletableFuture<Boolean> 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");
 		return doSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message);
 	}
 
@@ -166,10 +169,12 @@ public CompletableFuture<Boolean> send(Message message) {
 	 * @param message to publish
 	 * @return the {@link CompletableFuture} as an async result of the message publication.
 	 */
+	@Override
 	public CompletableFuture<Boolean> send(String queue, Message message) {
 		return doSend(null, null, queue, message);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> send(String exchange, @Nullable String routingKey, Message message) {
 		return doSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message);
 	}
@@ -226,30 +231,38 @@ private CompletableFuture<Boolean> doSend(@Nullable String exchange, @Nullable S
 	 * @param message to publish
 	 * @return the {@link CompletableFuture} as an async result of the message publication.
 	 */
+	@Override
 	public CompletableFuture<Boolean> 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");
 		return doConvertAndSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message, null);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> convertAndSend(String queue, Object message) {
 		return doConvertAndSend(null, null, queue, message, null);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message) {
 		return doConvertAndSend(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message, null);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> convertAndSend(Object message,
 			@Nullable MessagePostProcessor messagePostProcessor) {
 
 		return doConvertAndSend(null, null, null, message, messagePostProcessor);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> convertAndSend(String queue, Object message,
 			@Nullable MessagePostProcessor messagePostProcessor) {
 
 		return doConvertAndSend(null, null, queue, message, messagePostProcessor);
 	}
 
+	@Override
 	public CompletableFuture<Boolean> convertAndSend(String exchange, @Nullable String routingKey, Object message,
 			@Nullable MessagePostProcessor messagePostProcessor) {
 
@@ -269,11 +282,13 @@ private CompletableFuture<Boolean> doConvertAndSend(@Nullable String exchange, @
 		return doSend(exchange, routingKey, queue, message);
 	}
 
+	@Override
 	public CompletableFuture<Message> receive() {
 		return receive(getRequiredQueue());
 	}
 
 	@SuppressWarnings("try")
+	@Override
 	public CompletableFuture<Message> receive(String queueName) {
 		CompletableFuture<Message> messageFuture = new CompletableFuture<>();
 
@@ -293,10 +308,12 @@ public CompletableFuture<Message> receive(String queueName) {
 				.whenComplete((message, exception) -> consumer.close());
 	}
 
+	@Override
 	public CompletableFuture<Object> receiveAndConvert() {
 		return receiveAndConvert(getRequiredQueue());
 	}
 
+	@Override
 	public CompletableFuture<Object> receiveAndConvert(String queueName) {
 		return receive(queueName)
 				.thenApply(this.messageConverter::fromMessage);
@@ -309,6 +326,7 @@ public CompletableFuture<Object> receiveAndConvert(String queueName) {
 	 * @param type the type to covert received result.
 	 * @return the CompletableFuture with a result.
 	 */
+	@Override
 	public <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T> type) {
 		return receiveAndConvert(getRequiredQueue(), type);
 	}
@@ -322,6 +340,7 @@ public <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T>
 	 * @return the CompletableFuture with a result.
 	 */
 	@SuppressWarnings("unchecked")
+	@Override
 	public <T> CompletableFuture<T> receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) {
 		SmartMessageConverter smartMessageConverter = getRequiredSmartMessageConverter();
 		return receive(queueName)

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpAdminTests.java

@@ -43,12 +43,6 @@
 @ContextConfiguration
 public class RabbitAmqpAdminTests extends RabbitAmqpTestBase {
 
-	@Autowired
-	RabbitAmqpTemplate template;
-
-	@Autowired
-	RabbitAmqpAdmin admin;
-
 	@Autowired
 	@Qualifier("ds")
 	Declarables declarables;

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java

@@ -26,13 +26,15 @@
 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.Queue;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.test.context.ContextConfiguration;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 
 /**
  * @author Artem Bilan
@@ -58,6 +60,17 @@ void tearDown() {
 		this.rabbitAmqpTemplate.destroy();
 	}
 
+	@Test
+	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");
+
+		assertThatIllegalStateException()
+				.isThrownBy(() -> this.template.convertAndSend(new byte[0]))
+				.withMessage("For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided");
+	}
+
 	@Test
 	void defaultExchangeAndRoutingKey() {
 		this.rabbitAmqpTemplate.setExchange("e1");

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java

@@ -21,6 +21,7 @@
 import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;
 
 import org.springframework.amqp.rabbit.junit.AbstractTestContainerTests;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.test.annotation.DirtiesContext;
@@ -37,6 +38,18 @@
 @DirtiesContext
 abstract class RabbitAmqpTestBase extends AbstractTestContainerTests {
 
+	@Autowired
+	protected Environment environment;
+
+	@Autowired
+	protected Connection connection;
+
+	@Autowired
+	protected RabbitAmqpAdmin admin;
+
+	@Autowired
+	protected RabbitAmqpTemplate template;
+
 	@Configuration
 	public static class AmqpCommonConfig {