Add BoundRabbitChannelAdvice

Polishing and docs

Polishing - DEBUG log for confirms; add integration test

Polishing - PR Comments

Renamed Advice

Verify acks logged.

Polishing - more PR comments

Renamed to BoundRabbitChannelAdvice.

* Extract `ConfirmCallback`s instances for optimization
* Remove unused constant

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.support;
+
+import java.lang.reflect.UndeclaredThrowableException;
+import java.time.Duration;
+
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.amqp.rabbit.core.RabbitOperations;
+import org.springframework.integration.handler.advice.HandleMessageAdvice;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
+
+import com.rabbitmq.client.ConfirmCallback;
+
+/**
+ * An advice that causes all downstream {@link RabbitOperations} operations to be executed
+ * on the same channel, as long as there are no thread handoffs, since the channel is
+ * bound to the thread. The same RabbitOperations must be used in this and all downstream
+ * components. Typically used with a splitter or some other mechanism that would cause
+ * multiple messages to be sent. Optionally waits for publisher confirms if the channel is
+ * so configured.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 5.1
+ *
+ */
+public class BoundRabbitChannelAdvice implements HandleMessageAdvice {
+
+	private final Log logger = LogFactory.getLog(getClass());
+
+	private final RabbitOperations operations;
+
+	private final Duration waitForConfirmsTimeout;
+
+	private final ConfirmCallback ackCallback = this::handleAcks;
+
+	private final ConfirmCallback nackCallback = this::handleNacks;
+
+	/**
+	 * Construct an instance that doesn't wait for confirms.
+	 * @param operations the operations.
+	 */
+	public BoundRabbitChannelAdvice(RabbitOperations operations) {
+		this(operations, null);
+	}
+
+	/**
+	 * Construct an instance that waits for publisher confirms (if
+	 * configured and waitForConfirmsTimeout is not null).
+	 * @param operations the operations.
+	 * @param waitForConfirmsTimeout the timeout.
+	 */
+	public BoundRabbitChannelAdvice(RabbitOperations operations, @Nullable Duration waitForConfirmsTimeout) {
+		Assert.notNull(operations, "'operations' cannot be null");
+		this.operations = operations;
+		this.waitForConfirmsTimeout = waitForConfirmsTimeout;
+	}
+
+	@Override
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		try {
+			return this.operations.invoke(operations -> {
+				try {
+					Object result = invocation.proceed();
+					if (this.waitForConfirmsTimeout != null) {
+						this.operations.waitForConfirmsOrDie(this.waitForConfirmsTimeout.toMillis());
+					}
+					return result;
+				}
+				catch (Throwable t) { // NOSONAR - rethrown below
+					ReflectionUtils.rethrowRuntimeException(t);
+					return null; // not reachable - satisfy compiler
+				}
+			}, this.ackCallback, this.nackCallback);
+		}
+		catch (UndeclaredThrowableException ute) {
+			throw ute.getCause();
+		}
+	}
+
+	private void handleAcks(long deliveryTag, boolean multiple) {
+		doHandleAcks(deliveryTag, multiple, true);
+	}
+
+	private void handleNacks(long deliveryTag, boolean multiple) {
+		doHandleAcks(deliveryTag, multiple, false);
+	}
+
+	private void doHandleAcks(long deliveryTag, boolean multiple, boolean ack) {
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug("Publisher confirm " + (!ack ? "n" : "") + "ack: " + deliveryTag + ", " +
+					"multiple: " + multiple);
+		}
+	}
+
+}

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

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.support;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.spy;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.amqp.dsl.Amqp;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 5.1
+ *
+ */
+@SpringJUnitConfig
+@RabbitAvailable(queues = BoundRabbitChannelAdviceIntegrationTests.QUEUE)
+public class BoundRabbitChannelAdviceIntegrationTests {
+
+	public static final String QUEUE = "dedicated.advice";
+
+	@Autowired
+	private Config.Gate gate;
+
+	@Autowired
+	private Config config;
+
+	@Test
+	public 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);
+		willReturn(true).given(logger).isDebugEnabled();
+		final CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(i -> {
+			latch.countDown();
+			return i.callRealMethod();
+		}).given(logger).debug(anyString());
+		this.gate.send("a,b,c");
+		assertTrue(this.config.latch.await(10, TimeUnit.SECONDS));
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertThat(this.config.received).containsExactly("A", "B", "C");
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		private final CountDownLatch latch = new CountDownLatch(3);
+
+		private final List<String> received = new ArrayList<>();
+
+		@Bean
+		public CachingConnectionFactory cf() throws Exception {
+			CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
+			ccf.setSimplePublisherConfirms(true);
+			return ccf;
+		}
+
+		@Bean
+		public RabbitTemplate template() throws Exception {
+			return new RabbitTemplate(cf());
+		}
+
+		@Bean
+		public BoundRabbitChannelAdvice advice(RabbitTemplate template) {
+			return new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10));
+		}
+
+		@Bean
+		public IntegrationFlow flow(RabbitTemplate template, BoundRabbitChannelAdvice advice) {
+			return IntegrationFlows.from(Gate.class)
+					.split(s -> s.delimiters(",")
+							.advice(advice))
+					.<String, String>transform(String::toUpperCase)
+					.handle(Amqp.outboundAdapter(template).routingKey(QUEUE))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow listener(CachingConnectionFactory ccf) {
+			return IntegrationFlows.from(Amqp.inboundAdapter(ccf, QUEUE))
+					.handle(m -> {
+						received.add((String) m.getPayload());
+						this.latch.countDown();
+					})
+					.get();
+		}
+
+		public interface Gate {
+
+			void send(String out);
+
+		}
+
+	}
+
+}

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

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.amqp.support;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+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;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.amqp.dsl.Amqp;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * @author Gary Russell
+ * @since 5.1
+ *
+ */
+@SpringJUnitConfig
+public class BoundRabbitChannelAdviceTests {
+
+	@Autowired
+	private Config.Gate gate;
+
+	@Autowired
+	private Config config;
+
+	@Test
+	public void testAdvice() throws Exception {
+		this.gate.send("a,b,c");
+		verify(this.config.connection, times(1)).createChannel();
+		verify(this.config.channel).confirmSelect();
+		verify(this.config.channel).basicPublish(eq(""), eq("rk"), anyBoolean(), any(), eq("A".getBytes()));
+		verify(this.config.channel).basicPublish(eq(""), eq("rk"), anyBoolean(), any(), eq("B".getBytes()));
+		verify(this.config.channel).basicPublish(eq(""), eq("rk"), anyBoolean(), any(), eq("C".getBytes()));
+		verify(this.config.channel).waitForConfirmsOrDie(10_000L);
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+		private Connection connection;
+
+		private Channel channel;
+
+		@Bean
+		public CachingConnectionFactory cf() throws Exception {
+			ConnectionFactory cf = mock(ConnectionFactory.class);
+			cf.setHost("localhost");
+			cf = spy(cf);
+			willAnswer(i -> {
+				this.connection = mock(Connection.class);
+				willAnswer(ii -> {
+					this.channel = mock(Channel.class);
+					given(this.channel.isOpen()).willReturn(true);
+					return this.channel;
+				}).given(this.connection).createChannel();
+				return this.connection;
+			}).given(cf).newConnection((ExecutorService) isNull(), anyString());
+			cf.setAutomaticRecoveryEnabled(false);
+			CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
+			ccf.setSimplePublisherConfirms(true);
+			return ccf;
+		}
+
+		@Bean
+		public RabbitTemplate template() throws Exception {
+			return new RabbitTemplate(cf());
+		}
+
+		@Bean
+		public IntegrationFlow flow(RabbitTemplate template) {
+			return IntegrationFlows.from(Gate.class)
+					.split(s -> s.delimiters(",")
+							.advice(new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10))))
+					.<String, String>transform(String::toUpperCase)
+					.handle(Amqp.outboundAdapter(template).routingKey("rk"))
+					.get();
+		}
+
+		public interface Gate {
+
+			void send(String out);
+
+		}
+
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -1298,7 +1298,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * Typically used with a Java 8 Lambda expression:
 	 * <pre class="code">
 	 * {@code
-	 *  .split(s -> s.applySequence(false).get().getT2().setDelimiters(","))
+	 *  .split(s -> s.applySequence(false).delimiters(","))
 	 * }
 	 * </pre>
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options

src/reference/asciidoc/amqp.adoc

@@ -1504,6 +1504,58 @@ Negated patterns get priority, so a list such as
 IMPORTANT: If you have a user defined header that begins with `!` that you *do* wish to map, you need to escape it with
 `\` thus: `STANDARD_REQUEST_HEADERS,\!myBangHeader` and it *WILL* be mapped.
 
+[[amqp-strict-ordering]]
+=== Strict Message Ordering
+
+==== Inbound
+
+If you require strict ordering of inbound messages, you must configure the inbound listener container's `prefetchCount` property to `1`.
+This is because if a message fails and is redelivered, it will arrive after existing prefetched messages.
+Since Spring AMQP _version 2.0_, the `prefetchCount` defaults to `250` for improved performance.
+Strict ordering requirements come at the cost of decreased performance.
+
+==== Outbound
+
+Consider the following simple integration flow:
+
+[source, java]
+----
+@Bean
+public IntegrationFlow flow(RabbitTemplate template) {
+    return IntegrationFlows.from(Gateway.class)
+            .split(s -> s.delimiters(","))
+            .<String, String>transform(String::toUpperCase)
+            .handle(Amqp.outboundAdapter(template).routingKey("rk"))
+            .get();
+}
+----
+
+Let's say we send a message `a,b,c` to the gateway, while it is likely that messages `A`, `B`, `C` will be sent in order, there is no guarantee.
+This is because the template "borrows" a channel from the cache for each send and there is no guarantee that the same channel will be used for each.
+One solution is to start a transaction before the splitter, but transactions are very expensive in RabbitMQ and can reduce performance several hundred fold.
+
+To solve this problem in a more efficient manner, starting with _version 5.1_, Spring Integration provides the `BoundRabbitChannelAdvice` which is a `HandleMessageAdvice` - see <<handle-message-advice>>.
+When applied before the splitter, this ensures that all downstream operations are performed on the same channel and, optionally, can wait until publisher confirms for all sent messages are received (if the connection factory is configured for confirms).
+
+[source, java]
+----
+@Bean
+public IntegrationFlow flow(RabbitTemplate template) {
+    return IntegrationFlows.from(Gateway.class)
+            .split(s -> s.delimiters(",")
+                    .advice(new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10))))
+            .<String, String>transform(String::toUpperCase)
+            .handle(Amqp.outboundAdapter(template).routingKey("rk"))
+            .get();
+}
+----
+
+Notice that the same `RabbitTemplate` (which implements `RabbitOperations`) is used in the advice and the outbound adapter.
+The advice runs the downstream flow within the template's `invoke` method so that all operations run on the same channel.
+If the optional timeout is provided, when the flow completes, the advice calls the `waitForConfirmsOrDie` method, which will throw an exception if the confirms are not received within the specified time.
+
+IMPORTANT: There must be no thread handoffs in the downstream flow (`QueueChannel`, `ExecutorChannel`, etc).
+
 === AMQP Samples
 
 To experiment with the AMQP adapters, check out the samples available in the Spring Integration Samples Git repository at:

src/reference/asciidoc/handler-advice.adoc

@@ -515,6 +515,7 @@ For `MessageHandler` s that produce a reply (`AbstractReplyProducingMessageHandl
 For other message handlers, the advice is applied to `MessageHandler.handleMessage()`.
 
 There are some circumstances where, even if a message handler is an `AbstractReplyProducingMessageHandler`, the advice must be applied to the `handleMessage` method - for example, the <<idempotent-receiver, Idempotent Receiver>> might return `null` and this would cause an exception if the handler's `replyRequired` property is true.
+Another example is the `BoundRabbitChannelAdvice` - see <<amqp-strict-ordering>>.
 
 Starting with _version 4.3.1_, a new `HandleMessageAdvice` and the `AbstractHandleMessageAdvice` base implementation have been introduced.
 `Advice` s that implement `HandleMessageAdvice` will always be applied to the `handleMessage()` method, regardless of the handler type.

src/reference/asciidoc/whats-new.adoc

@@ -9,6 +9,10 @@ If you are interested in more details, please see the Issue Tracker tickets that
 [[x5.1-new-components]]
 === New Components
 
+==== AmqpDedicatedChannelAdvice
+
+See <<amqp-strict-ordering>>.
+
 [[x5.1-general]]
 === General Changes