Fix AMQP CF publisher confirms type usage

Related to spring-projects/spring-amqp#1067

Author: artembilan

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java

@@ -47,6 +47,8 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.2
  *
  */
@@ -65,15 +67,16 @@ void testConfirmOk(@Autowired IntegrationFlow flow, @Autowired RabbitTemplate te
 	void testWithReturn(@Autowired IntegrationFlow flow) {
 		assertThatThrownBy(() -> flow.getInputChannel()
 				.send(new GenericMessage<>("test", Collections.singletonMap("rk", "junkjunk"))))
-						.isInstanceOf(MessageHandlingException.class)
-						.hasCauseInstanceOf(AmqpException.class)
-						.extracting(ex -> ex.getCause())
-						.extracting(ex -> ex.getMessage())
-						.isEqualTo("Message was returned by the broker");
+				.isInstanceOf(MessageHandlingException.class)
+				.hasCauseInstanceOf(AmqpException.class)
+				.extracting(Throwable::getCause)
+				.extracting(Throwable::getMessage)
+				.isEqualTo("Message was returned by the broker");
 	}
 
 	@Test
-	@DisabledIf("#{systemEnvironment['TRAVIS'] ?: false}") // needs RabbitMQ 3.7
+	@DisabledIf("#{systemEnvironment['TRAVIS'] ?: false}")
+		// needs RabbitMQ 3.7
 	void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admin,
 			@Autowired RabbitTemplate template) {
 
@@ -82,10 +85,10 @@ void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admi
 		flow.getInputChannel().send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName())));
 		assertThatThrownBy(() -> flow.getInputChannel()
 				.send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName()))))
-						.hasCauseInstanceOf(AmqpException.class)
-						.extracting(ex -> ex.getCause())
-						.extracting(ex -> ex.getMessage())
-						.matches(msg -> msg.matches("Negative publisher confirm received: .*"));
+				.hasCauseInstanceOf(AmqpException.class)
+				.extracting(Throwable::getCause)
+				.extracting(Throwable::getMessage)
+				.matches(msg -> msg.matches("Negative publisher confirm received: .*"));
 		assertThat(template.receive(queue.getName())).isNotNull();
 		admin.deleteQueue(queue.getName());
 	}
@@ -97,7 +100,7 @@ public static class Config {
 		@Bean
 		public IntegrationFlow flow(RabbitTemplate template) {
 			return f -> f.handle(Amqp.outboundAdapter(template)
-							.exchangeName("")
+					.exchangeName("")
 					.routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))
 					.confirmCorrelationFunction(msg -> msg)
 					.waitForConfirm(true));
@@ -107,7 +110,7 @@ public IntegrationFlow flow(RabbitTemplate template) {
 		public CachingConnectionFactory cf() {
 			CachingConnectionFactory ccf = new CachingConnectionFactory(
 					RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
-			ccf.setPublisherConfirms(true);
+			ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
 			ccf.setPublisherReturns(true);
 			return ccf;
 		}

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java

@@ -31,17 +31,14 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
-import org.junit.AfterClass;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.core.AmqpReplyTimeoutException;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.amqp.rabbit.junit.BrokerRunning;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
@@ -54,7 +51,7 @@
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.support.MessageBuilder;
-import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
+import org.springframework.integration.test.condition.LogLevels;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ErrorMessage;
@@ -67,25 +64,14 @@
  * @since 4.3
  *
  */
-public class AsyncAmqpGatewayTests {
-
-	@ClassRule
-	public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("asyncQ1", "asyncRQ1");
-
-	@Rule
-	public Log4j2LevelAdjuster adjuster =
-			Log4j2LevelAdjuster.trace()
-					.categories(true, "org.springframework.amqp");
-
-	@AfterClass
-	public static void tearDown() {
-		brokerRunning.removeTestQueues();
-	}
+@RabbitAvailable(queues = { "asyncQ1", "asyncRQ1" })
+@LogLevels(categories = "org.springframework.amqp", level = "trace")
+class AsyncAmqpGatewayTests {
 
 	@Test
-	public void testConfirmsAndReturns() throws Exception {
+	void testConfirmsAndReturns() throws Exception {
 		CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
-		ccf.setPublisherConfirms(true);
+		ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
 		ccf.setPublisherReturns(true);
 		RabbitTemplate template = new RabbitTemplate(ccf);
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
@@ -213,7 +199,7 @@ public void testConfirmsAndReturns() throws Exception {
 				any(org.springframework.amqp.core.Message.class));
 		DirectFieldAccessor dfa = new DirectFieldAccessor(future);
 		dfa.setPropertyValue("nackCause", "nacknack");
-		SettableListenableFuture<Boolean> confirmFuture = new SettableListenableFuture<Boolean>();
+		SettableListenableFuture<Boolean> confirmFuture = new SettableListenableFuture<>();
 		confirmFuture.set(false);
 		dfa.setPropertyValue("confirm", confirmFuture);
 		new DirectFieldAccessor(gateway).setPropertyValue("template", asyncTemplate);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdviceIntegrationTests.java

@@ -47,14 +47,16 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.1
  *
  */
 @SpringJUnitConfig
 @RabbitAvailable(queues = BoundRabbitChannelAdviceIntegrationTests.QUEUE)
 public class BoundRabbitChannelAdviceIntegrationTests {
 
-	public static final String QUEUE = "dedicated.advice";
+	static final String QUEUE = "dedicated.advice";
 
 	@Autowired
 	private Config.Gate gate;
@@ -63,7 +65,7 @@ public class BoundRabbitChannelAdviceIntegrationTests {
 	private Config config;
 
 	@Test
-	public void testAdvice() throws Exception {
+	void testAdvice() throws Exception {
 		BoundRabbitChannelAdvice advice = this.config.advice(this.config.template());
 		Log logger = spy(TestUtils.getPropertyValue(advice, "logger", Log.class));
 		new DirectFieldAccessor(advice).setPropertyValue("logger", logger);
@@ -88,14 +90,14 @@ public static class Config {
 		private final List<String> received = new ArrayList<>();
 
 		@Bean
-		public CachingConnectionFactory cf() throws Exception {
+		public CachingConnectionFactory cf() {
 			CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
-			ccf.setSimplePublisherConfirms(true);
+			ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
 			return ccf;
 		}
 
 		@Bean
-		public RabbitTemplate template() throws Exception {
+		public RabbitTemplate template() {
 			return new RabbitTemplate(cf());
 		}
 

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdviceTests.java

@@ -50,6 +50,8 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.1
  *
  */
@@ -63,7 +65,7 @@ public class BoundRabbitChannelAdviceTests {
 	private Config config;
 
 	@Test
-	public void testAdvice() throws Exception {
+	void testAdvice() throws Exception {
 		this.gate.send("a,b,c");
 		verify(this.config.connection, times(1)).createChannel();
 		verify(this.config.channel).confirmSelect();
@@ -97,7 +99,7 @@ public CachingConnectionFactory cf() throws Exception {
 			}).given(cf).newConnection((ExecutorService) isNull(), anyString());
 			cf.setAutomaticRecoveryEnabled(false);
 			CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
-			ccf.setSimplePublisherConfirms(true);
+			ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
 			return ccf;
 		}