Use native connection factory with message listener containers

This commit updates the auto-configuration to use the native connection
factory for configuring message listener containers. Previously, the
connection factory that could have been wrapped in a caching connection
factory was used.

While using a caching connection factory is suitable for sending
messages (i.e. JmsTemplate usage), it isn't for message listeners as
they need to own the connection for local recovery purposes.

Closes gh-39816

spring-boot-project/spring-boot/build.gradle

@@ -73,11 +73,15 @@ dependencies {
 	optional("org.liquibase:liquibase-core") {
 		exclude(group: "javax.xml.bind", module: "jaxb-api")
 	}
+	optional("org.messaginghub:pooled-jms") {
+		exclude group: "org.apache.geronimo.specs", module: "geronimo-jms_2.0_spec"
+	}
 	optional("org.postgresql:postgresql")
 	optional("org.slf4j:jul-to-slf4j")
 	optional("org.slf4j:slf4j-api")
 	optional("org.springframework:spring-messaging")
 	optional("org.springframework:spring-orm")
+	optional("org.springframework:spring-jms")
 	optional("org.springframework:spring-oxm")
 	optional("org.springframework:spring-r2dbc")
 	optional("org.springframework:spring-test")

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/jms/ConnectionFactoryUnwrapper.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2012-2024 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
+ *
+ *      https://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.boot.jms;
+
+import jakarta.jms.ConnectionFactory;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import org.springframework.jms.connection.CachingConnectionFactory;
+
+/**
+ * Unwrap a {@link ConnectionFactory} that may have been wrapped to perform caching or
+ * pooling.
+ *
+ * @author Stephane Nicoll
+ * @since 6.4.0
+ */
+public abstract class ConnectionFactoryUnwrapper {
+
+	/**
+	 * Return the native {@link ConnectionFactory} by unwrapping it from a cache or pool
+	 * connection factory. Return the given {@link ConnectionFactory} if no caching
+	 * wrapper has been detected.
+	 * @param connectionFactory a connection factory
+	 * @return the native connection factory that it wraps, if any
+	 */
+	public static ConnectionFactory unwrap(ConnectionFactory connectionFactory) {
+		if (connectionFactory instanceof CachingConnectionFactory ccf) {
+			return unwrap(ccf.getTargetConnectionFactory());
+		}
+		ConnectionFactory unwrapedConnectionFactory = unwrapFromJmsPoolConnectionFactory(connectionFactory);
+		return (unwrapedConnectionFactory != null) ? unwrap(unwrapedConnectionFactory) : connectionFactory;
+	}
+
+	private static ConnectionFactory unwrapFromJmsPoolConnectionFactory(ConnectionFactory connectionFactory) {
+		try {
+			if (connectionFactory instanceof JmsPoolConnectionFactory poolConnectionFactory) {
+				return (ConnectionFactory) poolConnectionFactory.getConnectionFactory();
+			}
+		}
+		catch (Throwable ex) {
+			// ignore
+		}
+		return null;
+	}
+
+}

spring-boot-project/spring-boot/src/test/java/org/springframework/boot/jms/ConnectionFactoryUnwrapperTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2012-2024 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
+ *
+ *      https://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.boot.jms;
+
+import jakarta.jms.ConnectionFactory;
+import org.junit.jupiter.api.Test;
+import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
+
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.connection.SingleConnectionFactory;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link ConnectionFactoryUnwrapper}.
+ *
+ * @author Stephane Nicoll
+ */
+class ConnectionFactoryUnwrapperTests {
+
+	@Test
+	void unwrapWithSingleConnectionFactory() {
+		ConnectionFactory connectionFactory = new SingleConnectionFactory();
+		assertThat(ConnectionFactoryUnwrapper.unwrap(connectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(connectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithCachingConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(new CachingConnectionFactory(connectionFactory)))
+			.isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithNestedCachingConnectionFactories() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		CachingConnectionFactory firstCachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
+		CachingConnectionFactory secondCachingConnectionFactory = new CachingConnectionFactory(
+				firstCachingConnectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(secondCachingConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithJmsPoolConnectionFactory() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		JmsPoolConnectionFactory poolConnectionFactory = new JmsPoolConnectionFactory();
+		poolConnectionFactory.setConnectionFactory(connectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(poolConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+	@Test
+	void unwrapWithNestedJmsPoolConnectionFactories() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		JmsPoolConnectionFactory firstPooledConnectionFactory = new JmsPoolConnectionFactory();
+		firstPooledConnectionFactory.setConnectionFactory(connectionFactory);
+		JmsPoolConnectionFactory secondPooledConnectionFactory = new JmsPoolConnectionFactory();
+		secondPooledConnectionFactory.setConnectionFactory(firstPooledConnectionFactory);
+		assertThat(ConnectionFactoryUnwrapper.unwrap(secondPooledConnectionFactory)).isSameAs(connectionFactory);
+	}
+
+}