Spring/spring-integration-extensions
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DSL: Add Jms factory

Browse Source 
* Add `Spec` implementations for `JmsChannel`s
* Fix the bug around `SourcePollingChannelAdapterFactoryBean` named bean registration
* Remove unused classes
* Add test for `Jms.pollableChannel`
This commit is contained in:
Artem Bilan
2014-06-10 18:42:23 +03:00
parent 36063ec8f2
commit 7394516a2d
14 changed files with 447 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
View File

@@ -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'.");
}

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java
View File

@@ -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) {

10
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java
View File

@@ -77,18 +77,16 @@ public abstract class Amqp {
return new AmqpOutboundEndpointSpec(endpoint, expectReply);
}
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S>
pollableChannel(ConnectionFactory connectionFactory) {
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S>pollableChannel(ConnectionFactory connectionFactory) {
return pollableChannel(null, connectionFactory);
}
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S>
pollableChannel(String id, ConnectionFactory connectionFactory) {
public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(String id,
ConnectionFactory connectionFactory) {
return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
}
public static
<S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
return channel(null, connectionFactory);
}

53
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java
View File

@@ -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;
}
}

95
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java
View File

@@ -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<T> extends AbstractFactoryBean<T> 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;
}
}
}

4
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides configuration classes of the Spring Integration Java DSL.
*/
package org.springframework.integration.dsl.config;

12
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java
View File

@@ -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()

52
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java Normal file
View File

@@ -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 <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(ConnectionFactory connectionFactory) {
return pollableChannel(null, connectionFactory);
}
public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(String id,
ConnectionFactory connectionFactory) {
return new JmsPollableMessageChannelSpec<S>(connectionFactory).id(id);
}
public static <S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
return channel(null, connectionFactory);
}
public static <S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(String id,
ConnectionFactory connectionFactory) {
return new JmsMessageChannelSpec<S>(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);
}
}

141
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java Normal file
View File

@@ -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<S extends JmsMessageChannelSpec<S>> extends JmsPollableMessageChannelSpec<S> {
private Integer cacheLevel;
JmsMessageChannelSpec(ConnectionFactory connectionFactory) {
super(new JmsChannelFactoryBean(true), connectionFactory);
}
public S containerType(Class<? extends AbstractMessageListenerContainer> 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;
}
}

133
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPollableMessageChannelSpec.java Normal file
View File

@@ -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<S extends JmsPollableMessageChannelSpec<S>>
extends MessageChannelSpec<S, AbstractJmsChannel> {
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();
}
}

52
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsPublishSubscribeMessageChannelSpec.java Normal file
View File

@@ -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> {
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();
}
}

4
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides JMS Components support for Spring Integration Java DSL.
*/
package org.springframework.integration.dsl.jms;

56
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
View File

@@ -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<String> 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();
}