JMS Supplier/Source

* Addressing PR review comments
2020-06-16 15:34:45 -04:00
parent 9e8be19b2a
commit bdfc69c953
9 changed files with 577 additions and 0 deletions
--- a/pom.xml
+++ b/pom.xml
@@ -77,6 +77,7 @@
 		<module>supplier/geode-supplier</module>
 		<module>supplier/http-supplier</module>
 		<module>supplier/jdbc-supplier</module>
+		<module>supplier/jms-supplier</module>
 		<module>supplier/mongodb-supplier</module>
 		<module>supplier/mqtt-supplier</module>
 		<module>supplier/tcp-supplier</module>
--- a/supplier/jms-supplier/README.adoc
+++ b/supplier/jms-supplier/README.adoc
@@ -0,0 +1,33 @@
+# JMS Supplier
+
+This module provides a JMS supplier that can be reused and composed in other applications.
+The `Supplier` uses the JMS support provided by Spring Framework and Spring Integration under the covers.
+`jmsSupplier` is implemented as a `java.util.function.Supplier`.
+This supplier gives you a reactive stream from JMS sources. The supplier has a signature of `Supplier<Flux<Message<?>>>`.
+Users have to subscribe to this `Flux` and then receive the data.
+
+## Beans for injection
+
+You can import the `JmsSupplierConfiguration` in the application and then inject the following bean.
+
+`jmsSupplier`
+
+You need to inject this as `Supplier<Flux<Message<?>>>`.
+
+You can use `jmsSupplier` as a qualifier when injecting.
+
+Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`.
+
+## Configuration Options
+
+All configuration properties are prefixed with `jms`.
+
+For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierProperties.java[JmsSupplierProperties].
+
+## Examples
+
+See this link:src/test/java/org/springframework/cloud/fn/supplier/jms/[test suite] for the various ways, this supplier is used.
+
+## Other usage
+
+See this link:../../../applications/source/jms-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream JMS Source.
--- a/supplier/jms-supplier/pom.xml
+++ b/supplier/jms-supplier/pom.xml
@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>jms-supplier</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>jms-supplier</name>
+    <description>jms supplier</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-jms</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.jms</groupId>
+            <artifactId>javax.jms-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-broker</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
--- a/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java
+++ b/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java
@@ -0,0 +1,123 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import java.util.function.Supplier;
+
+import javax.jms.ConnectionFactory;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.jms.JmsProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
+import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.SimpleMessageListenerContainer;
+import org.springframework.messaging.Message;
+
+@Configuration
+@EnableConfigurationProperties(JmsSupplierProperties.class)
+public class JmsSupplierConfiguration {
+
+	@Autowired
+	JmsSupplierProperties properties;
+
+	@Autowired
+	private JmsProperties jmsProperties;
+
+	@Autowired
+	private ConnectionFactory connectionFactory;
+
+	@Bean
+	public Supplier<Flux<Message<?>>> jmsSupplier() {
+		return () -> Flux.from(output())
+				.doOnSubscribe(subscription -> adapter().start());
+	}
+
+	@Bean
+	public FluxMessageChannel output() {
+		return new FluxMessageChannel();
+	}
+
+	@Bean
+	public JmsMessageDrivenEndpoint adapter() {
+		final JmsMessageDrivenEndpoint jmsMessageDrivenEndpoint = new JmsMessageDrivenEndpoint(container(), listener());
+		jmsMessageDrivenEndpoint.setAutoStartup(false);
+		return jmsMessageDrivenEndpoint;
+	}
+
+	@Bean
+	public ChannelPublishingJmsMessageListener listener() {
+		ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener();
+		listener.setRequestChannel(output());
+		return listener;
+	}
+
+	@Bean
+	public AbstractMessageListenerContainer container() {
+		AbstractMessageListenerContainer container;
+		JmsProperties.Listener listenerProperties = this.jmsProperties.getListener();
+		if (this.properties.isSessionTransacted()) {
+			DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
+			dmlc.setSessionTransacted(true);
+			if (listenerProperties.getConcurrency() != null) {
+				dmlc.setConcurrentConsumers(listenerProperties.getConcurrency());
+			}
+			if (listenerProperties.getMaxConcurrency() != null) {
+				dmlc.setMaxConcurrentConsumers(listenerProperties.getMaxConcurrency());
+			}
+			container = dmlc;
+		}
+		else {
+			SimpleMessageListenerContainer smlc = new SimpleMessageListenerContainer();
+			smlc.setSessionTransacted(false);
+			if (listenerProperties != null  && listenerProperties.getConcurrency() != null) {
+				smlc.setConcurrentConsumers(listenerProperties.getConcurrency());
+			}
+			container = smlc;
+		}
+		container.setConnectionFactory(this.connectionFactory);
+		if (this.properties.getClientId() != null) {
+			container.setClientId(this.properties.getClientId());
+		}
+		container.setDestinationName(this.properties.getDestination());
+		if (this.properties.getMessageSelector() != null) {
+			container.setMessageSelector(this.properties.getMessageSelector());
+		}
+		container.setPubSubDomain(this.jmsProperties.isPubSubDomain());
+		if (this.properties.getMessageSelector() != null
+				&& listenerProperties.getAcknowledgeMode() != null) {
+			container.setSessionAcknowledgeMode(listenerProperties.getAcknowledgeMode().getMode());
+		}
+		if (this.properties.getSubscriptionDurable() != null) {
+			container.setSubscriptionDurable(this.properties.getSubscriptionDurable());
+		}
+		if (this.properties.getSubscriptionName() != null) {
+			container.setSubscriptionName(this.properties.getSubscriptionName());
+		}
+		if (this.properties.getSubscriptionShared() != null) {
+			container.setSubscriptionShared(this.properties.getSubscriptionShared());
+		}
+		return container;
+	}
+}
--- a/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierProperties.java
+++ b/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierProperties.java
@@ -0,0 +1,126 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Properties for the JMS Supplier.
+ *
+ * @author Gary Russell
+ *
+ */
+@ConfigurationProperties(prefix = "jms.supplier")
+@Validated
+public class JmsSupplierProperties {
+
+	/**
+	 * True to enable transactions and select a DefaultMessageListenerContainer, false to
+	 * select a SimpleMessageListenerContainer.
+	 */
+	private boolean sessionTransacted = true;
+
+	/**
+	 * Client id for durable subscriptions.
+	 */
+	private String clientId;
+
+	/**
+	 * The destination from which to receive messages (queue or topic).
+	 */
+	private String destination;
+
+	/**
+	 * The name of a durable or shared subscription.
+	 */
+	private String subscriptionName;
+
+	/**
+	 * A selector for messages.
+	 */
+	private String messageSelector = null;
+
+	/**
+	 * True for a durable subscription.
+	 */
+	private Boolean subscriptionDurable;
+
+	/**
+	 * True for a shared subscription.
+	 */
+	private Boolean subscriptionShared;
+
+	public String getClientId() {
+		return this.clientId;
+	}
+
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
+	@NotNull
+	public String getDestination() {
+		return this.destination;
+	}
+
+	public void setDestination(String destination) {
+		this.destination = destination;
+	}
+
+	public String getSubscriptionName() {
+		return this.subscriptionName;
+	}
+
+	public void setSubscriptionName(String subscriptionName) {
+		this.subscriptionName = subscriptionName;
+	}
+
+	public String getMessageSelector() {
+		return this.messageSelector;
+	}
+
+	public void setMessageSelector(String messageSelector) {
+		this.messageSelector = messageSelector;
+	}
+
+	public Boolean getSubscriptionDurable() {
+		return this.subscriptionDurable;
+	}
+
+	public void setSubscriptionDurable(Boolean subscriptionDurable) {
+		this.subscriptionDurable = subscriptionDurable;
+	}
+
+	public Boolean getSubscriptionShared() {
+		return this.subscriptionShared;
+	}
+
+	public void setSubscriptionShared(Boolean subscriptionShared) {
+		this.subscriptionShared = subscriptionShared;
+	}
+
+	public boolean isSessionTransacted() {
+		return this.sessionTransacted;
+	}
+
+	public void setSessionTransacted(boolean sessionTransacted) {
+		this.sessionTransacted = sessionTransacted;
+	}
+}
--- a/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/AbstractJmsSupplierTests.java
+++ b/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/AbstractJmsSupplierTests.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
+import org.springframework.test.annotation.DirtiesContext;
+
+@SpringBootTest
+@DirtiesContext
+public class AbstractJmsSupplierTests {
+
+	@Autowired
+	protected JmsMessageDrivenEndpoint endpoint;
+
+	@SpringBootApplication
+	public static class JmsSupplierTestApplication {
+
+	}
+}
--- a/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated1Tests.java
+++ b/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated1Tests.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import javax.jms.Session;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.SimpleMessageListenerContainer;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@TestPropertySource(properties = {
+		"jms.supplier.sessionTransacted = false",
+		"jms.supplier.destination = topic",
+		"jms.supplier.messageSelector = JMSCorrelationId=foo",
+		"jms.supplier.subscriptionDurable = false",
+		"jms.supplier.subscriptionShared = false",
+		"spring.jms.listener.acknowledgeMode = DUPS_OK",
+		"spring.jms.listener.concurrency = 3",
+		"spring.jms.listener.maxConcurrency = 4",
+		"spring.jms.pubSubDomain = true"
+})
+public class PropertiesPopulated1Tests extends AbstractJmsSupplierTests {
+
+	@Test
+	public void test() {
+		AbstractMessageListenerContainer container = TestUtils.getPropertyValue(this.endpoint, "listenerContainer",
+				AbstractMessageListenerContainer.class);
+		assertThat(container).isInstanceOf(SimpleMessageListenerContainer.class);
+		assertThat(TestUtils.getPropertyValue(container, "sessionAcknowledgeMode")).isEqualTo(Session.DUPS_OK_ACKNOWLEDGE);
+		assertThat(TestUtils.getPropertyValue(container, "sessionTransacted", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "clientId")).isNull();
+		assertThat(TestUtils.getPropertyValue(container, "destination")).isEqualTo("topic");
+		assertThat(TestUtils.getPropertyValue(container, "messageSelector")).isEqualTo("JMSCorrelationId=foo");
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionDurable", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionShared", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(3);
+		assertThat(TestUtils.getPropertyValue(container, "pubSubDomain", Boolean.class)).isTrue();
+	}
+}
--- a/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated2Tests.java
+++ b/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated2Tests.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import javax.jms.Session;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@TestPropertySource(properties = {
+		"jms.supplier.sessionTransacted = true", "jms.supplier.clientId = client", "jms.supplier.destination = topic",
+		"jms.supplier.subscriptionName = subName", "jms.supplier.subscriptionDurable = true",
+		"jms.supplier.subscriptionShared = false", "spring.jms.listener.acknowledgeMode = AUTO",
+		"spring.jms.listener.concurrency = 3",
+		"spring.jms.listener.maxConcurrency = 4"
+})
+public class PropertiesPopulated2Tests extends AbstractJmsSupplierTests {
+
+	@Test
+	public void test() {
+
+		AbstractMessageListenerContainer container = TestUtils.getPropertyValue(this.endpoint, "listenerContainer",
+				AbstractMessageListenerContainer.class);
+		assertThat(container).isInstanceOf(DefaultMessageListenerContainer.class);
+
+		assertThat(TestUtils.getPropertyValue(container, "sessionAcknowledgeMode")).isEqualTo(Session.AUTO_ACKNOWLEDGE);
+		assertThat(TestUtils.getPropertyValue(container, "sessionTransacted", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(container, "clientId")).isEqualTo("client");
+		assertThat(TestUtils.getPropertyValue(container, "destination")).isEqualTo("topic");
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionDurable", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionName")).isEqualTo("subName");
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionShared", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(3);
+		assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isEqualTo(4);
+		assertThat(TestUtils.getPropertyValue(container, "pubSubDomain", Boolean.class)).isTrue();
+	}
+}
--- a/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated3Tests.java
+++ b/supplier/jms-supplier/src/test/java/org/springframework/cloud/fn/supplier/jms/PropertiesPopulated3Tests.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2016-2020 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.cloud.fn.supplier.jms;
+
+import java.util.function.Supplier;
+
+import javax.jms.Session;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.messaging.Message;
+import org.springframework.test.context.TestPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@TestPropertySource(properties = {
+		"jms.supplier.sessionTransacted = true", "jms.supplier.destination = jmssource.test.queue",
+		"jms.supplier.messageSelector = JMSCorrelationId=foo",
+		"jms.supplier.subscriptionDurable = false", "jms.supplier.subscriptionShared = false",
+		"spring.jms.listener.acknowledgeMode = AUTO",
+		"spring.jms.listener.concurrency = 3",
+		"spring.jms.listener.maxConcurrency = 4",
+		"spring.jms.pubSubDomain = false" })
+public class PropertiesPopulated3Tests extends AbstractJmsSupplierTests {
+
+	@Autowired
+	private JmsTemplate template;
+
+	@Autowired
+	private Supplier<Flux<Message<?>>> jmsSupplier;
+
+	@Test
+	public void test() throws Exception {
+		AbstractMessageListenerContainer container = TestUtils.getPropertyValue(this.endpoint, "listenerContainer",
+				AbstractMessageListenerContainer.class);
+		assertThat(container).isInstanceOf(DefaultMessageListenerContainer.class);
+		assertThat(TestUtils.getPropertyValue(container, "sessionAcknowledgeMode")).isEqualTo(Session.AUTO_ACKNOWLEDGE);
+		assertThat(TestUtils.getPropertyValue(container, "sessionTransacted", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(container, "destination")).isEqualTo("jmssource.test.queue");
+		assertThat(TestUtils.getPropertyValue(container, "messageSelector")).isEqualTo("JMSCorrelationId=foo");
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionDurable", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "subscriptionShared", Boolean.class)).isFalse();
+		assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(3);
+		assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isEqualTo(4);
+		assertThat(TestUtils.getPropertyValue(container, "pubSubDomain", Boolean.class)).isFalse();
+
+		final Flux<Message<?>> messageFlux = jmsSupplier.get();
+
+		final StepVerifier stepVerifier = StepVerifier.create(messageFlux)
+				.assertNext((message) -> {
+							assertThat(message.getPayload())
+									.isEqualTo("Hello, world!");
+						}
+				)
+				.thenCancel()
+				.verifyLater();
+
+		template.convertAndSend("jmssource.test.queue", "Hello, world!");
+
+		stepVerifier.verify();
+	}
+}