GH-1331: ThreadChannelConnFactory Improvements

Resolves https://github.com/spring-projects/spring-amqp/issues/1331

Provide a mechanism so that a thread can hand off its channel(s) to another thread.
Add protection to close a channel that would be orphaned if a second channel is transferred
to the thread and the thread failed to close its channel beforehand.

Other improvements:

- only call the channel listener when a channel is actually created
- physically close a transactional channel that is no longer in the thread local because
  `closeThreadChannel` was called
- move reset of physical close flag to the actual close

* Suggestions from PR review + other improvements:

- protect against calling 'prepare' when no channels bound
- log warning for unclaimed context switches
- add more tests

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -17,8 +17,12 @@
 package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
 
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
@@ -28,6 +32,7 @@ import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConnectionFactory;
@@ -44,6 +49,10 @@ import com.rabbitmq.client.ShutdownListener;
  */
 public class ThreadChannelConnectionFactory extends AbstractConnectionFactory implements ShutdownListener {
 
+	private final Map<UUID, Context> contextSwitches = new ConcurrentHashMap<>();
+
+	private final Map<UUID, Thread> switchesInProgress = new ConcurrentHashMap<>();
+
 	private volatile ConnectionWrapper connection;
 
 	private boolean simplePublisherConfirms;
@@ -142,6 +151,79 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 			this.connection.forceClose();
 			this.connection = null;
 		}
+		if (this.switchesInProgress.size() > 0) {
+			if (this.logger.isWarnEnabled()) {
+				this.logger.warn("Unclaimed context switches from threads:" +
+						this.switchesInProgress.values()
+								.stream()
+								.map(t -> t.getName())
+								.collect(Collectors.toList()));
+			}
+		}
+		this.contextSwitches.clear();
+		this.switchesInProgress.clear();
+	}
+
+	/**
+	 * Call to prepare to switch the channel(s) owned by this thread to another thread.
+	 * @return an opaque object representing the context to switch. If there are no channels
+	 * or no open channels assigned to this thread, null is returned.
+	 * @since 2.3.7
+	 * @see #switchContext(Object)
+	 */
+	@Nullable
+	public Object prepareSwitchContext() {
+		return prepareSwitchContext(UUID.randomUUID());
+	}
+
+	@Nullable
+	Object prepareSwitchContext(UUID uuid) {
+		Object pubContext = null;
+		if (getPublisherConnectionFactory() instanceof ThreadChannelConnectionFactory) {
+			pubContext = ((ThreadChannelConnectionFactory) getPublisherConnectionFactory()).prepareSwitchContext(uuid);
+		}
+		Context context = ((ConnectionWrapper) createConnection()).prepareSwitchContext();
+		if (context.getNonTx() == null && context.getTx() == null) {
+			this.logger.debug("No channels are bound to this thread");
+			return pubContext;
+		}
+		if (this.switchesInProgress.values().contains(Thread.currentThread())) {
+			this.logger
+					.warn("A previous context switch from this thread has not been claimed yet; possible memory leak?");
+		}
+		this.contextSwitches.put(uuid, context);
+		this.switchesInProgress.put(uuid, Thread.currentThread());
+		return uuid;
+	}
+
+	/**
+	 * Acquire ownership of another thread's channel(s) after that thread called
+	 * {@link #prepareSwitchContext()}.
+	 * @param toSwitch the context returned by {@link #prepareSwitchContext()}.
+	 * @since 2.3.7
+	 * @see #prepareSwitchContext()
+	 */
+	public void switchContext(@Nullable Object toSwitch) {
+		if (toSwitch != null) {
+			Assert.state(doSwitch(toSwitch), () -> "No context to switch for " + toSwitch.toString());
+		}
+		else {
+			this.logger.debug("Attempted to switch a null context - no channels to acquire");
+		}
+	}
+
+	boolean doSwitch(@Nullable Object toSwitch) {
+		boolean switched = false;
+		if (getPublisherConnectionFactory() instanceof ThreadChannelConnectionFactory) {
+			switched = ((ThreadChannelConnectionFactory) getPublisherConnectionFactory()).doSwitch(toSwitch);
+		}
+		Context context = this.contextSwitches.remove(toSwitch);
+		this.switchesInProgress.remove(toSwitch);
+		if (context != null) {
+			((ConnectionWrapper) createConnection()).switchContext(context);
+			switched = true;
+		}
+		return switched;
 	}
 
 	private final class ConnectionWrapper extends SimpleConnection {
@@ -183,8 +265,8 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 					}
 					this.channels.set(channel);
 				}
+				getChannelListener().onCreate(channel, transactional);
 			}
-			getChannelListener().onCreate(channel, transactional);
 			return channel;
 		}
 
@@ -223,7 +305,7 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 
 		private void handleClose(Channel channel, boolean transactional) {
 
-			if (ConnectionWrapper.this.channels.get() == null) {
+			if (transactional && this.txChannels.get() == null ? true : this.channels.get() == null) {
 				physicalClose(channel);
 			}
 			else {
@@ -235,7 +317,6 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 					else {
 						this.channels.remove();
 					}
-					RabbitUtils.clearPhysicalCloseRequired();
 				}
 			}
 		}
@@ -266,6 +347,9 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 				catch (IOException | TimeoutException e) {
 					logger.debug("Error on close", e);
 				}
+				finally {
+					RabbitUtils.clearPhysicalCloseRequired();
+				}
 			}
 		}
 
@@ -274,6 +358,54 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 			getConnectionListener().onClose(this);
 		}
 
+		Context prepareSwitchContext() {
+			Context context = new Context(this.channels.get(), this.txChannels.get());
+			this.channels.remove();
+			this.txChannels.remove();
+			return context;
+		}
+
+		void switchContext(Context context) {
+			if (context.getNonTx() != null) {
+				doSwitch(context.getNonTx(), this.channels);
+			}
+			if (context.getTx() != null) {
+				doSwitch(context.getTx(), this.txChannels);
+			}
+		}
+
+		private void doSwitch(Channel channel, ThreadLocal<Channel> channelTL) {
+			Channel toClose = channelTL.get();
+			if (toClose != null) {
+				RabbitUtils.setPhysicalCloseRequired(channel, true);
+				physicalClose(toClose);
+			}
+			channelTL.set(channel);
+		}
+
+	}
+
+	private static class Context {
+
+		private final Channel nonTx;
+
+		private final Channel tx;
+
+		Context(@Nullable Channel nonTx, @Nullable Channel tx) {
+			this.nonTx = nonTx;
+			this.tx = tx;
+		}
+
+		@Nullable
+		Channel getNonTx() {
+			return this.nonTx;
+		}
+
+		@Nullable
+		Channel getTx() {
+			return this.tx;
+		}
+
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2021 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.
@@ -17,20 +17,39 @@
 package org.springframework.amqp.rabbit.connection;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
 import org.springframework.amqp.utils.test.TestUtils;
+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.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConnectionFactory;
 
@@ -39,7 +58,7 @@ import com.rabbitmq.client.ConnectionFactory;
  * @since 2.3
  *
  */
-@RabbitAvailable
+@RabbitAvailable(queues = "ThreadChannelConnectionFactoryTests.q1")
 @SpringJUnitConfig
 @DirtiesContext
 public class ThreadChannelConnectionFactoryTests {
@@ -48,8 +67,8 @@ public class ThreadChannelConnectionFactoryTests {
 	void testBasic() throws Exception {
 		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
 		rabbitConnectionFactory.setHost("localhost");
-		ThreadChannelConnectionFactory scf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
-		Connection conn = scf.createConnection();
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		Connection conn = tccf.createConnection();
 		Channel chann1 = conn.createChannel(false);
 		chann1.close();
 		Channel chann2 = conn.createChannel(false);
@@ -78,12 +97,12 @@ public class ThreadChannelConnectionFactoryTests {
 		chann2 = conn.createChannel(false);
 		RabbitUtils.setPhysicalCloseRequired(chann2, true);
 		chann2.close();
-		scf.setSimplePublisherConfirms(true);
+		tccf.setSimplePublisherConfirms(true);
 		chann1 = conn.createChannel(false);
 		assertThat(chann1).isNotSameAs(chann2);
 		assertThat(((ChannelProxy) chann1).isConfirmSelected()).isTrue();
 		chann1.close();
-		scf.destroy();
+		tccf.destroy();
 		assertThat(((Channel) TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isOpen())
 				.isFalse();
 		assertThat(((Channel) TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isOpen())
@@ -104,6 +123,227 @@ public class ThreadChannelConnectionFactoryTests {
 		assertThat(config.closed).isTrue();
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void contextSwitch() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		TaskExecutor exec = new SimpleAsyncTaskExecutor();
+		AtomicReference<Channel> nonTx = new AtomicReference<>();
+		AtomicReference<Channel> tx = new AtomicReference<>();
+		BlockingQueue<Object> context = new LinkedBlockingQueue<>();
+		exec.execute(() -> {
+			Connection conn = tccf.createConnection();
+			nonTx.set(conn.createChannel(false));
+			tx.set(conn.createChannel(true));
+			Object ctx = tccf.prepareSwitchContext();
+			assertThat(tccf.prepareSwitchContext()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNull();
+			context.add(ctx);
+		});
+		Object ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		assertThatIllegalStateException().isThrownBy(() -> tccf.switchContext(ctx));
+		Connection conn = tccf.createConnection();
+		Channel chann1 = conn.createChannel(false);
+		assertThat(chann1).isSameAs(nonTx.get());
+		Channel chann2 = conn.createChannel(true);
+		assertThat(chann2).isSameAs(tx.get());
+		assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty();
+		tccf.switchContext(null); // test no-op
+		tccf.destroy();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void contextSwitchMulti() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		TaskExecutor exec = new SimpleAsyncTaskExecutor();
+		AtomicReference<Channel> nonTx = new AtomicReference<>();
+		AtomicReference<Channel> tx = new AtomicReference<>();
+		BlockingQueue<Object> context = new LinkedBlockingQueue<>();
+		CountDownLatch latch = new CountDownLatch(1);
+		exec.execute(() -> {
+			Connection conn = tccf.createConnection();
+			nonTx.set(conn.createChannel(false));
+			tx.set(conn.createChannel(true));
+			Object ctx = tccf.prepareSwitchContext();
+			assertThat(tccf.prepareSwitchContext()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn, "txChannels", ThreadLocal.class).get()).isNull();
+			conn.createChannel(false);
+			context.add(ctx);
+			context.add(tccf.prepareSwitchContext());
+			latch.countDown();
+		});
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		Object ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		Connection conn = tccf.createConnection();
+		Channel chann1 = conn.createChannel(false);
+		assertThat(chann1).isSameAs(nonTx.get());
+		Channel chann2 = conn.createChannel(true);
+		assertThat(chann2).isSameAs(tx.get());
+		assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).hasSize(1);
+		ctx = context.poll(10, TimeUnit.SECONDS);
+		tccf.switchContext(ctx);
+		assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty();
+		tccf.destroy();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void contextSwitchBothFactories() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		TaskExecutor exec = new SimpleAsyncTaskExecutor();
+		AtomicReference<Channel> nonTx1 = new AtomicReference<>();
+		AtomicReference<Channel> tx1 = new AtomicReference<>();
+		BlockingQueue<Object> context = new LinkedBlockingQueue<>();
+		AtomicReference<Channel> nonTx2 = new AtomicReference<>();
+		AtomicReference<Channel> tx2 = new AtomicReference<>();
+		exec.execute(() -> {
+			Connection conn1 = tccf.createConnection();
+			nonTx1.set(conn1.createChannel(false));
+			tx1.set(conn1.createChannel(true));
+			org.springframework.amqp.rabbit.connection.ConnectionFactory pcf = tccf.getPublisherConnectionFactory();
+			Connection conn2 = pcf.createConnection();
+			nonTx2.set(conn2.createChannel(false));
+			tx2.set(conn2.createChannel(true));
+			Object ctx = tccf.prepareSwitchContext();
+			assertThat(tccf.prepareSwitchContext()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn1, "channels", ThreadLocal.class).get()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn1, "txChannels", ThreadLocal.class).get()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn2, "channels", ThreadLocal.class).get()).isNull();
+			assertThat(TestUtils.getPropertyValue(conn2, "txChannels", ThreadLocal.class).get()).isNull();
+			context.add(ctx);
+		});
+		Object ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		assertThatIllegalStateException().isThrownBy(() -> tccf.switchContext(ctx));
+		Connection conn1 = tccf.createConnection();
+		Channel chann1 = conn1.createChannel(false);
+		assertThat(chann1).isSameAs(nonTx1.get());
+		Channel chann2 = conn1.createChannel(true);
+		assertThat(chann2).isSameAs(tx1.get());
+		org.springframework.amqp.rabbit.connection.ConnectionFactory pcf = tccf.getPublisherConnectionFactory();
+		Connection conn2 = pcf.createConnection();
+		Channel chann3 = conn2.createChannel(false);
+		assertThat(chann3).isSameAs(nonTx2.get());
+		Channel chann4 = conn2.createChannel(true);
+		assertThat(chann4).isSameAs(tx2.get());
+		assertThat(TestUtils.getPropertyValue(tccf, "switchesInProgress", Map.class)).isEmpty();
+		assertThat(TestUtils.getPropertyValue(tccf, "publisherConnectionFactory.switchesInProgress", Map.class))
+				.isEmpty();
+		tccf.destroy();
+	}
+
+	@Test
+	void contextSwitchViaTemplate() throws Exception {
+		// Template uses the nested publisher factory
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		RabbitTemplate template = new RabbitTemplate(tccf);
+		TaskExecutor exec = new SimpleAsyncTaskExecutor();
+		AtomicReference<Channel> nonTx = new AtomicReference<>();
+		AtomicReference<Channel> tx = new AtomicReference<>();
+		BlockingQueue<Object> context = new LinkedBlockingQueue<>();
+		exec.execute(() -> {
+			template.execute(this::sendChannelNameAsBody);
+			context.add(tccf.prepareSwitchContext());
+		});
+		Object ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		template.execute(this::sendChannelNameAsBody);
+		Message received1 = template.receive("ThreadChannelConnectionFactoryTests.q1", 10_000);
+		Message received2 = template.receive("ThreadChannelConnectionFactoryTests.q1", 10_000);
+		assertThat(new String(received1.getBody())).isEqualTo(new String(received2.getBody()));
+		tccf.destroy();
+	}
+
+	@Test
+	void orphanClosed() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		RabbitTemplate template = new RabbitTemplate(tccf);
+		TaskExecutor exec = new SimpleAsyncTaskExecutor();
+		AtomicReference<Channel> nonTx = new AtomicReference<>();
+		AtomicReference<Channel> tx = new AtomicReference<>();
+		BlockingQueue<Object> context = new LinkedBlockingQueue<>();
+		Runnable task = () -> {
+			template.execute(channel -> null);
+			context.add(tccf.prepareSwitchContext());
+		};
+		exec.execute(task);
+		Object ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		Channel toBeOrphaned = template.execute(channel -> channel);
+		exec.execute(task);
+		ctx = context.poll(10, TimeUnit.SECONDS);
+		assertThat(ctx).isNotNull();
+		tccf.switchContext(ctx);
+		template.execute(channel -> null);
+		assertThat(toBeOrphaned.isOpen()).isFalse();
+		tccf.destroy();
+	}
+
+	@Test
+	void unclaimed() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		Log log = spy(TestUtils.getPropertyValue(tccf, "logger", Log.class));
+		new DirectFieldAccessor(tccf).setPropertyValue("logger", log);
+		given(log.isWarnEnabled()).willReturn(true);
+		Connection conn = tccf.createConnection();
+		conn.createChannel(false);
+		tccf.prepareSwitchContext();
+		tccf.destroy();
+		verify(log).warn("Unclaimed context switches from threads:[main]");
+	}
+
+	@Test
+	void neitherBound() throws Exception {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		rabbitConnectionFactory.setAutomaticRecoveryEnabled(false);
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+		Log log1 = spy(TestUtils.getPropertyValue(tccf, "logger", Log.class));
+		new DirectFieldAccessor(tccf).setPropertyValue("logger", log1);
+		Log log2 = spy(TestUtils.getPropertyValue(tccf, "publisherConnectionFactory.logger", Log.class));
+		new DirectFieldAccessor(tccf).setPropertyValue("publisherConnectionFactory.logger", log2);
+		given(log1.isDebugEnabled()).willReturn(true);
+		given(log2.isDebugEnabled()).willReturn(true);
+		assertThat(tccf.prepareSwitchContext()).isNull();
+		verify(log1).debug("No channels are bound to this thread");
+		verify(log2).debug("No channels are bound to this thread");
+		tccf.destroy();
+	}
+
+	Channel sendChannelNameAsBody(Channel channel) throws IOException {
+		channel.basicPublish("", "ThreadChannelConnectionFactoryTests.q1", new BasicProperties(),
+				channel.toString().getBytes());
+		return channel;
+	}
+
 	@Configuration
 	public static class Config {
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java

@@ -66,7 +66,6 @@ import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
 import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.SerializationUtils;
@@ -576,7 +575,8 @@ public class RabbitTemplateTests {
 		assertThatIllegalStateException().isThrownBy(() ->
 				template.setReturnCallback(mock(RabbitTemplate.ReturnCallback.class)));
 		RabbitTemplate template2 = new RabbitTemplate();
-		ReturnCallback callback = mock(RabbitTemplate.ReturnCallback.class);
+		org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback callback =
+				mock(org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback.class);
 		template2.setReturnCallback(callback);
 		template2.setReturnCallback(callback);
 	}

src/reference/asciidoc/amqp.adoc

@@ -257,6 +257,8 @@ This factory manages a single connection and two `ThreadLocal` s, one for transa
 This factory ensures that all operations on the same thread use the same channel (as long as it remains open).
 This facilitates strict message ordering without the need for <<scoped-operations>>.
 To avoid memory leaks, if your application uses many short-lived threads, you must call the factory's `closeThreadChannel()` to release the channel resource.
+Starting with version 2.3.7, a thread can transfer its channel(s) to another thread.
+See <<multi-strict>> for more information.
 
 ====== `CachingConnectionFactory`
 
@@ -1386,6 +1388,97 @@ class Service {
 
 Even though the publishing is performed on two different threads, they will both use the same channel because the cache is capped at a single channel.
 
+Starting with version 2.3.7, the `ThreadChannelConnectionFactory` supports transferring a thread's channel(s) to another thread, using the `prepareContextSwitch` and `switchContext` methods.
+The first method returns a context which is passed to the second thread which calls the second method.
+A thread can have either a non-transactional channel or a transactional channel (or one of each) bound to it; you cannot transfer them individually, unless you use two connection factories.
+An example follows:
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class Application {
+
+	private static final Logger log = LoggerFactory.getLogger(Application.class);
+
+	public static void main(String[] args) {
+		SpringApplication.run(Application.class, args);
+	}
+
+	@Bean
+	TaskExecutor exec() {
+		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
+		exec.setCorePoolSize(10);
+		return exec;
+	}
+
+	@Bean
+	ThreadChannelConnectionFactory tccf() {
+		ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
+		rabbitConnectionFactory.setHost("localhost");
+		return new ThreadChannelConnectionFactory(rabbitConnectionFactory);
+	}
+
+	@RabbitListener(queues = "queue")
+	void listen(String in) {
+		log.info(in);
+	}
+
+	@Bean
+	Queue queue() {
+		return new Queue("queue");
+	}
+
+
+	@Bean
+	public ApplicationRunner runner(Service service, TaskExecutor exec) {
+		return args -> {
+			exec.execute(() -> service.mainService("test"));
+		};
+	}
+
+}
+
+@Component
+class Service {
+
+	private static final Logger LOG = LoggerFactory.getLogger(Service.class);
+
+	private final RabbitTemplate template;
+
+	private final TaskExecutor exec;
+
+	private final ThreadChannelConnectionFactory connFactory;
+
+	Service(RabbitTemplate template, TaskExecutor exec,
+			ThreadChannelConnectionFactory tccf) {
+
+		this.template = template;
+		this.exec = exec;
+		this.connFactory = tccf;
+	}
+
+	void mainService(String toSend) {
+		LOG.info("Publishing from main service");
+		this.template.convertAndSend("queue", toSend);
+		Object context = this.connFactory.prepareSwitchContext();
+		this.exec.execute(() -> secondaryService(toSend.toUpperCase(), context));
+	}
+
+	void secondaryService(String toSend, Object threadContext) {
+		LOG.info("Publishing from secondary service");
+		this.connFactory.switchContext(threadContext);
+		this.template.convertAndSend("queue", toSend);
+		this.connFactory.closeThreadChannel();
+	}
+
+}
+----
+====
+
+IMPORTANT: Once the `prepareSwitchContext` is called, if the current thread performs any more operations, they will be performed on a new channel.
+It is important to close the thread-bound channel when it is no longer needed.
+
 [[template-messaging]]
 ===== Messaging Integration