diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java new file mode 100644 index 0000000000..17e9cb70e5 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -0,0 +1,136 @@ +/* + * Copyright 2016 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.dsl; + +import java.util.Queue; +import java.util.concurrent.Executor; + +import org.springframework.integration.dsl.channel.DirectChannelSpec; +import org.springframework.integration.dsl.channel.ExecutorChannelSpec; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.channel.PriorityChannelSpec; +import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec; +import org.springframework.integration.dsl.channel.QueueChannelSpec; +import org.springframework.integration.dsl.channel.RendezvousChannelSpec; +import org.springframework.integration.store.ChannelMessageStore; +import org.springframework.integration.store.PriorityCapableChannelMessageStore; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + * @author Gary Russell + * + * @since 5.0 + */ +public class Channels { + + public DirectChannelSpec direct() { + return MessageChannels.direct(); + } + + public DirectChannelSpec direct(String id) { + return MessageChannels.direct(id); + } + + public QueueChannelSpec queue() { + return MessageChannels.queue(); + } + + public QueueChannelSpec queue(String id) { + return MessageChannels.queue(id); + } + + public QueueChannelSpec queue(Integer capacity) { + return MessageChannels.queue(capacity); + } + + public QueueChannelSpec queue(String id, Integer capacity) { + return MessageChannels.queue(id, capacity); + } + + public QueueChannelSpec queue(Queue> queue) { + return MessageChannels.queue(queue); + } + + public QueueChannelSpec queue(String id, Queue> queue) { + return MessageChannels.queue(id, queue); + } + + public QueueChannelSpec.MessageStoreSpec queue(ChannelMessageStore messageGroupStore, Object groupId) { + return MessageChannels.queue(messageGroupStore, groupId); + } + + public QueueChannelSpec.MessageStoreSpec queue(String id, ChannelMessageStore messageGroupStore, Object groupId) { + return MessageChannels.queue(id, messageGroupStore, groupId); + } + + public PriorityChannelSpec priority() { + return MessageChannels.priority(); + } + + public PriorityChannelSpec priority(String id) { + return MessageChannels.priority(id); + } + + public QueueChannelSpec.MessageStoreSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore, + Object groupId) { + return MessageChannels.priority(id, messageGroupStore, groupId); + } + + public QueueChannelSpec.MessageStoreSpec priority(PriorityCapableChannelMessageStore messageGroupStore, + Object groupId) { + return MessageChannels.priority(messageGroupStore, groupId); + } + + public RendezvousChannelSpec rendezvous() { + return MessageChannels.rendezvous(); + } + + public RendezvousChannelSpec rendezvous(String id) { + return MessageChannels.rendezvous(id); + } + + public PublishSubscribeChannelSpec> publishSubscribe() { + return MessageChannels.publishSubscribe(); + } + + public PublishSubscribeChannelSpec> publishSubscribe(Executor executor) { + return MessageChannels.publishSubscribe(executor); + } + + public PublishSubscribeChannelSpec> publishSubscribe(String id, + Executor executor) { + return MessageChannels.publishSubscribe(id, executor); + } + + public PublishSubscribeChannelSpec> publishSubscribe(String id) { + return MessageChannels.publishSubscribe(id); + } + + public ExecutorChannelSpec executor(Executor executor) { + return MessageChannels.executor(executor); + } + + public ExecutorChannelSpec executor(String id, Executor executor) { + return MessageChannels.executor(id, executor); + } + + Channels() { + super(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 5f6700a584..f57450e21f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -212,6 +212,18 @@ public abstract class IntegrationFlowDefinition> channels) { + Assert.notNull(channels); + return channel(channels.apply(new Channels())); + } + /** * The {@link org.springframework.integration.channel.PublishSubscribeChannel} {@link #channel} * method specific implementation to allow the use of the 'subflow' subscriber capability. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index ffaded3166..4a3d1307fd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -17,6 +17,7 @@ package org.springframework.integration.dsl; import java.util.function.Consumer; +import java.util.function.Function; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.MessageSource; @@ -74,6 +75,20 @@ public final class IntegrationFlows { : from(messageChannelName); } + /** + * Populate the {@link MessageChannel} object to the + * {@link IntegrationFlowBuilder} chain using the fluent API from {@link Channels} factory. + * The {@link org.springframework.integration.dsl.IntegrationFlow} {@code inputChannel}. + * @param channels the {@link Function} to use method chain to configure. + * {@link MessageChannel} via {@link Channels} factory. + * @return new {@link IntegrationFlowBuilder}. + * @see Channels + */ + public static IntegrationFlowBuilder from(Function> channels) { + Assert.notNull(channels); + return from(channels.apply(new Channels())); + } + /** * Populate the {@link MessageChannel} object to the * {@link IntegrationFlowBuilder} chain using the fluent API from {@link MessageChannelSpec}. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 3b10bd4b27..1def736cba 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -242,7 +242,7 @@ public class CorrelationHandlerTests { @Bean @DependsOn("barrierFlow") public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) { - return IntegrationFlows.from(MessageChannels.queue("releaseChannel")) + return IntegrationFlows.from(c -> c.queue("releaseChannel")) .trigger(barrierTriggerAction, e -> e.poller(p -> p.fixedDelay(100))) .get(); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java new file mode 100644 index 0000000000..55d4bbc351 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java @@ -0,0 +1,104 @@ +/* + * Copyright 2016 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.jms; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.messaging.MessageChannel; + +/** + * A wrapper around the {@link JmsMessageDrivenEndpoint} implementing + * {@link MessagingGatewaySupport}. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 5.0 + */ +public class JmsInboundGateway extends MessagingGatewaySupport implements DisposableBean, OrderlyShutdownCapable { + + private final JmsMessageDrivenEndpoint endpoint; + + public JmsInboundGateway(AbstractMessageListenerContainer listenerContainer, + ChannelPublishingJmsMessageListener listener) { + this.endpoint = new JmsMessageDrivenEndpoint(listenerContainer, listener); + } + + @Override + public void setRequestChannel(MessageChannel requestChannel) { + super.setRequestChannel(requestChannel); + this.endpoint.getListener().setRequestChannel(requestChannel); + } + + @Override + public String getComponentType() { + return this.endpoint.getComponentType(); + } + + @Override + public void setComponentName(String componentName) { + super.setComponentName(componentName); + this.endpoint.setComponentName(getComponentName()); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + super.setApplicationContext(applicationContext); + this.endpoint.setApplicationContext(applicationContext); + this.endpoint.setBeanFactory(applicationContext); + this.endpoint.getListener().setBeanFactory(applicationContext); + } + + @Override + protected void onInit() throws Exception { + this.endpoint.afterPropertiesSet(); + } + + public ChannelPublishingJmsMessageListener getListener() { + return this.endpoint.getListener(); + } + + @Override + protected void doStart() { + this.endpoint.start(); + } + + @Override + protected void doStop() { + this.endpoint.stop(); + } + + @Override + public void destroy() throws Exception { + this.endpoint.destroy(); + } + + @Override + public int beforeShutdown() { + return this.endpoint.beforeShutdown(); + } + + @Override + public int afterShutdown() { + return this.endpoint.afterShutdown(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index 06e8bf5bf5..69a3e28713 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -16,12 +16,15 @@ package org.springframework.integration.jms; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.ApplicationContext; import org.springframework.integration.context.OrderlyShutdownCapable; -import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.jms.util.JmsAdapterUtils; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; /** @@ -33,7 +36,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan */ -public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements DisposableBean, OrderlyShutdownCapable { +public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements DisposableBean, OrderlyShutdownCapable { private final AbstractMessageListenerContainer listenerContainer; @@ -89,13 +92,64 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos this.sessionAcknowledgeMode = sessionAcknowledgeMode; } + @Override + public void setOutputChannel(MessageChannel outputChannel) { + super.setOutputChannel(outputChannel); + this.listener.setRequestChannel(outputChannel); + } + + @Override + public void setOutputChannelName(String outputChannelName) { + super.setOutputChannelName(outputChannelName); + this.listener.setRequestChannelName(outputChannelName); + } + + @Override + public void setErrorChannel(MessageChannel errorChannel) { + super.setErrorChannel(errorChannel); + this.listener.setErrorChannel(errorChannel); + } + + @Override + public void setErrorChannelName(String errorChannelName) { + super.setErrorChannelName(errorChannelName); + this.listener.setErrorChannelName(errorChannelName); + } + + @Override + public void setSendTimeout(long sendTimeout) { + super.setSendTimeout(sendTimeout); + this.listener.setRequestTimeout(sendTimeout); + } + + @Override + public void setShouldTrack(boolean shouldTrack) { + super.setShouldTrack(shouldTrack); + this.listener.setShouldTrack(shouldTrack); + } + + public ChannelPublishingJmsMessageListener getListener() { + return this.listener; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + super.setApplicationContext(applicationContext); + this.listener.setBeanFactory(applicationContext); + } + @Override public String getComponentType() { return "jms:message-driven-channel-adapter"; } @Override - protected void onInit() throws Exception { + public void afterSingletonsInstantiated() { + // skip the output channel requirement assertion for when the listener is pre-built + } + + @Override + protected void onInit() { this.listener.afterPropertiesSet(); if (!this.listenerContainer.isActive()) { this.listenerContainer.afterPropertiesSet(); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java new file mode 100644 index 0000000000..49b2ff0d82 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java @@ -0,0 +1,282 @@ +/* + * Copyright 2014-2016 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.jms.dsl; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; + +/** + * Factory class for JMS components. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 5.0 + */ +public final class Jms { + + /** + * The factory to produce a {@link JmsPollableMessageChannelSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param the {@link JmsPollableMessageChannelSpec} inheritor type + * @return the {@link JmsPollableMessageChannelSpec} instance + */ + public static > JmsPollableMessageChannelSpec pollableChannel( + ConnectionFactory connectionFactory) { + return pollableChannel(null, connectionFactory); + } + + /** + * The factory to produce a {@link JmsPollableMessageChannelSpec}. + * @param id the bean name for the target {@code PollableChannel} component + * @param connectionFactory the JMS ConnectionFactory to build on + * @param the {@link JmsPollableMessageChannelSpec} inheritor type + * @return the {@link JmsPollableMessageChannelSpec} instance + */ + public static > JmsPollableMessageChannelSpec pollableChannel( + String id, ConnectionFactory connectionFactory) { + return new JmsPollableMessageChannelSpec(connectionFactory).id(id); + } + + /** + * The factory to produce a {@link JmsMessageChannelSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param the {@link JmsMessageChannelSpec} inheritor type + * @return the {@link JmsMessageChannelSpec} instance + */ + public static > JmsMessageChannelSpec channel( + ConnectionFactory connectionFactory) { + return channel(null, connectionFactory); + } + + /** + * The factory to produce a {@link JmsMessageChannelSpec}. + * @param id the bean name for the target {@code MessageChannel} component + * @param connectionFactory the JMS ConnectionFactory to build on + * @param the {@link JmsMessageChannelSpec} inheritor type + * @return the {@link JmsMessageChannelSpec} instance + */ + public static > JmsMessageChannelSpec channel(String id, + ConnectionFactory connectionFactory) { + return new JmsMessageChannelSpec(connectionFactory).id(id); + } + + /** + * The factory to produce a {@link JmsPublishSubscribeMessageChannelSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsPublishSubscribeMessageChannelSpec} instance + */ + public static JmsPublishSubscribeMessageChannelSpec publishSubscribeChannel(ConnectionFactory connectionFactory) { + return publishSubscribeChannel(null, connectionFactory); + } + + /** + * The factory to produce a {@link JmsPublishSubscribeMessageChannelSpec}. + * @param id the bean name for the target {@code MessageChannel} component + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsPublishSubscribeMessageChannelSpec} instance + */ + public static JmsPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id, + ConnectionFactory connectionFactory) { + return new JmsPublishSubscribeMessageChannelSpec(connectionFactory).id(id); + } + + /** + * The factory to produce a {@link JmsOutboundChannelAdapterSpec}. + * @param jmsTemplate the JmsTemplate to build on + * @param the {@link JmsOutboundChannelAdapterSpec} inheritor type + * @return the {@link JmsOutboundChannelAdapterSpec} instance + */ + public static > JmsOutboundChannelAdapterSpec outboundAdapter( + JmsTemplate jmsTemplate) { + return new JmsOutboundChannelAdapterSpec(jmsTemplate); + } + + /** + * The factory to produce a {@link JmsOutboundChannelAdapterSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsOutboundChannelAdapterSpec} instance + */ + public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter( + ConnectionFactory connectionFactory) { + return new JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware(connectionFactory); + } + + /** + * The factory to produce a {@link JmsInboundChannelAdapterSpec}. + * @param jmsTemplate the JmsTemplate to build on + * @param the {@link JmsInboundChannelAdapterSpec} inheritor type + * @return the {@link JmsInboundChannelAdapterSpec} instance + */ + public static > JmsInboundChannelAdapterSpec inboundAdapter( + JmsTemplate jmsTemplate) { + return new JmsInboundChannelAdapterSpec(jmsTemplate); + } + + /** + * The factory to produce a {@link JmsInboundChannelAdapterSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsInboundChannelAdapterSpec} instance + */ + public static JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware inboundAdapter( + ConnectionFactory connectionFactory) { + return new JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware(connectionFactory); + } + + /** + * The factory to produce a {@link JmsOutboundGatewaySpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsOutboundGatewaySpec} instance + */ + public static JmsOutboundGatewaySpec outboundGateway(ConnectionFactory connectionFactory) { + return new JmsOutboundGatewaySpec(connectionFactory); + } + + /** + * The factory to produce a {@link JmsOutboundGatewaySpec}. + * @param listenerContainer the JMS {@link AbstractMessageListenerContainer} to build on + * @param the {@link JmsInboundGatewaySpec} inheritor type + * @return the {@link JmsOutboundGatewaySpec} instance + */ + public static > JmsInboundGatewaySpec inboundGateway( + AbstractMessageListenerContainer listenerContainer) { + return new JmsInboundGatewaySpec(listenerContainer); + } + + /** + * The factory to produce a {@link JmsOutboundGatewaySpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsOutboundGatewaySpec} instance + */ + public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec + inboundGateway(ConnectionFactory connectionFactory) { + return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class); + } + + /** + * The factory to produce a {@link JmsOutboundGatewaySpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param containerClass the {@link AbstractMessageListenerContainer} implementation class + * to instantiate listener container + * @param the {@link JmsListenerContainerSpec} inheritor type + * @param the {@link AbstractMessageListenerContainer} inheritor type + * @return the {@link JmsOutboundGatewaySpec} instance + */ + public static , C extends AbstractMessageListenerContainer> + JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec inboundGateway(ConnectionFactory connectionFactory, + Class containerClass) { + try { + JmsListenerContainerSpec spec = + new JmsListenerContainerSpec(containerClass) + .connectionFactory(connectionFactory); + return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec(spec); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + /** + * The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}. + * @param listenerContainer the {@link AbstractMessageListenerContainer} to build on + * @param the {@link JmsMessageDrivenChannelAdapterSpec} inheritor type + * @return the {@link JmsMessageDrivenChannelAdapterSpec} instance + */ + public static > + JmsMessageDrivenChannelAdapterSpec messageDrivenChannelAdapter( + AbstractMessageListenerContainer listenerContainer) { + return new JmsMessageDrivenChannelAdapterSpec(listenerContainer); + } + + /** + * The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @return the {@link JmsMessageDrivenChannelAdapterSpec} instance + */ + public static JmsMessageDrivenChannelAdapterSpec + .JmsMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConnectionFactory connectionFactory) { + return messageDrivenChannelAdapter(connectionFactory, DefaultMessageListenerContainer.class); + } + + /** + * The factory to produce a {@link JmsMessageDrivenChannelAdapterSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param containerClass the {@link AbstractMessageListenerContainer} implementation class + * to instantiate listener container + * @param the {@link JmsListenerContainerSpec} inheritor type + * @param the {@link AbstractMessageListenerContainer} inheritor type + * @return the {@link JmsMessageDrivenChannelAdapterSpec} instance + */ + public static , C extends AbstractMessageListenerContainer> + JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec + messageDrivenChannelAdapter(ConnectionFactory connectionFactory, Class containerClass) { + try { + JmsListenerContainerSpec spec = + new JmsListenerContainerSpec(containerClass) + .connectionFactory(connectionFactory); + return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec(spec); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + /** + * The factory to produce a {@link JmsListenerContainerSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param destination the {@link Destination} to listen to + * @return the {@link JmsListenerContainerSpec} instance + */ + public static JmsDefaultListenerContainerSpec container(ConnectionFactory connectionFactory, + Destination destination) { + try { + return new JmsDefaultListenerContainerSpec() + .connectionFactory(connectionFactory) + .destination(destination); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + /** + * The factory to produce a {@link JmsListenerContainerSpec}. + * @param connectionFactory the JMS ConnectionFactory to build on + * @param destinationName the destination name to listen to + * @return the {@link JmsListenerContainerSpec} instance + */ + public static JmsDefaultListenerContainerSpec container(ConnectionFactory connectionFactory, + String destinationName) { + try { + return new JmsDefaultListenerContainerSpec() + .connectionFactory(connectionFactory) + .destination(destinationName); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + private Jms() { + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDefaultListenerContainerSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDefaultListenerContainerSpec.java new file mode 100644 index 0000000000..309c5aa5b6 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDefaultListenerContainerSpec.java @@ -0,0 +1,205 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.concurrent.Executor; + +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.backoff.BackOff; + +/** + * A {@link DefaultMessageListenerContainer} specific {@link JmsListenerContainerSpec} extension. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsDefaultListenerContainerSpec + extends JmsListenerContainerSpec { + + JmsDefaultListenerContainerSpec() throws Exception { + super(DefaultMessageListenerContainer.class); + } + + /** + * Specify an {@link Executor}. + * @param taskExecutor the {@link Executor} to use. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setTaskExecutor(Executor) + */ + public JmsDefaultListenerContainerSpec taskExecutor(Executor taskExecutor) { + this.target.setTaskExecutor(taskExecutor); + return this; + } + + /** + * Specify a {@link BackOff}. + * @param backOff the {@link BackOff} to use. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setBackOff(BackOff) + */ + public JmsDefaultListenerContainerSpec backOff(BackOff backOff) { + this.target.setBackOff(backOff); + return this; + } + + /** + * Specify a recovery interval. + * @param recoveryInterval the recovery interval to use. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setRecoveryInterval(long) + */ + public JmsDefaultListenerContainerSpec recoveryInterval(long recoveryInterval) { + this.target.setRecoveryInterval(recoveryInterval); + return this; + } + + /** + * Specify the level of caching that this listener container is allowed to apply, + * in the form of the name of the corresponding constant: e.g. "CACHE_CONNECTION". + * @param constantName the cache level constant name. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see #cacheLevel(int) + * @see DefaultMessageListenerContainer#setCacheLevelName(String) + */ + public JmsDefaultListenerContainerSpec cacheLevelName(String constantName) { + this.target.setCacheLevelName(constantName); + return this; + } + + /** + * Specify the level of caching that this listener container is allowed to apply. + * @param cacheLevel the level of caching. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setCacheLevel(int) + */ + public JmsDefaultListenerContainerSpec cacheLevel(int cacheLevel) { + this.target.setCacheLevel(cacheLevel); + return this; + } + + /** + * The concurrency to use. + * @param concurrency the concurrency. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setConcurrency(String) + */ + public JmsDefaultListenerContainerSpec concurrency(String concurrency) { + this.target.setConcurrency(concurrency); + return this; + } + + /** + * The concurrent consumers number to use. + * @param concurrentConsumers the concurrent consumers count. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setConcurrentConsumers(int) + */ + public JmsDefaultListenerContainerSpec concurrentConsumers(int concurrentConsumers) { + this.target.setConcurrentConsumers(concurrentConsumers); + return this; + } + + /** + * The max for concurrent consumers number to use. + * @param maxConcurrentConsumers the max concurrent consumers count. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setMaxConcurrentConsumers(int) + */ + public JmsDefaultListenerContainerSpec maxConcurrentConsumers(int maxConcurrentConsumers) { + this.target.setMaxConcurrentConsumers(maxConcurrentConsumers); + return this; + } + + /** + * The max messages per task. + * @param maxMessagesPerTask the max messages per task. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setMaxMessagesPerTask(int) + */ + public JmsDefaultListenerContainerSpec maxMessagesPerTask(int maxMessagesPerTask) { + this.target.setMaxMessagesPerTask(maxMessagesPerTask); + return this; + } + + /** + * The max for concurrent consumers number to use. + * @param idleConsumerLimit the limit for idle consumer. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setMaxConcurrentConsumers(int) + */ + public JmsDefaultListenerContainerSpec idleConsumerLimit(int idleConsumerLimit) { + this.target.setIdleConsumerLimit(idleConsumerLimit); + return this; + } + + /** + * The the limit for idle task. + * @param idleTaskExecutionLimit the limit for idle task. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setIdleTaskExecutionLimit(int) + */ + public JmsDefaultListenerContainerSpec idleTaskExecutionLimit(int idleTaskExecutionLimit) { + this.target.setIdleTaskExecutionLimit(idleTaskExecutionLimit); + return this; + } + + /** + * A {@link PlatformTransactionManager} reference. + * @param transactionManager the {@link PlatformTransactionManager} to use. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setTransactionManager(PlatformTransactionManager) + */ + public JmsDefaultListenerContainerSpec transactionManager(PlatformTransactionManager transactionManager) { + this.target.setTransactionManager(transactionManager); + return this; + } + + /** + * A name for transaction. + * @param transactionName the name for transaction. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setTransactionName(String) + */ + public JmsDefaultListenerContainerSpec transactionName(String transactionName) { + this.target.setTransactionName(transactionName); + return this; + } + + /** + * A transaction timeout. + * @param transactionTimeout the transaction timeout. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setTransactionTimeout(int) + */ + public JmsDefaultListenerContainerSpec transactionTimeout(int transactionTimeout) { + this.target.setTransactionTimeout(transactionTimeout); + return this; + } + + /** + * A receive timeout. + * @param receiveTimeout the receive timeout. + * @return current {@link JmsDefaultListenerContainerSpec}. + * @see DefaultMessageListenerContainer#setReceiveTimeout(long) + */ + public JmsDefaultListenerContainerSpec receiveTimeout(long receiveTimeout) { + this.target.setReceiveTimeout(receiveTimeout); + return this; + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDestinationAccessorSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDestinationAccessorSpec.java new file mode 100644 index 0000000000..ed3e7c98a1 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsDestinationAccessorSpec.java @@ -0,0 +1,103 @@ +/* + * Copyright 2016 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.jms.dsl; + +import javax.jms.ConnectionFactory; + +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.jms.support.destination.JmsDestinationAccessor; + +/** + * A base {@link IntegrationComponentSpec} for {@link JmsDestinationAccessor}s. + * + * @param the target {@link JmsDestinationAccessorSpec} implementation type. + * @param the target {@link JmsDestinationAccessor} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public abstract class + JmsDestinationAccessorSpec, A extends JmsDestinationAccessor> + extends IntegrationComponentSpec { + + protected JmsDestinationAccessorSpec(A accessor) { + this.target = accessor; + } + + S connectionFactory(ConnectionFactory connectionFactory) { + this.target.setConnectionFactory(connectionFactory); + return _this(); + } + + /** + * A {@link DestinationResolver} to use. + * @param destinationResolver the {@link DestinationResolver} to use. + * @return the spec + * @see JmsDestinationAccessor#setDestinationResolver + */ + public S destinationResolver(DestinationResolver destinationResolver) { + this.target.setDestinationResolver(destinationResolver); + return _this(); + } + + /** + * A {@code pubSubDomain} flag. + * @param pubSubDomain the {@code pubSubDomain} flag. + * @return the spec + * @see JmsDestinationAccessor#setPubSubDomain(boolean) + */ + public S pubSubDomain(boolean pubSubDomain) { + target.setPubSubDomain(pubSubDomain); + return _this(); + } + + /** + * A session acknowledgement mode. + * @param sessionAcknowledgeMode the acknowledgement mode constant + * @return the spec + * @see javax.jms.Session#AUTO_ACKNOWLEDGE etc. + * @see JmsDestinationAccessor#setSessionAcknowledgeMode + */ + public S sessionAcknowledgeMode(int sessionAcknowledgeMode) { + this.target.setSessionAcknowledgeMode(sessionAcknowledgeMode); + return _this(); + } + + /** + * A session acknowledgement mode name. + * @param constantName the name of the {@link javax.jms.Session} acknowledge mode constant. + * @return the spec. + * @see JmsDestinationAccessor#setSessionAcknowledgeModeName + */ + public S sessionAcknowledgeModeName(String constantName) { + target.setSessionAcknowledgeModeName(constantName); + return _this(); + } + + /** + * A session transaction mode. + * @param sessionTransacted the transaction mode. + * @return the spec. + * @see JmsDestinationAccessor#setSessionTransacted + */ + public S sessionTransacted(boolean sessionTransacted) { + this.target.setSessionTransacted(sessionTransacted); + return _this(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..9963669651 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundChannelAdapterSpec.java @@ -0,0 +1,116 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.function.Consumer; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.integration.dsl.MessageSourceSpec; +import org.springframework.integration.jms.JmsDestinationPollingSource; +import org.springframework.integration.jms.JmsHeaderMapper; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.util.Assert; + +/** + * A {@link MessageSourceSpec} for a {@link JmsDestinationPollingSource}. + * + * @param the target {@link JmsInboundChannelAdapterSpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsInboundChannelAdapterSpec> + extends MessageSourceSpec { + + protected final JmsTemplateSpec jmsTemplateSpec = new JmsTemplateSpec(); + + JmsInboundChannelAdapterSpec(JmsTemplate jmsTemplate) { + this.target = new JmsDestinationPollingSource(jmsTemplate); + } + + private JmsInboundChannelAdapterSpec(ConnectionFactory connectionFactory) { + this.target = new JmsDestinationPollingSource(this.jmsTemplateSpec.connectionFactory(connectionFactory).get()); + } + + /** + * @param messageSelector the messageSelector. + * @return the spec. + * @see JmsDestinationPollingSource#setMessageSelector(String) + */ + public S messageSelector(String messageSelector) { + this.target.setMessageSelector(messageSelector); + return _this(); + } + + /** + * Configure a {@link JmsHeaderMapper} to map from JMS headers and properties to + * Spring Integration headers. + * @param headerMapper the headerMapper. + * @return the spec. + */ + public S headerMapper(JmsHeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + return _this(); + } + + /** + * Configure the destination from which to receive messages. + * @param destination the destination. + * @return the spec. + */ + public S destination(Destination destination) { + this.target.setDestination(destination); + return _this(); + } + + /** + * Configure the name of destination from which to receive messages. + * @param destination the destination. + * @return the spec. + */ + public S destination(String destination) { + this.target.setDestinationName(destination); + return _this(); + } + + /** + * A {@link JmsTemplate}-based {@link JmsInboundChannelAdapterSpec} extension. + */ + public static class JmsInboundChannelSpecTemplateAware extends + JmsInboundChannelAdapterSpec { + + JmsInboundChannelSpecTemplateAware(ConnectionFactory connectionFactory) { + super(connectionFactory); + } + + /** + * Configure the channel adapter to use the template specification created by invoking the + * {@link Consumer} callback, passing in a {@link JmsTemplateSpec}. + * @param configurer the configurer. + * @return the spec. + */ + public JmsInboundChannelSpecTemplateAware configureJmsTemplate(Consumer configurer) { + Assert.notNull(configurer); + configurer.accept(this.jmsTemplateSpec); + return _this(); + } + + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java new file mode 100644 index 0000000000..a5cdf0c545 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java @@ -0,0 +1,224 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.function.Consumer; + +import javax.jms.Destination; + +import org.springframework.integration.dsl.MessagingGatewaySpec; +import org.springframework.integration.jms.ChannelPublishingJmsMessageListener; +import org.springframework.integration.jms.JmsHeaderMapper; +import org.springframework.integration.jms.JmsInboundGateway; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.util.Assert; + +/** + * A {@link MessagingGatewaySpec} for a {@link JmsInboundGateway}. + * + * @param the target {@link JmsInboundGatewaySpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsInboundGatewaySpec> + extends MessagingGatewaySpec { + + JmsInboundGatewaySpec(AbstractMessageListenerContainer listenerContainer) { + super(new JmsInboundGateway(listenerContainer, new ChannelPublishingJmsMessageListener())); + this.target.getListener().setExpectReply(true); + } + + /** + * @param defaultReplyDestination the defaultReplyDestination + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setDefaultReplyDestination(Destination) + */ + public S defaultReplyDestination(Destination defaultReplyDestination) { + this.target.getListener().setDefaultReplyDestination(defaultReplyDestination); + return _this(); + } + + /** + * @param destinationName the destinationName + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setDefaultReplyQueueName(String) + */ + public S defaultReplyQueueName(String destinationName) { + this.target.getListener().setDefaultReplyQueueName(destinationName); + return _this(); + } + + /** + * @param destinationName the destinationName + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setDefaultReplyTopicName(String) + */ + public S defaultReplyTopicName(String destinationName) { + this.target.getListener().setDefaultReplyTopicName(destinationName); + return _this(); + } + + /** + * @param replyTimeToLive the replyTimeToLive + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setReplyTimeToLive(long) + */ + public S replyTimeToLive(long replyTimeToLive) { + this.target.getListener().setReplyTimeToLive(replyTimeToLive); + return _this(); + } + + /** + * @param replyPriority the replyPriority + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setReplyPriority(int) + */ + public S replyPriority(int replyPriority) { + this.target.getListener().setReplyPriority(replyPriority); + return _this(); + } + + /** + * @param replyDeliveryPersistent the replyDeliveryPersistent + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setReplyDeliveryPersistent(boolean) + */ + public S replyDeliveryPersistent(boolean replyDeliveryPersistent) { + this.target.getListener().setReplyDeliveryPersistent(replyDeliveryPersistent); + return _this(); + } + + /** + * @param correlationKey the correlationKey + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setCorrelationKey(String) + */ + public S correlationKey(String correlationKey) { + this.target.getListener().setCorrelationKey(correlationKey); + return _this(); + } + + /** + * @param explicitQosEnabledForReplies the explicitQosEnabledForReplies. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setExplicitQosEnabledForReplies(boolean) + */ + public S explicitQosEnabledForReplies(boolean explicitQosEnabledForReplies) { + this.target.getListener().setExplicitQosEnabledForReplies(explicitQosEnabledForReplies); + return _this(); + } + + /** + * @param destinationResolver the destinationResolver. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setDestinationResolver(DestinationResolver) + */ + public S destinationResolver(DestinationResolver destinationResolver) { + this.target.getListener().setDestinationResolver(destinationResolver); + return _this(); + } + + /** + * @param messageConverter the messageConverter. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setMessageConverter(MessageConverter) + */ + public S jmsMessageConverter(MessageConverter messageConverter) { + this.target.getListener().setMessageConverter(messageConverter); + return _this(); + } + + /** + * @param headerMapper the headerMapper. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setHeaderMapper(JmsHeaderMapper) + */ + public S setHeaderMapper(JmsHeaderMapper headerMapper) { + this.target.getListener().setHeaderMapper(headerMapper); + return _this(); + } + + /** + * @param extractRequestPayload the extractRequestPayload. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setExtractRequestPayload(boolean) + */ + public S extractRequestPayload(boolean extractRequestPayload) { + this.target.getListener().setExtractRequestPayload(extractRequestPayload); + return _this(); + } + + /** + * @param extractReplyPayload the extractReplyPayload. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setExtractReplyPayload(boolean) + */ + public S extractReplyPayload(boolean extractReplyPayload) { + this.target.getListener().setExtractReplyPayload(extractReplyPayload); + return _this(); + } + + /** + * An {@link AbstractMessageListenerContainer}-based {@link JmsInboundGatewaySpec} extension. + * + * @param the target {@link JmsListenerContainerSpec} implementation type. + * @param the target {@link AbstractMessageListenerContainer} implementation type. + */ + public static class JmsInboundGatewayListenerContainerSpec, C extends AbstractMessageListenerContainer> + extends JmsInboundGatewaySpec> { + + private final JmsListenerContainerSpec spec; + + JmsInboundGatewayListenerContainerSpec(JmsListenerContainerSpec spec) { + super(spec.get()); + this.spec = spec; + this.spec.get().setAutoStartup(false); + } + + /** + * @param destination the destination + * @return the spec. + * @see JmsListenerContainerSpec#destination(Destination) + */ + public JmsInboundGatewayListenerContainerSpec destination(Destination destination) { + this.spec.destination(destination); + return _this(); + } + + /** + * @param destinationName the destinationName + * @return the spec. + * @see JmsListenerContainerSpec#destination(String) + */ + public JmsInboundGatewayListenerContainerSpec destination(String destinationName) { + this.spec.destination(destinationName); + return _this(); + } + + public JmsInboundGatewayListenerContainerSpec configureListenerContainer( + Consumer> configurer) { + Assert.notNull(configurer); + configurer.accept(this.spec); + return _this(); + } + + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsListenerContainerSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsListenerContainerSpec.java new file mode 100644 index 0000000000..a28f76ece9 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsListenerContainerSpec.java @@ -0,0 +1,146 @@ +/* + * Copyright 2016 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.jms.dsl; + +import javax.jms.Destination; +import javax.jms.ExceptionListener; + +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.util.ErrorHandler; + +/** + * A {@link JmsDestinationAccessorSpec} for {@link JmsListenerContainerSpec}s. + * + * @param the target {@link JmsListenerContainerSpec} implementation type. + * @param the target {@link AbstractMessageListenerContainer} implementation type. + * + * @author Artem Bilan + * @author Gary Russell + * @since 5.0 + */ +public class JmsListenerContainerSpec, C extends AbstractMessageListenerContainer> + extends JmsDestinationAccessorSpec { + + JmsListenerContainerSpec(Class aClass) throws Exception { + super(aClass.newInstance()); + if (DefaultMessageListenerContainer.class.isAssignableFrom(aClass)) { + this.target.setSessionTransacted(true); + } + } + + /** + * @param destination the destination. + * @return the spec. + * @see AbstractMessageListenerContainer#setDestination(Destination) + */ + S destination(Destination destination) { + this.target.setDestination(destination); + return _this(); + } + + /** + * @param destinationName the destinationName. + * @return the spec. + * @see AbstractMessageListenerContainer#setDestinationName(String) + */ + S destination(String destinationName) { + this.target.setDestinationName(destinationName); + return _this(); + } + + /** + * @param messageSelector the messageSelector. + * @return the spec. + * @see AbstractMessageListenerContainer#setMessageSelector(String) + */ + public S messageSelector(String messageSelector) { + this.target.setMessageSelector(messageSelector); + return _this(); + } + + /** + * @param subscriptionDurable the subscriptionDurable. + * @return the spec. + * @see AbstractMessageListenerContainer#setSubscriptionDurable(boolean) + */ + public S subscriptionDurable(boolean subscriptionDurable) { + this.target.setSubscriptionDurable(subscriptionDurable); + return _this(); + } + + /** + * @param durableSubscriptionName the durableSubscriptionName. + * @return the spec. + * @see AbstractMessageListenerContainer#setDurableSubscriptionName(String) + */ + public S durableSubscriptionName(String durableSubscriptionName) { + this.target.setDurableSubscriptionName(durableSubscriptionName); + return _this(); + } + + /** + * @param exceptionListener the exceptionListener. + * @return the spec. + * @see AbstractMessageListenerContainer#setExceptionListener(ExceptionListener) + */ + public S exceptionListener(ExceptionListener exceptionListener) { + this.target.setExceptionListener(exceptionListener); + return _this(); + } + + /** + * @param errorHandler the errorHandler. + * @return the spec. + * @see AbstractMessageListenerContainer#setErrorHandler(ErrorHandler) + */ + public S errorHandler(ErrorHandler errorHandler) { + this.target.setErrorHandler(errorHandler); + return _this(); + } + + /** + * @param exposeListenerSession the exposeListenerSession. + * @return the spec. + * @see AbstractMessageListenerContainer#setExposeListenerSession(boolean) + */ + public S exposeListenerSession(boolean exposeListenerSession) { + this.target.setExposeListenerSession(exposeListenerSession); + return _this(); + } + + /** + * @param acceptMessagesWhileStopping the acceptMessagesWhileStopping. + * @return the spec. + * @see AbstractMessageListenerContainer#setAcceptMessagesWhileStopping(boolean) + */ + public S acceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + this.target.setAcceptMessagesWhileStopping(acceptMessagesWhileStopping); + return _this(); + } + + /** + * @param clientId the clientId. + * @return the spec. + * @see AbstractMessageListenerContainer#setClientId(String) + */ + public S clientId(String clientId) { + this.target.setClientId(clientId); + return _this(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageChannelSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageChannelSpec.java new file mode 100644 index 0000000000..6d9beef280 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageChannelSpec.java @@ -0,0 +1,231 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.concurrent.Executor; + +import javax.jms.ConnectionFactory; + +import org.springframework.integration.jms.AbstractJmsChannel; +import org.springframework.integration.jms.SubscribableJmsChannel; +import org.springframework.integration.jms.config.JmsChannelFactoryBean; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.ErrorHandler; + +/** + * A {@link JmsMessageChannelSpec} for subscribable {@link AbstractJmsChannel}s. + * + * @param the target {@link JmsMessageChannelSpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsMessageChannelSpec> extends JmsPollableMessageChannelSpec { + + JmsMessageChannelSpec(ConnectionFactory connectionFactory) { + super(new JmsChannelFactoryBean(true), connectionFactory); + } + + /** + * Configure the type of the container. + * {@link AbstractMessageListenerContainer}. Defaults to + * {@link DefaultMessageListenerContainer}. + * @param containerType the containerType. + * @return the current {@link JmsMessageChannelSpec}. + */ + public S containerType(Class containerType) { + this.jmsChannelFactoryBean.setContainerType(containerType); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer} + * or a {@link org.springframework.jms.listener.SimpleMessageListenerContainer}. + * @param concurrentConsumers the concurrentConsumers. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setConcurrentConsumers(int) + * @see org.springframework.jms.listener.SimpleMessageListenerContainer#setConcurrentConsumers(int) + */ + public S concurrentConsumers(int concurrentConsumers) { + this.jmsChannelFactoryBean.setConcurrentConsumers(concurrentConsumers); + return _this(); + } + + /** + * @param maxSubscribers the maxSubscribers. + * @return the current {@link JmsMessageChannelSpec}. + * @see SubscribableJmsChannel#setMaxSubscribers(int) + */ + public S maxSubscribers(int maxSubscribers) { + this.jmsChannelFactoryBean.setMaxSubscribers(maxSubscribers); + return _this(); + } + + /** + * @param autoStartup the autoStartup. + * @return the current {@link JmsMessageChannelSpec}. + * @see org.springframework.context.SmartLifecycle + */ + public S autoStartup(boolean autoStartup) { + this.jmsChannelFactoryBean.setAutoStartup(autoStartup); + return _this(); + } + + /** + * @param phase the phase. + * @return the current {@link JmsMessageChannelSpec}. + * @see org.springframework.context.SmartLifecycle + */ + public S phase(int phase) { + this.jmsChannelFactoryBean.setPhase(phase); + return _this(); + } + + /** + * @param errorHandler the errorHandler. + * @return the current {@link JmsMessageChannelSpec}. + * @see AbstractMessageListenerContainer#setErrorHandler(ErrorHandler) + */ + public S errorHandler(ErrorHandler errorHandler) { + this.jmsChannelFactoryBean.setErrorHandler(errorHandler); + return _this(); + } + + /** + * @param exposeListenerSession the exposeListenerSession. + * @return the current {@link JmsMessageChannelSpec}. + * @see AbstractMessageListenerContainer#setExposeListenerSession(boolean) + */ + public S exposeListenerSession(boolean exposeListenerSession) { + this.jmsChannelFactoryBean.setExposeListenerSession(exposeListenerSession); + return _this(); + } + + /** + * @param acceptMessagesWhileStopping the acceptMessagesWhileStopping. + * @return the current {@link JmsMessageChannelSpec}. + * @see AbstractMessageListenerContainer#setAcceptMessagesWhileStopping(boolean) + */ + public S acceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + this.jmsChannelFactoryBean.setAcceptMessagesWhileStopping(acceptMessagesWhileStopping); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param idleTaskExecutionLimit the idleTaskExecutionLimit. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setIdleTaskExecutionLimit(int) + */ + public S idleTaskExecutionLimit(int idleTaskExecutionLimit) { + this.jmsChannelFactoryBean.setIdleTaskExecutionLimit(idleTaskExecutionLimit); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param maxMessagesPerTask the maxMessagesPerTask. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setMaxMessagesPerTask(int) + */ + public S maxMessagesPerTask(int maxMessagesPerTask) { + this.jmsChannelFactoryBean.setMaxMessagesPerTask(maxMessagesPerTask); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param recoveryInterval the recoveryInterval. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setRecoveryInterval(long) + */ + public S recoveryInterval(long recoveryInterval) { + this.jmsChannelFactoryBean.setRecoveryInterval(recoveryInterval); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer} + * or a {@link org.springframework.jms.listener.SimpleMessageListenerContainer}. + * @param taskExecutor the taskExecutor. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setTaskExecutor(Executor) + * @see org.springframework.jms.listener.SimpleMessageListenerContainer#setTaskExecutor(Executor) + */ + public S taskExecutor(Executor taskExecutor) { + this.jmsChannelFactoryBean.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param transactionManager the transactionManager. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setTransactionManager(PlatformTransactionManager) + */ + public S transactionManager(PlatformTransactionManager transactionManager) { + this.jmsChannelFactoryBean.setTransactionManager(transactionManager); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param transactionName the transactionName. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setTransactionName(String) + */ + public S transactionName(String transactionName) { + this.jmsChannelFactoryBean.setTransactionName(transactionName); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer}. + * @param transactionTimeout the transactionTimeout. + * @return the current {@link JmsMessageChannelSpec}. + * @see DefaultMessageListenerContainer#setTransactionTimeout(int) + */ + public S transactionTimeout(int transactionTimeout) { + this.jmsChannelFactoryBean.setTransactionTimeout(transactionTimeout); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a + * {@link DefaultMessageListenerContainer}. + * @param cacheLevel the value for {@code DefaultMessageListenerContainer.cacheLevel} + * @return the current {@link JmsMessageChannelSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setCacheLevel(int) + */ + public S cacheLevel(Integer cacheLevel) { + this.jmsChannelFactoryBean.setCacheLevel(cacheLevel); + return _this(); + } + + /** + * @param subscriptionShared the subscription shared {@code boolean} flag. + * @return the current {@link JmsMessageChannelSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setSubscriptionShared + */ + public S subscriptionShared(boolean subscriptionShared) { + this.jmsChannelFactoryBean.setSubscriptionShared(subscriptionShared); + return _this(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java new file mode 100644 index 0000000000..06f0e29e2c --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java @@ -0,0 +1,130 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.function.Consumer; + +import javax.jms.Destination; + +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.jms.ChannelPublishingJmsMessageListener; +import org.springframework.integration.jms.JmsHeaderMapper; +import org.springframework.integration.jms.JmsMessageDrivenEndpoint; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.util.Assert; + +/** + * A {@link MessageProducerSpec} for {@link JmsMessageDrivenEndpoint}s. + * + * @param the target {@link JmsMessageDrivenChannelAdapterSpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsMessageDrivenChannelAdapterSpec> + extends MessageProducerSpec { + + JmsMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer listenerContainer) { + super(new JmsMessageDrivenEndpoint(listenerContainer, new ChannelPublishingJmsMessageListener())); + this.target.getListener().setExpectReply(false); + } + + /** + * @param messageConverter the messageConverter. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setMessageConverter(MessageConverter) + */ + public S jmsMessageConverter(MessageConverter messageConverter) { + this.target.getListener().setMessageConverter(messageConverter); + return _this(); + } + + /** + * @param headerMapper the headerMapper. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setHeaderMapper(JmsHeaderMapper) + */ + public S headerMapper(JmsHeaderMapper headerMapper) { + this.target.getListener().setHeaderMapper(headerMapper); + return _this(); + } + + /** + * @param extractRequestPayload the extractRequestPayload. + * @return the spec. + * @see ChannelPublishingJmsMessageListener#setExtractRequestPayload(boolean) + */ + public S extractPayload(boolean extractRequestPayload) { + this.target.getListener().setExtractRequestPayload(extractRequestPayload); + return _this(); + } + + /** + * + * @param the target {@link JmsListenerContainerSpec} implementation type. + * @param the target {@link AbstractMessageListenerContainer} implementation type. + */ + public static class + JmsMessageDrivenChannelAdapterListenerContainerSpec, C extends AbstractMessageListenerContainer> + extends JmsMessageDrivenChannelAdapterSpec> { + + private final JmsListenerContainerSpec spec; + + JmsMessageDrivenChannelAdapterListenerContainerSpec(JmsListenerContainerSpec spec) { + super(spec.get()); + this.spec = spec; + this.spec.get().setAutoStartup(false); + } + + /** + * @param destination the destination. + * @return the spec. + * @see JmsListenerContainerSpec#destination(Destination) + */ + public JmsMessageDrivenChannelAdapterListenerContainerSpec destination(Destination destination) { + this.spec.destination(destination); + return _this(); + } + + /** + * Specify a destination name to use. + * @param destinationName the destinationName. + * @return the spec. + * @see JmsListenerContainerSpec#destination(String) + */ + public JmsMessageDrivenChannelAdapterListenerContainerSpec destination(String destinationName) { + this.spec.destination(destinationName); + return _this(); + } + + /** + * Configure a listener container by invoking the {@link Consumer} callback, with a + * {@link JmsListenerContainerSpec} argument. + * @param configurer the configurer. + * @return the spec. + */ + public JmsMessageDrivenChannelAdapterListenerContainerSpec configureListenerContainer( + Consumer> configurer) { + Assert.notNull(configurer); + configurer.accept(this.spec); + return _this(); + } + + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java new file mode 100644 index 0000000000..42b2a5ef82 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.java @@ -0,0 +1,145 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.function.Consumer; +import java.util.function.Function; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.jms.JmsHeaderMapper; +import org.springframework.integration.jms.JmsSendingMessageHandler; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandlerSpec} for a {@link JmsSendingMessageHandler}. + * + * @param the target {@link JmsOutboundChannelAdapterSpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsOutboundChannelAdapterSpec> + extends MessageHandlerSpec { + + protected final JmsTemplateSpec jmsTemplateSpec = new JmsTemplateSpec(); + + JmsOutboundChannelAdapterSpec(JmsTemplate jmsTemplate) { + this.target = new JmsSendingMessageHandler(jmsTemplate); + } + + private JmsOutboundChannelAdapterSpec(ConnectionFactory connectionFactory) { + this.target = new JmsSendingMessageHandler(this.jmsTemplateSpec.connectionFactory(connectionFactory).get()); + } + + /** + * @param extractPayload the extractPayload flag. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setExtractPayload(boolean) + */ + public S extractPayload(boolean extractPayload) { + this.target.setExtractPayload(extractPayload); + return _this(); + } + + /** + * @param headerMapper the headerMapper. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setHeaderMapper(JmsHeaderMapper) + */ + public S headerMapper(JmsHeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + return _this(); + } + + /** + * Configure the destination to which this adapter will send messages. + * @param destination the destination. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setDestination(Destination) + */ + public S destination(Destination destination) { + this.target.setDestination(destination); + return _this(); + } + + /** + * Configure the name of the destination to which this adapter will send messages. + * @param destination the destination name. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setDestinationName(String) + */ + public S destination(String destination) { + this.target.setDestinationName(destination); + return _this(); + } + + /** + * Configure a SpEL expression that will evaluate, at run time, the destination to + * which a message will be sent. + * @param destination the destination name. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setDestinationName(String) + */ + public S destinationExpression(String destination) { + this.target.setDestinationExpression(PARSER.parseExpression(destination)); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the destination to + * which a message will be sent. Typically used with a Java 8 Lambda expression: + * 
+	 * {@code
+	 * .destination(m -> m.getPayload().getState())
+	 * }
+	 *
+ * @param destinationFunction the destination function. + * @param

the expected payload type. + * @return the current {@link JmsOutboundChannelAdapterSpec}. + * @see JmsSendingMessageHandler#setDestinationName(String) + * @see FunctionExpression + */ + public

S destination(Function, ?> destinationFunction) { + this.target.setDestinationExpression(new FunctionExpression>(destinationFunction)); + return _this(); + } + + /** + * A {@link JmsTemplate}-based {@link JmsOutboundChannelAdapterSpec} extension. + */ + public static class JmsOutboundChannelSpecTemplateAware extends + JmsOutboundChannelAdapterSpec { + + JmsOutboundChannelSpecTemplateAware(ConnectionFactory connectionFactory) { + super(connectionFactory); + } + + public JmsOutboundChannelSpecTemplateAware configureJmsTemplate(Consumer configurer) { + Assert.notNull(configurer); + configurer.accept(this.jmsTemplateSpec); + return _this(); + } + + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java new file mode 100644 index 0000000000..8c25f36e41 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.java @@ -0,0 +1,438 @@ +/* + * Copyright 2016 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.jms.dsl; + +import java.util.concurrent.Executor; +import java.util.function.Consumer; +import java.util.function.Function; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.jms.JmsHeaderMapper; +import org.springframework.integration.jms.JmsOutboundGateway; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandlerSpec} for a {@link JmsOutboundGateway}. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsOutboundGatewaySpec extends MessageHandlerSpec { + + JmsOutboundGatewaySpec(ConnectionFactory connectionFactory) { + this.target = new JmsOutboundGateway(); + this.target.setConnectionFactory(connectionFactory); + this.target.setRequiresReply(true); + } + + /** + * @param extractPayload the extractPayload. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setExtractRequestPayload(boolean) + */ + public JmsOutboundGatewaySpec extractRequestPayload(boolean extractPayload) { + this.target.setExtractRequestPayload(extractPayload); + return _this(); + } + + /** + * @param extractPayload the extractPayload. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setExtractReplyPayload(boolean) + */ + public JmsOutboundGatewaySpec extractReplyPayload(boolean extractPayload) { + this.target.setExtractReplyPayload(extractPayload); + return _this(); + } + + /** + * @param headerMapper the headerMapper. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setHeaderMapper(JmsHeaderMapper) + */ + public JmsOutboundGatewaySpec headerMapper(JmsHeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + return _this(); + } + + /** + * @param destination the destination. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setRequestDestination(Destination) + */ + public JmsOutboundGatewaySpec requestDestination(Destination destination) { + this.target.setRequestDestination(destination); + return _this(); + } + + /** + * @param destination the destination name. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setRequestDestinationName(String) + */ + public JmsOutboundGatewaySpec requestDestination(String destination) { + this.target.setRequestDestinationName(destination); + return _this(); + } + + /** + * @param destination the destination expression. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setRequestDestinationExpression(org.springframework.expression.Expression) + */ + public JmsOutboundGatewaySpec requestDestinationExpression(String destination) { + this.target.setRequestDestinationExpression(PARSER.parseExpression(destination)); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the destination to + * which a message will be sent. Typically used with a Java 8 Lambda expression: + * 

+	 * {@code
+	 * .destination(m -> m.getPayload().getState())
+	 * }
+	 *
+ * @param destinationFunction the destination function. + * @param

the expected payload type. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setRequestDestinationExpression(org.springframework.expression.Expression) + * @see FunctionExpression + */ + public

JmsOutboundGatewaySpec requestDestination(Function, ?> destinationFunction) { + this.target.setRequestDestinationExpression(new FunctionExpression>(destinationFunction)); + return _this(); + } + + /** + * @param destination the destination. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReplyDestination(Destination) + */ + public JmsOutboundGatewaySpec replyDestination(Destination destination) { + this.target.setReplyDestination(destination); + return _this(); + } + + /** + * @param destination the destination name. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReplyDestinationName(String) + */ + public JmsOutboundGatewaySpec replyDestination(String destination) { + this.target.setReplyDestinationName(destination); + return _this(); + } + + /** + * @param destination the destination expression. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReplyDestinationExpression(org.springframework.expression.Expression) + */ + public JmsOutboundGatewaySpec replyDestinationExpression(String destination) { + this.target.setReplyDestinationExpression(PARSER.parseExpression(destination)); + return _this(); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the destination from + * which a reply will be received. Typically used with a Java 8 Lambda expression: + * 

+	 * {@code
+	 * .replyDestination(m -> m.getPayload().getState())
+	 * }
+	 *
+ * @param destinationFunction the destination function. + * @param

the expected payload type. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReplyDestinationExpression(org.springframework.expression.Expression) + * @see FunctionExpression + */ + public

JmsOutboundGatewaySpec replyDestination(Function, ?> destinationFunction) { + this.target.setReplyDestinationExpression(new FunctionExpression>(destinationFunction)); + return _this(); + } + + /** + * @param destinationResolver the destinationResolver. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setDestinationResolver(DestinationResolver) + */ + public JmsOutboundGatewaySpec destinationResolver(DestinationResolver destinationResolver) { + this.target.setDestinationResolver(destinationResolver); + return _this(); + } + + /** + * @param messageConverter the messageConverter. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setMessageConverter(MessageConverter) + */ + public JmsOutboundGatewaySpec jmsMessageConverter(MessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return _this(); + } + + /** + * @param correlationKey the correlationKey + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setCorrelationKey(String) + */ + public JmsOutboundGatewaySpec correlationKey(String correlationKey) { + this.target.setCorrelationKey(correlationKey); + return _this(); + } + + /** + * @param pubSubDomain the pubSubDomain + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReplyPubSubDomain(boolean) + */ + public JmsOutboundGatewaySpec requestPubSubDomain(boolean pubSubDomain) { + this.target.setRequestPubSubDomain(pubSubDomain); + return _this(); + } + + /** + * @param pubSubDomain the pubSubDomain + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setRequestPubSubDomain(boolean) + */ + public JmsOutboundGatewaySpec replyPubSubDomain(boolean pubSubDomain) { + this.target.setReplyPubSubDomain(pubSubDomain); + return _this(); + } + + /** + * @param deliveryPersistent the deliveryPersistent. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setDeliveryPersistent(boolean) + */ + public JmsOutboundGatewaySpec deliveryPersistent(boolean deliveryPersistent) { + this.target.setDeliveryPersistent(deliveryPersistent); + return _this(); + } + + /** + * Default priority. May be overridden at run time with a message + * priority header. + * @param priority the priority. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setPriority(int) + */ + public JmsOutboundGatewaySpec priority(int priority) { + this.target.setPriority(priority); + return _this(); + } + + /** + * @param timeToLive the timeToLive. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setTimeToLive(long) + */ + public JmsOutboundGatewaySpec timeToLive(long timeToLive) { + this.target.setTimeToLive(timeToLive); + return _this(); + } + + /** + * @param receiveTimeout the receiveTimeout. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setReceiveTimeout(long) + */ + public JmsOutboundGatewaySpec receiveTimeout(long receiveTimeout) { + this.target.setReceiveTimeout(receiveTimeout); + return _this(); + } + + /** + * @param explicitQosEnabled the explicitQosEnabled. + * @return the current {@link JmsOutboundGatewaySpec}. + * @see JmsOutboundGateway#setExplicitQosEnabled(boolean) + */ + public JmsOutboundGatewaySpec explicitQosEnabled(boolean explicitQosEnabled) { + this.target.setExplicitQosEnabled(explicitQosEnabled); + return _this(); + } + + /** + * Configure a reply container with default properties. + * @return the current {@link JmsOutboundGatewaySpec}. + */ + public JmsOutboundGatewaySpec replyContainer() { + this.target.setReplyContainerProperties(new JmsOutboundGateway.ReplyContainerProperties()); + return _this(); + } + + /** + * Configure a reply container with a reply container specification determined by + * invoking the {@link Consumer} callback with a {@link ReplyContainerSpec}. + * @param configurer the configurer. + * @return the current {@link JmsOutboundGatewaySpec}. + */ + public JmsOutboundGatewaySpec replyContainer(Consumer configurer) { + Assert.notNull(configurer); + ReplyContainerSpec spec = new ReplyContainerSpec(); + configurer.accept(spec); + this.target.setReplyContainerProperties(spec.get()); + return _this(); + } + + + /** + * An {@link IntegrationComponentSpec} for {@link JmsOutboundGateway.ReplyContainerProperties}. + * + */ + public class ReplyContainerSpec + extends IntegrationComponentSpec { + + ReplyContainerSpec() { + this.target = new JmsOutboundGateway.ReplyContainerProperties(); + } + + /** + * @param sessionTransacted the sessionTransacted. + * @return the current {@link ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setSessionTransacted(boolean) + */ + public ReplyContainerSpec sessionTransacted(Boolean sessionTransacted) { + this.target.setSessionTransacted(sessionTransacted); + return _this(); + } + + /** + * @param sessionAcknowledgeMode the acknowledgement mode constant + * @return the current {@link ReplyContainerSpec}. + * @see javax.jms.Session#AUTO_ACKNOWLEDGE etc. + */ + public ReplyContainerSpec sessionAcknowledgeMode(Integer sessionAcknowledgeMode) { + this.target.setSessionAcknowledgeMode(sessionAcknowledgeMode); + return _this(); + } + + /** + * @param receiveTimeout the receiveTimeout. + * @return the current {@link ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setReceiveTimeout(long) + */ + public ReplyContainerSpec receiveTimeout(Long receiveTimeout) { + this.target.setReceiveTimeout(receiveTimeout); + return _this(); + } + + /** + * @param recoveryInterval the recoveryInterval. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setRecoveryInterval(long) + */ + public ReplyContainerSpec recoveryInterval(Long recoveryInterval) { + this.target.setRecoveryInterval(recoveryInterval); + return _this(); + } + + /** + * @param cacheLevel the value for + * {@code org.springframework.jms.listener.DefaultMessageListenerContainer.cacheLevel}. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setCacheLevel(int) + */ + public ReplyContainerSpec cacheLevel(Integer cacheLevel) { + this.target.setCacheLevel(cacheLevel); + return _this(); + } + + /** + * @param concurrentConsumers the concurrentConsumers. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setConcurrentConsumers(int) + */ + public ReplyContainerSpec concurrentConsumers(Integer concurrentConsumers) { + this.target.setConcurrentConsumers(concurrentConsumers); + return _this(); + } + + /** + * @param maxConcurrentConsumers the maxConcurrentConsumers. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setMaxConcurrentConsumers(int) + */ + public ReplyContainerSpec maxConcurrentConsumers(Integer maxConcurrentConsumers) { + this.target.setMaxConcurrentConsumers(maxConcurrentConsumers); + return _this(); + } + + /** + * @param maxMessagesPerTask the maxMessagesPerTask. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setMaxMessagesPerTask(int) + */ + public ReplyContainerSpec maxMessagesPerTask(Integer maxMessagesPerTask) { + this.target.setMaxMessagesPerTask(maxMessagesPerTask); + return _this(); + } + + /** + * @param idleConsumerLimit the idleConsumerLimit. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setIdleConsumerLimit(int) + */ + public ReplyContainerSpec idleConsumerLimit(Integer idleConsumerLimit) { + this.target.setIdleConsumerLimit(idleConsumerLimit); + return _this(); + } + + /** + * @param idleTaskExecutionLimit the idleTaskExecutionLimit. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setIdleTaskExecutionLimit(int) + */ + public ReplyContainerSpec idleTaskExecutionLimit(Integer idleTaskExecutionLimit) { + this.target.setIdleTaskExecutionLimit(idleTaskExecutionLimit); + return _this(); + } + + /** + * @param taskExecutor the taskExecutor. + * @return the current {ReplyContainerSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setTaskExecutor(Executor) + */ + public ReplyContainerSpec taskExecutor(Executor taskExecutor) { + this.target.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * @param idleReplyContainerTimeout the timeout in seconds. + * @return the current {ReplyContainerSpec}. + * @see JmsOutboundGateway#setIdleReplyContainerTimeout + */ + public ReplyContainerSpec idleReplyContainerTimeout(long idleReplyContainerTimeout) { + JmsOutboundGatewaySpec.this.target.setIdleReplyContainerTimeout(idleReplyContainerTimeout); + return this; + } + + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPollableMessageChannelSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPollableMessageChannelSpec.java new file mode 100644 index 0000000000..fd167c55e9 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPollableMessageChannelSpec.java @@ -0,0 +1,225 @@ +/* + * Copyright 2016 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.jms.dsl; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.jms.AbstractJmsChannel; +import org.springframework.integration.jms.config.JmsChannelFactoryBean; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.jms.support.destination.DestinationResolver; + +/** + * A {@link MessageChannelSpec} for an {@link AbstractJmsChannel}. + * + * @param the target {@link JmsPollableMessageChannelSpec} implementation type. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsPollableMessageChannelSpec> + extends MessageChannelSpec { + + protected final JmsChannelFactoryBean jmsChannelFactoryBean; + + JmsPollableMessageChannelSpec(ConnectionFactory connectionFactory) { + this(new JmsChannelFactoryBean(false), connectionFactory); + } + + JmsPollableMessageChannelSpec(JmsChannelFactoryBean jmsChannelFactoryBean, ConnectionFactory connectionFactory) { + this.jmsChannelFactoryBean = jmsChannelFactoryBean; + this.jmsChannelFactoryBean.setConnectionFactory(connectionFactory); + this.jmsChannelFactoryBean.setSingleton(false); + this.jmsChannelFactoryBean.setBeanFactory(new DefaultListableBeanFactory()); + } + + @Override + protected S id(String id) { + this.jmsChannelFactoryBean.setBeanName(id); + return super.id(id); + } + + /** + * Configure the destination name that backs this channel. + * @param destination the destination. + * @return the current {@link MessageChannelSpec}. + */ + public S destination(String destination) { + this.jmsChannelFactoryBean.setDestinationName(destination); + return _this(); + } + + /** + * @param destinationResolver the destinationResolver. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setDestinationResolver(DestinationResolver) + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setDestinationResolver(DestinationResolver) + */ + public S destinationResolver(DestinationResolver destinationResolver) { + this.jmsChannelFactoryBean.setDestinationResolver(destinationResolver); + return _this(); + } + + /** + * Configure the destination that backs this channel. + * @param destination the destination. + * @return the current {@link MessageChannelSpec}. + */ + public S destination(Destination destination) { + this.jmsChannelFactoryBean.setDestination(destination); + return _this(); + } + + /** + * Configure a message selector in the + * {@link org.springframework.jms.listener.DefaultMessageListenerContainer} (when + * message driven) or the {@link org.springframework.jms.core.JmsTemplate} (when + * polled). + * @param messageSelector the messageSelector. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setMessageSelector(String) + * @see org.springframework.jms.core.JmsTemplate#receiveSelectedAndConvert(String) + */ + public S messageSelector(String messageSelector) { + this.jmsChannelFactoryBean.setMessageSelector(messageSelector); + return _this(); + } + + /** + * Configure the {@link MessageConverter} used for both sending and + * receiving. + * @param messageConverter the messageConverter. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setMessageConverter(MessageConverter) + */ + public S jmsMessageConverter(MessageConverter messageConverter) { + this.jmsChannelFactoryBean.setMessageConverter(messageConverter); + return _this(); + } + + /** + * @param deliveryPersistent the deliveryPersistent. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setDeliveryPersistent(boolean) + */ + public S deliveryPersistent(boolean deliveryPersistent) { + this.jmsChannelFactoryBean.setDeliveryPersistent(deliveryPersistent); + return _this(); + } + + /** + * @param explicitQosEnabled the explicitQosEnabled. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setExplicitQosEnabled(boolean) + */ + public S explicitQosEnabled(boolean explicitQosEnabled) { + this.jmsChannelFactoryBean.setExplicitQosEnabled(explicitQosEnabled); + return _this(); + } + + /** + * @param messageIdEnabled the messageIdEnabled. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setMessageIdEnabled(boolean) + */ + public S messageIdEnabled(boolean messageIdEnabled) { + this.jmsChannelFactoryBean.setMessageIdEnabled(messageIdEnabled); + return _this(); + } + + /** + * @param messageTimestampEnabled the messageTimestampEnabled. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setMessageTimestampEnabled(boolean) + */ + public S messageTimestampEnabled(boolean messageTimestampEnabled) { + this.jmsChannelFactoryBean.setMessageTimestampEnabled(messageTimestampEnabled); + return _this(); + } + + /** + * Default priority. May be overridden at run time with a message + * priority header. + * @param priority the priority. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setPriority(int) + */ + public S priority(int priority) { + this.jmsChannelFactoryBean.setPriority(priority); + return _this(); + } + + /** + * @param timeToLive the timeToLive. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setTimeToLive(long) + */ + public S timeToLive(long timeToLive) { + this.jmsChannelFactoryBean.setTimeToLive(timeToLive); + return _this(); + } + + /** + * @param receiveTimeout the receiveTimeout. + * @return the current {@link MessageChannelSpec}. + * @see org.springframework.jms.core.JmsTemplate#setReceiveTimeout(long) + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setReceiveTimeout(long) + */ + public S receiveTimeout(long receiveTimeout) { + this.jmsChannelFactoryBean.setReceiveTimeout(receiveTimeout); + return _this(); + } + + /** + * @param sessionAcknowledgeMode the acknowledgement mode constant + * @return the current {@link MessageChannelSpec}. + * @see javax.jms.Session#AUTO_ACKNOWLEDGE etc. + */ + public S sessionAcknowledgeMode(int sessionAcknowledgeMode) { + this.jmsChannelFactoryBean.setSessionAcknowledgeMode(sessionAcknowledgeMode); + return _this(); + } + + /** + * Configure transactional sessions for both the + * {@link org.springframework.jms.core.JmsTemplate} (sends and polled receives) and + * {@link org.springframework.jms.listener.DefaultMessageListenerContainer} + * (message-driven receives). + * @param sessionTransacted the sessionTransacted. + * @return the current {@link MessageChannelSpec}. + */ + public S sessionTransacted(boolean sessionTransacted) { + this.jmsChannelFactoryBean.setSessionTransacted(sessionTransacted); + return _this(); + } + + @Override + protected AbstractJmsChannel doGet() { + try { + this.channel = this.jmsChannelFactoryBean.getObject(); + } + catch (Exception e) { + throw new BeanCreationException("Cannot create the JMS MessageChannel", e); + } + return super.doGet(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPublishSubscribeMessageChannelSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPublishSubscribeMessageChannelSpec.java new file mode 100644 index 0000000000..b492513ba7 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsPublishSubscribeMessageChannelSpec.java @@ -0,0 +1,81 @@ +/* + * Copyright 2016 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.jms.dsl; + +import javax.jms.ConnectionFactory; + +import org.springframework.jms.listener.DefaultMessageListenerContainer; + +/** + * A {@link JmsMessageChannelSpec} for a {@link org.springframework.integration.jms.SubscribableJmsChannel} + * configured with a topic. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsPublishSubscribeMessageChannelSpec + extends JmsMessageChannelSpec { + + JmsPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) { + super(connectionFactory); + this.jmsChannelFactoryBean.setPubSubDomain(true); + } + + /** + * @param durable the durable. + * @return the current {@link JmsPublishSubscribeMessageChannelSpec}. + * @see org.springframework.jms.listener.AbstractMessageListenerContainer#setSubscriptionDurable(boolean) + */ + public JmsPublishSubscribeMessageChannelSpec subscriptionDurable(boolean durable) { + this.jmsChannelFactoryBean.setSubscriptionDurable(durable); + return _this(); + } + + /** + * @param durableSubscriptionName the durableSubscriptionName. + * @return the current {@link JmsPublishSubscribeMessageChannelSpec}. + * @see org.springframework.jms.listener.AbstractMessageListenerContainer#setDurableSubscriptionName(String) + */ + public JmsPublishSubscribeMessageChannelSpec durableSubscriptionName(String durableSubscriptionName) { + this.jmsChannelFactoryBean.setDurableSubscriptionName(durableSubscriptionName); + return _this(); + } + + /** + * @param clientId the clientId. + * @return the current {@link JmsPublishSubscribeMessageChannelSpec}. + * @see org.springframework.jms.listener.AbstractMessageListenerContainer#setClientId(String) + */ + public JmsPublishSubscribeMessageChannelSpec clientId(String clientId) { + this.jmsChannelFactoryBean.setClientId(clientId); + return _this(); + } + + /** + * Only applies if the {@link #containerType(Class)} is a {@link DefaultMessageListenerContainer} + * or a {@link org.springframework.jms.listener.SimpleMessageListenerContainer}. + * @param pubSubNoLocal the pubSubNoLocal. + * @return the current {@link JmsPublishSubscribeMessageChannelSpec}. + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#setPubSubNoLocal(boolean) + * @see org.springframework.jms.listener.SimpleMessageListenerContainer#setPubSubNoLocal(boolean) + */ + public JmsPublishSubscribeMessageChannelSpec pubSubNoLocal(boolean pubSubNoLocal) { + this.jmsChannelFactoryBean.setPubSubNoLocal(pubSubNoLocal); + return _this(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsTemplateSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsTemplateSpec.java new file mode 100644 index 0000000000..bcac4c80ca --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsTemplateSpec.java @@ -0,0 +1,95 @@ +/* + * Copyright 2016 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.jms.dsl; + +import org.springframework.integration.jms.DynamicJmsTemplate; +import org.springframework.jms.support.converter.MessageConverter; + +/** + * A {@link JmsDestinationAccessorSpec} for a {@link DynamicJmsTemplate}. + * + * @author Artem Bilan + * @since 5.0 + */ +public class JmsTemplateSpec extends JmsDestinationAccessorSpec { + + JmsTemplateSpec() { + super(new DynamicJmsTemplate()); + } + + /** + * @param messageConverter the messageConverter. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setMessageConverter(MessageConverter) + */ + public JmsTemplateSpec jmsMessageConverter(MessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return _this(); + } + + /** + * @param deliveryPersistent the deliveryPersistent. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setDeliveryPersistent(boolean) + */ + public JmsTemplateSpec deliveryPersistent(boolean deliveryPersistent) { + this.target.setDeliveryPersistent(deliveryPersistent); + return _this(); + } + + /** + * @param explicitQosEnabled the explicitQosEnabled. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setExplicitQosEnabled(boolean) + */ + public JmsTemplateSpec explicitQosEnabled(boolean explicitQosEnabled) { + this.target.setExplicitQosEnabled(explicitQosEnabled); + return _this(); + } + + /** + * May be overridden at run time by the message priority header. + * @param priority the priority. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setPriority(int) + */ + public JmsTemplateSpec priority(int priority) { + this.target.setPriority(priority); + return _this(); + } + + /** + * @param timeToLive the timeToLive. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setTimeToLive(long) + */ + public JmsTemplateSpec timeToLive(long timeToLive) { + this.target.setTimeToLive(timeToLive); + return _this(); + } + + /** + * @param receiveTimeout the receiveTimeout. + * @return the spec. + * @see org.springframework.jms.core.JmsTemplate#setReceiveTimeout(long) + */ + public JmsTemplateSpec receiveTimeout(long receiveTimeout) { + this.target.setReceiveTimeout(receiveTimeout); + return _this(); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/package-info.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/package-info.java new file mode 100644 index 0000000000..e3d6f8c571 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides JMS Component support for the Java DSL. + */ +package org.springframework.integration.jms.dsl; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java new file mode 100644 index 0000000000..65223cc805 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java @@ -0,0 +1,434 @@ +/* + * Copyright 2016 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.jms.dsl; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.jms.ConnectionFactory; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.ChannelInterceptorAware; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowDefinition; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.simp.SimpMessageHeaderAccessor; +import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.stereotype.Component; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Artem Bilan + * @author Gary Russell + * @author Nasko Vasilev + * + * @since 5.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class JmsTests { + + @Autowired + private ListableBeanFactory beanFactory; + + @Autowired + private ControlBusGateway controlBus; + + @Autowired + @Qualifier("flow1QueueChannel") + private PollableChannel outputChannel; + + @Autowired + @Qualifier("jmsOutboundFlow.input") + private MessageChannel jmsOutboundInboundChannel; + + @Autowired + @Qualifier("jmsOutboundInboundReplyChannel") + private PollableChannel jmsOutboundInboundReplyChannel; + + @Autowired + @Qualifier("jmsOutboundGatewayFlow.input") + private MessageChannel jmsOutboundGatewayChannel; + + @Autowired + private TestChannelInterceptor testChannelInterceptor; + + @Autowired + private ConnectionFactory jmsConnectionFactory; + + @Autowired + private PollableChannel jmsPubSubBridgeChannel; + + @Autowired + @Qualifier("jmsOutboundGateway.handler") + private MessageHandler jmsOutboundGatewayHandler; + + @Autowired + private AtomicBoolean jmsMessageDrivenChannelCalled; + + @Autowired + private AtomicBoolean jmsInboundGatewayChannelCalled; + + @Test + public void testPollingFlow() { + this.controlBus.send("@'jmsTests.ContextConfiguration.integerMessageSource.inboundChannelAdapter'.start()"); + assertThat(this.beanFactory.getBean("integerChannel"), instanceOf(FixedSubscriberChannel.class)); + for (int i = 0; i < 5; i++) { + Message message = this.outputChannel.receive(20000); + assertNotNull(message); + assertEquals("" + i, message.getPayload()); + } + this.controlBus.send("@'jmsTests.ContextConfiguration.integerMessageSource.inboundChannelAdapter'.stop()"); + + assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors() + .contains(this.testChannelInterceptor)); + assertThat(this.testChannelInterceptor.invoked.get(), Matchers.greaterThanOrEqualTo(5)); + + } + + @Test + public void testJmsOutboundInboundFlow() { + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsInbound") + .build()); + + Message receive = this.jmsOutboundInboundReplyChannel.receive(10000); + + assertNotNull(receive); + assertEquals("HELLO THROUGH THE JMS", receive.getPayload()); + + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriven") + .build()); + + receive = this.jmsOutboundInboundReplyChannel.receive(10000); + + assertNotNull(receive); + assertEquals("hello through the jms", receive.getPayload()); + + assertTrue(this.jmsMessageDrivenChannelCalled.get()); + + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload(" foo ") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "containerSpecDestination") + .build()); + + receive = this.jmsOutboundInboundReplyChannel.receive(10000); + + assertNotNull(receive); + assertEquals("foo", receive.getPayload()); + } + + @Test + public void testJmsPipelineFlow() { + assertEquals(new Long(10000), + TestUtils.getPropertyValue(this.jmsOutboundGatewayHandler, "idleReplyContainerTimeout", Long.class)); + PollableChannel replyChannel = new QueueChannel(); + Message message = MessageBuilder.withPayload("hello through the jms pipeline") + .setReplyChannel(replyChannel) + .setHeader("destination", "jmsPipelineTest") + .build(); + this.jmsOutboundGatewayChannel.send(message); + + Message receive = replyChannel.receive(5000); + + assertNotNull(receive); + assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload()); + + assertTrue(this.jmsInboundGatewayChannelCalled.get()); + } + + @Test + public void testPubSubFlow() { + JmsTemplate template = new JmsTemplate(this.jmsConnectionFactory); + template.setPubSubDomain(true); + template.setDefaultDestinationName("pubsub"); + template.convertAndSend("foo"); + Message received = this.jmsPubSubBridgeChannel.receive(5000); + assertNotNull(received); + assertEquals("foo", received.getPayload()); + } + + @Autowired + private CountDownLatch redeliveryLatch; + + @Test + public void testJmsRedeliveryFlow() throws InterruptedException { + this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("foo") + .setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDrivenRedelivery") + .build()); + + assertTrue(this.redeliveryLatch.await(10, TimeUnit.SECONDS)); + } + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + private interface ControlBusGateway { + + void send(String command); + + } + + @Configuration + @EnableIntegration + @IntegrationComponentScan + @ComponentScan + public static class ContextConfiguration { + + @Bean + public ConnectionFactory cachingConnectionFactory() { + return new CachingConnectionFactory(jmsConnectionFactory()); + } + + @Bean + public ActiveMQConnectionFactory jmsConnectionFactory() { + ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory( + "vm://localhost?broker.persistent=false"); + activeMQConnectionFactory.setTrustAllPackages(true); + return activeMQConnectionFactory; + } + + @Bean + public JmsTemplate jmsTemplate() { + return new JmsTemplate(jmsConnectionFactory()); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata poller() { + return Pollers.fixedRate(500).get(); + } + + @Bean + public IntegrationFlow controlBus() { + return IntegrationFlowDefinition::controlBus; + } + + @Bean + @InboundChannelAdapter(value = "flow1.input", autoStartup = "false", poller = @Poller(fixedRate = "100")) + public MessageSource integerMessageSource() { + MethodInvokingMessageSource source = new MethodInvokingMessageSource(); + source.setObject(new AtomicInteger()); + source.setMethodName("getAndIncrement"); + return source; + } + + @Bean + public IntegrationFlow flow1() { + return f -> f + .fixedSubscriberChannel("integerChannel") + .transform("payload.toString()") + .channel(Jms.pollableChannel("flow1QueueChannel", cachingConnectionFactory()) + .destination("flow1QueueChannel")); + } + + @Bean + public IntegrationFlow jmsOutboundFlow() { + return f -> f.handle(Jms.outboundAdapter(jmsConnectionFactory()) + .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)); + } + + @Bean + public MessageChannel jmsOutboundInboundReplyChannel() { + return MessageChannels.queue().get(); + } + + @Bean + public IntegrationFlow jmsInboundFlow() { + return IntegrationFlows + .from(Jms.inboundAdapter(cachingConnectionFactory()).destination("jmsInbound")) + .transform(String::toUpperCase) + .channel(this.jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public IntegrationFlow pubSubFlow() { + return IntegrationFlows + .from(Jms.publishSubscribeChannel(jmsConnectionFactory()) + .destination("pubsub")) + .channel(c -> c.queue("jmsPubSubBridgeChannel")) + .get(); + } + + @Bean + public IntegrationFlow jmsMessageDrivenFlow() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory()) + .outputChannel(jmsMessageDrivenInputChannel()) + .destination("jmsMessageDriven")) + .transform(String::toLowerCase) + .channel(jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public AtomicBoolean jmsMessageDrivenChannelCalled() { + return new AtomicBoolean(); + } + + @Bean + public MessageChannel jmsMessageDrivenInputChannel() { + DirectChannel directChannel = new DirectChannel(); + directChannel.addInterceptor(new ChannelInterceptorAdapter() { + + @Override + public Message preSend(Message message, MessageChannel channel) { + jmsMessageDrivenChannelCalled().set(true); + return super.preSend(message, channel); + } + + }); + return directChannel; + } + + @Bean + public IntegrationFlow jmsMessageDrivenFlowWithContainer() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter( + Jms.container(jmsConnectionFactory(), "containerSpecDestination") + .pubSubDomain(false) + .taskExecutor(Executors.newCachedThreadPool()) + .get())) + .transform(String::trim) + .channel(jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public IntegrationFlow jmsOutboundGatewayFlow() { + return f -> f.handle(Jms.outboundGateway(jmsConnectionFactory()) + .replyContainer(c -> c.idleReplyContainerTimeout(10)) + .requestDestination("jmsPipelineTest"), + e -> e.id("jmsOutboundGateway")); + } + + @Bean + public IntegrationFlow jmsInboundGatewayFlow() { + return IntegrationFlows.from(Jms.inboundGateway(jmsConnectionFactory()) + .requestChannel(jmsInboundGatewayInputChannel()) + .destination("jmsPipelineTest")) + .transform(String::toUpperCase) + .get(); + } + + @Bean + public AtomicBoolean jmsInboundGatewayChannelCalled() { + return new AtomicBoolean(); + } + + @Bean + public MessageChannel jmsInboundGatewayInputChannel() { + DirectChannel directChannel = new DirectChannel(); + directChannel.addInterceptor(new ChannelInterceptorAdapter() { + + @Override + public Message preSend(Message message, MessageChannel channel) { + jmsInboundGatewayChannelCalled().set(true); + return super.preSend(message, channel); + } + + }); + return directChannel; + } + + @Bean + public IntegrationFlow jmsMessageDrivenRedeliveryFlow() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory()) + .errorChannel(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) + .destination("jmsMessageDrivenRedelivery")) + .transform(p -> { + throw new RuntimeException("intentional"); + }) + .get(); + } + + @Bean + public CountDownLatch redeliveryLatch() { + return new CountDownLatch(3); + } + + @Bean + public IntegrationFlow errorHandlingFlow() { + return IntegrationFlows.from(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) + .handle(m -> { + MessagingException exception = (MessagingException) m.getPayload(); + redeliveryLatch().countDown(); + throw exception; + }) + .get(); + } + + } + + @Component + @GlobalChannelInterceptor(patterns = "flow1QueueChannel") + public static class TestChannelInterceptor extends ChannelInterceptorAdapter { + + private final AtomicInteger invoked = new AtomicInteger(); + + @Override + public Message preSend(Message message, MessageChannel channel) { + this.invoked.incrementAndGet(); + return message; + } + + } + +}