From 7394516a2dcd7ffc3777a3e36e36acb73edc2993 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 10 Jun 2014 18:42:23 +0300 Subject: [PATCH] DSL: Add `Jms` factory * Add `Spec` implementations for `JmsChannel`s * Fix the bug around `SourcePollingChannelAdapterFactoryBean` named bean registration * Remove unused classes * Add test for `Jms.pollableChannel` --- spring-integration-java-dsl/build.gradle | 7 + .../dsl/IntegrationFlowBuilder.java | 6 +- .../integration/dsl/IntegrationFlows.java | 6 +- .../integration/dsl/amqp/Amqp.java | 10 +- .../dsl/config/InstanceBeanDefinition.java | 53 ------- .../dsl/config/SimpleFactoryBean.java | 95 ------------ .../integration/dsl/config/package-info.java | 4 - .../IntegrationFlowBeanPostProcessor.java | 12 ++ .../integration/dsl/jms/Jms.java | 52 +++++++ .../dsl/jms/JmsMessageChannelSpec.java | 141 ++++++++++++++++++ .../jms/JmsPollableMessageChannelSpec.java | 133 +++++++++++++++++ ...JmsPublishSubscribeMessageChannelSpec.java | 52 +++++++ .../integration/dsl/jms/package-info.java | 4 + .../dsl/test/IntegrationFlowTests.java | 56 ++++--- 14 files changed, 447 insertions(+), 184 deletions(-) delete mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java delete mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java delete mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPollableMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPublishSubscribeMessageChannelSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/package-info.java diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index fc79b3b..92b6a00 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -23,8 +23,10 @@ compileTestJava { } ext { + activeMqVersion = '5.9.0' embedMongoVersion = '1.45' jacocoVersion = '0.7.0.201403182114' + jmsApiVersion = '1.1-rev-1' log4jVersion = '1.2.17' slf4jVersion = '1.7.6' springIntegrationVersion = '4.0.2.RELEASE' @@ -73,6 +75,7 @@ dependencies { , 'spring-integration-xmpp'].each { compile("org.springframework.integration:$it:$springIntegrationVersion", optional) } + compile ("javax.jms:jms-api:$jmsApiVersion", provided) testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion" testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:$embedMongoVersion" @@ -81,6 +84,10 @@ dependencies { testRuntime "log4j:log4j:$log4jVersion" testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion" testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" + testRuntime("org.apache.activemq:activemq-broker:$activeMqVersion") + testRuntime("org.apache.activemq:activemq-kahadb-store:$activeMqVersion") { + exclude group: "org.springframework" + } jacoco "org.jacoco:org.jacoco.agent:$jacocoVersion:runtime" } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 935c065..d96c631 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -570,9 +570,9 @@ public final class IntegrationFlowBuilder { messageProducer.setOutputChannel(outputChannel); } } - else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { + else if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) { SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = - (SourcePollingChannelAdapterFactoryBean) this.currentComponent; + ((SourcePollingChannelAdapterSpec) this.currentComponent).get().getT1(); if (channelName != null) { pollingChannelAdapterFactoryBean.setOutputChannelName(channelName); } @@ -611,7 +611,7 @@ public final class IntegrationFlowBuilder { if (this.flow.getIntegrationComponents().size() == 1) { if (this.currentComponent != null) { - if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { + if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) { throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent + ") " + "must be configured with at least one 'MessageChanel' or 'MessageHandler'."); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 4ce57f7..f7b7a7e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -18,7 +18,6 @@ package org.springframework.integration.dsl; import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.core.MessagingGatewaySpec; @@ -74,10 +73,9 @@ public final class IntegrationFlows { if (endpointConfigurer != null) { endpointConfigurer.configure(spec); } - SourcePollingChannelAdapterFactoryBean sourcePollingChannelAdapterFactoryBean = spec.get().getT1(); return new IntegrationFlowBuilder() - .addComponent(sourcePollingChannelAdapterFactoryBean) - .currentComponent(sourcePollingChannelAdapterFactoryBean); + .addComponent(spec) + .currentComponent(spec); } public static IntegrationFlowBuilder from(MessagingProducerSpec messagingProducerSpec) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java index 2364c74..231d584 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java @@ -77,18 +77,16 @@ public abstract class Amqp { return new AmqpOutboundEndpointSpec(endpoint, expectReply); } - public static > AmqpPollableMessageChannelSpec - pollableChannel(ConnectionFactory connectionFactory) { + public static > AmqpPollableMessageChannelSpecpollableChannel(ConnectionFactory connectionFactory) { return pollableChannel(null, connectionFactory); } - public static > AmqpPollableMessageChannelSpec - pollableChannel(String id, ConnectionFactory connectionFactory) { + public static > AmqpPollableMessageChannelSpec pollableChannel(String id, + ConnectionFactory connectionFactory) { return new AmqpPollableMessageChannelSpec(connectionFactory).id(id); } - public static - > AmqpMessageChannelSpec channel(ConnectionFactory connectionFactory) { + public static > AmqpMessageChannelSpec channel(ConnectionFactory connectionFactory) { return channel(null, connectionFactory); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java deleted file mode 100644 index 8ddce26..0000000 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright 2014 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.config; - -import org.springframework.beans.factory.config.ConstructorArgumentValues; -import org.springframework.beans.factory.support.GenericBeanDefinition; - -/** - * @author Artem Bilan - */ -@SuppressWarnings("serial") -public final class InstanceBeanDefinition extends GenericBeanDefinition { - - private final Object instance; - - public InstanceBeanDefinition(Object instance) { - this.instance = instance; - ConstructorArgumentValues args = new ConstructorArgumentValues(); - args.addGenericArgumentValue(this.instance); - this.setConstructorArgumentValues(args); - this.setBeanClass(SimpleFactoryBean.class); - } - - public InstanceBeanDefinition(InstanceBeanDefinition original) { - super(original); - this.instance = original.instance; - } - - @Override - public InstanceBeanDefinition cloneBeanDefinition() { - return new InstanceBeanDefinition(this); - } - - @Override - public Object getSource() { - return this.instance; - } - -} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java deleted file mode 100644 index b632e5d..0000000 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Copyright 2014 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.config; - -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.config.AbstractFactoryBean; -import org.springframework.beans.factory.config.AutowireCapableBeanFactory; -import org.springframework.context.Lifecycle; -import org.springframework.context.SmartLifecycle; - -/** - * @author Artem Bilan - */ -class SimpleFactoryBean extends AbstractFactoryBean implements BeanNameAware, SmartLifecycle { - - private final T target; - - private String name; - - public SimpleFactoryBean(T target) { - this.target = target; - } - - @Override - public void setBeanName(String name) { - this.name = name; - } - - @Override - public Class getObjectType() { - return this.target.getClass(); - } - - @Override - protected T createInstance() throws Exception { - ((AutowireCapableBeanFactory) this.getBeanFactory()).initializeBean(this.target, this.name); - return this.target; - } - - @Override - public boolean isAutoStartup() { - return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isAutoStartup(); - } - - @Override - public void stop(Runnable callback) { - if (this.target instanceof SmartLifecycle) { - ((SmartLifecycle) this.target).stop(callback); - } - } - - @Override - public void start() { - if (this.target instanceof Lifecycle) { - ((Lifecycle) this.target).start(); - } - } - - @Override - public void stop() { - if (this.target instanceof Lifecycle) { - ((Lifecycle) this.target).start(); - } - } - - @Override - public boolean isRunning() { - return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isRunning(); - } - - @Override - public int getPhase() { - if (this.target instanceof SmartLifecycle) { - return ((SmartLifecycle) this.target).getPhase(); - } - else { - return 0; - } - } - -} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java deleted file mode 100644 index 1651fd5..0000000 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -/** - * Provides configuration classes of the Spring Integration Java DSL. - */ -package org.springframework.integration.dsl.config; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java index f37993a..a6ea734 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java @@ -30,11 +30,14 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigUtils; +import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * @author Artem Bilan @@ -126,6 +129,15 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean } } } + else if (component instanceof SourcePollingChannelAdapterSpec) { + SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component; + String id = ((IntegrationComponentSpec) spec).getId(); + SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); + if (!StringUtils.hasText(id)) { + id = generateBeanName(pollingChannelAdapterFactoryBean); + } + registerComponent(pollingChannelAdapterFactoryBean, id); + } else if (!this.beanFactory .getBeansOfType(AopUtils.getTargetClass(component), false, false) .values() diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java new file mode 100644 index 0000000..efd9e7f --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java @@ -0,0 +1,52 @@ +/* + * Copyright 2014 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.jms; + +import javax.jms.ConnectionFactory; + +/** + * @author Artem Bilan + */ +public abstract class Jms { + + public static > JmsPollableMessageChannelSpec pollableChannel(ConnectionFactory connectionFactory) { + return pollableChannel(null, connectionFactory); + } + + public static > JmsPollableMessageChannelSpec pollableChannel(String id, + ConnectionFactory connectionFactory) { + return new JmsPollableMessageChannelSpec(connectionFactory).id(id); + } + + public static > JmsMessageChannelSpec channel(ConnectionFactory connectionFactory) { + return channel(null, connectionFactory); + } + + public static > JmsMessageChannelSpec channel(String id, + ConnectionFactory connectionFactory) { + return new JmsMessageChannelSpec(connectionFactory).id(id); + } + + public static JmsPublishSubscribeMessageChannelSpec publishSubscribeChannel(ConnectionFactory connectionFactory) { + return publishSubscribeChannel(null, connectionFactory); + } + + public static JmsPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id, + ConnectionFactory connectionFactory) { + return new JmsPublishSubscribeMessageChannelSpec(connectionFactory).id(id); + } +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java new file mode 100644 index 0000000..260f8ba --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java @@ -0,0 +1,141 @@ +/* + * Copyright 2014 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.jms; + +import java.util.concurrent.Executor; + +import javax.jms.ConnectionFactory; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.jms.AbstractJmsChannel; +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; + +/** + * @author Artem Bilan + */ +public class JmsMessageChannelSpec> extends JmsPollableMessageChannelSpec { + + private Integer cacheLevel; + + JmsMessageChannelSpec(ConnectionFactory connectionFactory) { + super(new JmsChannelFactoryBean(true), connectionFactory); + } + + public S containerType(Class containerType) { + this.jmsChannelFactoryBean.setContainerType(containerType); + return _this(); + } + + public S concurrentConsumers(int concurrentConsumers) { + this.jmsChannelFactoryBean.setConcurrentConsumers(concurrentConsumers); + return _this(); + } + + public S maxSubscribers(int maxSubscribers) { + this.jmsChannelFactoryBean.setMaxSubscribers(maxSubscribers); + return _this(); + } + + public S autoStartup(boolean autoStartup) { + this.jmsChannelFactoryBean.setAutoStartup(autoStartup); + return _this(); + } + + public S phase(int phase) { + this.jmsChannelFactoryBean.setPhase(phase); + return _this(); + } + + public S errorHandler(ErrorHandler errorHandler) { + this.jmsChannelFactoryBean.setErrorHandler(errorHandler); + return _this(); + } + + public S exposeListenerSession(boolean exposeListenerSession) { + this.jmsChannelFactoryBean.setExposeListenerSession(exposeListenerSession); + return _this(); + } + + public S acceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + this.jmsChannelFactoryBean.setAcceptMessagesWhileStopping(acceptMessagesWhileStopping); + return _this(); + } + + public S idleTaskExecutionLimit(int idleTaskExecutionLimit) { + this.jmsChannelFactoryBean.setIdleTaskExecutionLimit(idleTaskExecutionLimit); + return _this(); + } + + public S maxMessagesPerTask(int maxMessagesPerTask) { + this.jmsChannelFactoryBean.setMaxMessagesPerTask(maxMessagesPerTask); + return _this(); + } + + public S recoveryInterval(long recoveryInterval) { + this.jmsChannelFactoryBean.setRecoveryInterval(recoveryInterval); + return _this(); + } + + public S taskExecutor(Executor taskExecutor) { + this.jmsChannelFactoryBean.setTaskExecutor(taskExecutor); + return _this(); + } + + public S transactionManager(PlatformTransactionManager transactionManager) { + this.jmsChannelFactoryBean.setTransactionManager(transactionManager); + return _this(); + } + + public S transactionName(String transactionName) { + this.jmsChannelFactoryBean.setTransactionName(transactionName); + return _this(); + } + + public S transactionTimeout(int transactionTimeout) { + this.jmsChannelFactoryBean.setTransactionTimeout(transactionTimeout); + return _this(); + } + + /** + * @param cacheLevel the value for {@code DefaultMessageListenerContainer.cacheLevel} + * @return the current {@link org.springframework.integration.dsl.channel.MessageChannelSpec} + * @see org.springframework.jms.listener.DefaultMessageListenerContainer#CACHE_AUTO etc. + */ + public S cacheLevel(Integer cacheLevel) { + this.cacheLevel = cacheLevel; + return _this(); + } + + @Override + protected AbstractJmsChannel doGet() { + AbstractJmsChannel jmsChannel = super.doGet(); + if (this.cacheLevel != null) { + DirectFieldAccessor dfa = new DirectFieldAccessor(jmsChannel); + Object container = dfa.getPropertyValue("container"); + if (container instanceof DefaultMessageListenerContainer) { + ((DefaultMessageListenerContainer) container).setCacheLevel(this.cacheLevel); + } + + } + return jmsChannel; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPollableMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPollableMessageChannelSpec.java new file mode 100644 index 0000000..c2e0c09 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPollableMessageChannelSpec.java @@ -0,0 +1,133 @@ +/* + * Copyright 2014 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.jms; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; + +import org.springframework.beans.factory.BeanCreationException; +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; + +/** + * @author Artem Bilan + */ +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); + } + + @Override + protected S id(String id) { + this.jmsChannelFactoryBean.setBeanName(id); + return super.id(id); + } + + public S destination(String destination) { + this.jmsChannelFactoryBean.setDestinationName(destination); + return _this(); + } + + public S destination(Destination destination) { + this.jmsChannelFactoryBean.setDestination(destination); + return _this(); + } + + public S messageSelector(String messageSelector) { + this.jmsChannelFactoryBean.setMessageSelector(messageSelector); + return _this(); + } + + public S amqpMessageConverter(MessageConverter messageConverter) { + this.jmsChannelFactoryBean.setMessageConverter(messageConverter); + return _this(); + } + + public S deliveryPersistent(boolean deliveryPersistent) { + this.jmsChannelFactoryBean.setDeliveryPersistent(deliveryPersistent); + return _this(); + } + + public S explicitQosEnabled(boolean explicitQosEnabled) { + this.jmsChannelFactoryBean.setExplicitQosEnabled(explicitQosEnabled); + return _this(); + } + + public S messageIdEnabled(boolean messageIdEnabled) { + this.jmsChannelFactoryBean.setMessageIdEnabled(messageIdEnabled); + return _this(); + } + + public S messageTimestampEnabled(boolean messageTimestampEnabled) { + this.jmsChannelFactoryBean.setMessageTimestampEnabled(messageTimestampEnabled); + return _this(); + } + + public S priority(int priority) { + this.jmsChannelFactoryBean.setPriority(priority); + return _this(); + } + + public S timeToLive(long timeToLive) { + this.jmsChannelFactoryBean.setTimeToLive(timeToLive); + return _this(); + } + + 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(); + } + + 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-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPublishSubscribeMessageChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPublishSubscribeMessageChannelSpec.java new file mode 100644 index 0000000..fef4fcf --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPublishSubscribeMessageChannelSpec.java @@ -0,0 +1,52 @@ +/* + * Copyright 2014 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.jms; + +import javax.jms.ConnectionFactory; + +/** + * @author Artem Bilan + */ +public class JmsPublishSubscribeMessageChannelSpec + extends JmsMessageChannelSpec { + + JmsPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) { + super(connectionFactory); + this.jmsChannelFactoryBean.setPubSubDomain(true); + } + + public JmsPublishSubscribeMessageChannelSpec subscriptionDurable(boolean durable) { + this.jmsChannelFactoryBean.setSubscriptionDurable(durable); + return _this(); + } + + public JmsPublishSubscribeMessageChannelSpec durableSubscriptionName(String durableSubscriptionName) { + this.jmsChannelFactoryBean.setDurableSubscriptionName(durableSubscriptionName); + return _this(); + } + + public JmsPublishSubscribeMessageChannelSpec clientId(String clientId) { + this.jmsChannelFactoryBean.setClientId(clientId); + return _this(); + } + + public JmsPublishSubscribeMessageChannelSpec pubSubNoLocal(boolean pubSubNoLocal) { + this.jmsChannelFactoryBean.setPubSubNoLocal(pubSubNoLocal); + return _this(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/package-info.java new file mode 100644 index 0000000..07bf0a6 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides JMS Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.dsl.jms; diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 7ce93a8..8532eed 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -56,7 +56,6 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.aop.TargetSource; import org.springframework.aop.framework.Advised; -import org.springframework.aop.support.AopUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.ListableBeanFactory; @@ -81,18 +80,19 @@ import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.AbstractPollableChannel; -import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ChannelInterceptorAware; 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.IntegrationFlows; import org.springframework.integration.dsl.amqp.Amqp; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.jms.Jms; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.event.core.MessagingEvent; @@ -127,6 +127,8 @@ import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.stereotype.Component; import org.springframework.stereotype.Service; import org.springframework.test.annotation.DirtiesContext; @@ -157,7 +159,7 @@ public class IntegrationFlowTests { @Autowired @Qualifier("flow1QueueChannel") - private AbstractPollableChannel outputChannel; + private PollableChannel outputChannel; @Autowired private TestChannelInterceptor testChannelInterceptor; @@ -330,15 +332,18 @@ public class IntegrationFlowTests { @Test public void testPollingFlow() { + this.controlBus.send("@integerEndpoint.start()"); assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class)); - for (int i = 0; i < 10; i++) { - Message message = this.outputChannel.receive(5000); + for (int i = 0; i < 5; i++) { + Message message = this.outputChannel.receive(20000); assertNotNull(message); assertEquals("" + i, message.getPayload()); } - assertTrue(this.outputChannel.getChannelInterceptors().contains(this.testChannelInterceptor)); - assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(10)); + assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors() + .contains(this.testChannelInterceptor)); + assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(5)); + this.controlBus.send("@integerEndpoint.stop()"); } @Test @@ -807,7 +812,6 @@ public class IntegrationFlowTests { @Test public void testPriority() throws InterruptedException { - Message message = MessageBuilder.withPayload("1").setPriority(1).build(); this.priorityChannel.send(message); @@ -831,31 +835,31 @@ public class IntegrationFlowTests { Thread.sleep(2000); - Message receive = this.priorityReplyChannel.receive(1000); + Message receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("3", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("31", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("2", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("1", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("0", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("-1", receive.getPayload()); - receive = this.priorityReplyChannel.receive(1000); + receive = this.priorityReplyChannel.receive(2000); assertNotNull(receive); assertEquals("none", receive.getPayload()); } @@ -963,12 +967,19 @@ public class IntegrationFlowTests { return IntegrationFlows.from("controlBus").controlBus().get(); } + @Autowired + private javax.jms.ConnectionFactory jmsConnectionFactory; + @Bean public IntegrationFlow flow1() { - return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100, 1000))) + return IntegrationFlows.from(this.integerMessageSource(), + c -> c.poller(Pollers.fixedRate(1000, 2000)) + .id("integerEndpoint") + .autoStartup(false)) .fixedSubscriberChannel("integerChannel") .transform("payload.toString()") - .channel(MessageChannels.queue("flow1QueueChannel")) + .channel(Jms.pollableChannel("flow1QueueChannel", this.jmsConnectionFactory) + .destination("flow1QueueChannel")) .get(); } @@ -977,6 +988,13 @@ public class IntegrationFlowTests { return Pollers.fixedRate(500).get(); } + @Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME) + public TaskScheduler taskScheduler() { + ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); + threadPoolTaskScheduler.setPoolSize(100); + return threadPoolTaskScheduler; + } + @Bean public MessageChannel inputChannel() { return MessageChannels.direct().get(); @@ -1046,7 +1064,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) { return IntegrationFlows.from(MessageChannels.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup")) - .bridge(s -> s.poller(Pollers.fixedDelay(1000, 4000))) + .bridge(s -> s.poller(Pollers.fixedDelay(1000, 5000))) .channel(MessageChannels.queue("priorityReplyChannel")) .get(); }