diff --git a/pom.xml b/pom.xml index a81044f6..df5aacfd 100644 --- a/pom.xml +++ b/pom.xml @@ -77,6 +77,7 @@ supplier/geode-supplier supplier/http-supplier supplier/jdbc-supplier + supplier/jms-supplier supplier/mongodb-supplier supplier/mqtt-supplier supplier/tcp-supplier diff --git a/supplier/jms-supplier/README.adoc b/supplier/jms-supplier/README.adoc new file mode 100644 index 00000000..edfc6103 --- /dev/null +++ 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>>`. +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>>`. + +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. \ No newline at end of file diff --git a/supplier/jms-supplier/pom.xml b/supplier/jms-supplier/pom.xml new file mode 100644 index 00000000..b3eeda14 --- /dev/null +++ b/supplier/jms-supplier/pom.xml @@ -0,0 +1,60 @@ + + + 4.0.0 + jms-supplier + 1.0.0-SNAPSHOT + jms-supplier + jms supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.integration + spring-integration-jms + + + javax.jms + javax.jms-api + provided + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.apache.activemq + activemq-broker + test + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.integration + spring-integration-test + test + + + diff --git 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 new file mode 100644 index 00000000..0115814c --- /dev/null +++ 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>> 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; + } +} diff --git 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 new file mode 100644 index 00000000..b175b1d6 --- /dev/null +++ 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; + } +} diff --git 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 new file mode 100644 index 00000000..81d8675d --- /dev/null +++ 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 { + + } +} diff --git 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 new file mode 100644 index 00000000..e098e76b --- /dev/null +++ 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(); + } +} diff --git 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 new file mode 100644 index 00000000..6a9ff37c --- /dev/null +++ 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(); + } +} diff --git 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 new file mode 100644 index 00000000..e50811a8 --- /dev/null +++ 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>> 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> 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(); + } +}