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

Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP

Browse Source 
Conflicts:
	build.gradle
	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
	spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
	spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
	spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java
	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java
	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java
	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java
	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcInvalidConfigsTests.java
	spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
	spring-integration-jpa/src/test/java/org/springframework/integration/jpa/outbound/JpaOutboundGatewayIntegrationTests.java
	spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
	spring-integration-security/src/test/java/org/springframework/integration/security/SecurityTestUtils.java
	spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java

Resolved.
This commit is contained in:
Gary Russell
2013-12-06 14:23:23 -05:00
parent 116cedb3c6 43226da62a
commit feec3eaf84
55 changed files with 382 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -295,6 +295,9 @@ project('spring-integration-http') {
}
compile ("net.java.dev.rome:rome:1.0.0", optional)
testCompile project(":spring-integration-test")
// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)
compileJava.options.compilerArgs = ["${xLintArg},-deprecation"]
}
}

20
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.amqp.channel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageListener;
@@ -30,6 +31,7 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.support.MessageBuilder;
@@ -43,6 +45,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
@@ -51,11 +54,11 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
private final SimpleMessageListenerContainer container;
private volatile MessageDispatcher dispatcher;
private volatile AbstractDispatcher dispatcher;
private final boolean isPubSub;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container, AmqpTemplate amqpTemplate) {
this(channelName, container, amqpTemplate, false);
@@ -79,6 +82,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
if (this.dispatcher != null) {
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}
}
public boolean subscribe(MessageHandler handler) {
@@ -93,9 +99,13 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
public void onInit() throws Exception {
super.onInit();
this.dispatcher = this.createDispatcher();
if (this.dispatcher instanceof AbstractDispatcher) {
((AbstractDispatcher) this.dispatcher).setMaxSubscribers(this.maxSubscribers);
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(this.isPubSub ?
IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
}
this.setMaxSubscribers(this.maxSubscribers);
AmqpAdmin admin = new RabbitAdmin(this.container.getConnectionFactory());
Queue queue = this.initializeQueue(admin, this.channelName);
this.container.setQueues(queue);
@@ -110,7 +120,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
}
}
protected abstract MessageDispatcher createDispatcher();
protected abstract AbstractDispatcher createDispatcher();
protected abstract Queue initializeQueue(AmqpAdmin admin, String channelName);

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -20,7 +20,7 @@ import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -57,7 +57,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
}
@Override
protected MessageDispatcher createDispatcher() {
protected AbstractDispatcher createDispatcher() {
UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
return unicastingDispatcher;

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -23,8 +23,8 @@ import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
/**
* @author Mark Fisher
@@ -65,7 +65,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
}
@Override
protected MessageDispatcher createDispatcher() {
protected AbstractDispatcher createDispatcher() {
return new BroadcastingDispatcher(true);
}

10
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
View File

@@ -121,7 +121,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
private volatile Integer txSize;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public AmqpChannelFactoryBean() {
@@ -307,7 +307,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (this.exchange != null) {
pubsub.setExchange(this.exchange);
}
pubsub.setMaxSubscribers(this.maxSubscribers);
if (this.maxSubscribers != null) {
pubsub.setMaxSubscribers(this.maxSubscribers);
}
this.channel = pubsub;
}
else {
@@ -316,7 +318,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (StringUtils.hasText(this.queueName)) {
p2p.setQueueName(this.queueName);
}
p2p.setMaxSubscribers(this.maxSubscribers);
if (this.maxSubscribers != null) {
p2p.setMaxSubscribers(this.maxSubscribers);
}
this.channel = p2p;
}
}

17
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -16,12 +16,13 @@
package org.springframework.integration.amqp.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the 'channel' and 'publish-subscribe-channel' elements of the
@@ -29,6 +30,7 @@ import org.w3c.dom.Element;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
public class AmqpChannelParser extends AbstractChannelParser {
@@ -45,15 +47,10 @@ public class AmqpChannelParser extends AbstractChannelParser {
connectionFactory = "rabbitConnectionFactory";
}
builder.addPropertyReference("connectionFactory", connectionFactory);
if ("channel".equals(element.getLocalName())) {
builder.addPropertyValue("pubSub", false);
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
}
else if ("publish-subscribe-channel".equals(element.getLocalName())) {
builder.addPropertyValue("pubSub", true);
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
}
builder.addPropertyValue("pubSub", "publish-subscribe-channel".equals(element.getLocalName()));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "acknowledge-mode");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "advice-chain");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -171,12 +171,14 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
}
}
@SuppressWarnings("deprecation")
Object replyCorrelation = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_CORRELATION_HEADER);
if (replyCorrelation instanceof String) {
if (StringUtils.hasText((String) replyCorrelation)) {
headers.put(AmqpHeaders.SPRING_REPLY_CORRELATION, replyCorrelation);
}
}
@SuppressWarnings("deprecation")
Object replyToStack = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_REPLY_TO_HEADER);
if (replyToStack instanceof String) {
if (StringUtils.hasText((String) replyToStack)) {
@@ -195,6 +197,7 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
/**
* Extract user-defined headers from an AMQP MessageProperties instance.
*/
@SuppressWarnings("deprecation")
@Override
protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
Map<String, Object> headers = amqpMessageProperties.getHeaders();

11
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java
View File

@@ -16,8 +16,14 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
@@ -34,11 +40,6 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* @author Mark Fisher
* @author Artem Bilan

13
spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.channel;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -34,6 +35,8 @@ public class DirectChannel extends AbstractSubscribableChannel {
private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
private volatile Integer maxSubscribers;
/**
* Create a channel with default {@link RoundRobinLoadBalancingStrategy}
*/
@@ -64,6 +67,7 @@ public class DirectChannel extends AbstractSubscribableChannel {
* @param maxSubscribers
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
this.dispatcher.setMaxSubscribers(maxSubscribers);
}
@@ -72,4 +76,13 @@ public class DirectChannel extends AbstractSubscribableChannel {
return this.dispatcher;
}
@Override
protected void onInit() throws Exception {
super.onInit();
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -41,6 +42,7 @@ import org.springframework.util.ErrorHandler;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 1.0.3
*/
public class ExecutorChannel extends AbstractSubscribableChannel {
@@ -51,7 +53,7 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
private volatile boolean failover = true;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
private volatile LoadBalancingStrategy loadBalancingStrategy;
@@ -116,7 +118,10 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
}
this.dispatcher = new UnicastingDispatcher(this.executor);
this.dispatcher.setFailover(this.failover);
this.dispatcher.setMaxSubscribers(maxSubscribers);
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
}
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
if (this.loadBalancingStrategy != null) {
this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);
}

8
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -44,7 +45,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
private volatile int minSubscribers;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
@Override
public String getComponentType(){
@@ -147,7 +148,10 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
this.dispatcher.setIgnoreFailures(this.ignoreFailures);
this.dispatcher.setApplySequence(this.applySequence);
this.dispatcher.setMinSubscribers(this.minSubscribers);
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java
View File

@@ -20,10 +20,8 @@ import org.w3c.dom.Element;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
@@ -82,15 +80,8 @@ public abstract class AbstractChannelAdapterParser extends AbstractBeanDefinitio
if (parserContext.isNested()) {
return null;
}
String channelId = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(channelId)) {
parserContext.getReaderContext().error("The channel-adapter's 'id' attribute is required when no 'channel' "
+ "reference has been provided, because that 'id' would be used for the created channel.", element);
}
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
return channelId;
return IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
/**

44
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2013 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.
@@ -19,11 +19,7 @@ package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.PropertyValue;
import org.springframework.beans.PropertyValues;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
@@ -35,9 +31,10 @@ import org.springframework.util.xml.DomUtils;
/**
* Base class for channel parsers.
*
*
* @author Mark Fisher
* @author Dave Syer
* @author Artem Bilan
*/
public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser {
@@ -67,7 +64,7 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
beanDefinition.setSource(parserContext.extractSource(element));
return beanDefinition;
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.xml.AbstractBeanDefinitionParser#registerBeanDefinition(org.springframework.beans.factory.config.BeanDefinitionHolder, org.springframework.beans.factory.support.BeanDefinitionRegistry)
*/
@@ -89,37 +86,4 @@ public abstract class AbstractChannelParser extends AbstractBeanDefinitionParser
*/
protected abstract BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext);
protected void setMaxSubscribersProperty(ParserContext parserContext, BeanDefinitionBuilder builder, Element element, String channelInitializerPropertyName) {
String maxSubscribers = element.getAttribute("max-subscribers");
if (!StringUtils.hasText(maxSubscribers)) {
maxSubscribers = getDefaultMaxSubscribers(parserContext, channelInitializerPropertyName);
}
if (StringUtils.hasText(maxSubscribers)) {
builder.addPropertyValue("maxSubscribers", maxSubscribers);
}
}
protected String getDefaultMaxSubscribers(ParserContext parserContext, String channelInitializerPropertyName) {
String maxSubscribers = null;
BeanDefinition channelInitializer = parserContext.getRegistry().getBeanDefinition(
AbstractIntegrationNamespaceHandler.CHANNEL_INITIALIZER_BEAN_NAME);
if (channelInitializer != null) {
PropertyValues propertyValues = channelInitializer.getPropertyValues();
if (propertyValues != null) {
PropertyValue propertyValue = propertyValues
.getPropertyValue(channelInitializerPropertyName);
if (propertyValue != null) {
Object propertyValueValue = propertyValue.getValue();
if (propertyValueValue instanceof TypedStringValue) {
maxSubscribers = ((TypedStringValue) propertyValueValue).getValue();
}
else if (propertyValueValue instanceof String) {
maxSubscribers = (String) propertyValueValue;
}
}
}
}
return maxSubscribers;
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java
View File

@@ -96,10 +96,10 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
BeanDefinitionRegistry registry = parserContext.getRegistry();
if (registry instanceof ListableBeanFactory) {
alreadyRegistered = ((ListableBeanFactory) registry)
.containsBean(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME);
.containsBean(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME);
}
else {
alreadyRegistered = registry.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME);
alreadyRegistered = registry.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME);
}
if (!alreadyRegistered) {
ResourcePatternResolver resourceResolver =
@@ -115,7 +115,7 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
.genericBeanDefinition(PropertiesFactoryBean.class)
.addPropertyValue("locations", resources);
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME,
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME,
integrationPropertiesBuilder.getBeanDefinition());
}
catch (IOException e) {
@@ -146,8 +146,7 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
alreadyRegistered = parserContext.getRegistry().isBeanNameInUse(CHANNEL_INITIALIZER_BEAN_NAME);
}
if (!alreadyRegistered) {
String channelsAutoCreateExpression = "#{@" +IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME +
"['" + IntegrationProperties.CHANNELS_AUTOCREATE + "']}";
String channelsAutoCreateExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.CHANNELS_AUTOCREATE);
BeanDefinitionBuilder channelDef = BeanDefinitionBuilder.genericBeanDefinition(ChannelInitializer.class)
.addPropertyValue("autoCreate", channelsAutoCreateExpression);
BeanDefinitionHolder channelCreatorHolder = new BeanDefinitionHolder(channelDef.getBeanDefinition(), CHANNEL_INITIALIZER_BEAN_NAME);

43
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChannelInitializer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -19,15 +19,12 @@ import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.util.Assert;
/**
@@ -47,17 +44,12 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
public static String AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME = "$autoCreateChannelCandidates";
public static String CHANNEL_NAMES_ATTR = "channelNames";
private Log logger = LogFactory.getLog(this.getClass());
private volatile BeanFactory beanFactory;
private volatile boolean autoCreate = true;
private volatile int defaultMaxUnicastSubscribers = Integer.MAX_VALUE;
private volatile int defaultMaxBroadcastSubscribers = Integer.MAX_VALUE;
public void setAutoCreate(boolean autoCreate) {
this.autoCreate = autoCreate;
@@ -67,32 +59,6 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
this.beanFactory = beanFactory;
}
public int getDefaultMaxUnicastSubscribers() {
return defaultMaxUnicastSubscribers;
}
/**
* Set the default max-subscribers for all unicasting channels that don't have the
* attribute set on their dispatcher. Default {@link Integer#MAX_VALUE}.
* @param defaultMaxUnicastSubscribers
*/
public void setDefaultMaxUnicastSubscribers(int defaultMaxUnicastSubscribers) {
this.defaultMaxUnicastSubscribers = defaultMaxUnicastSubscribers;
}
public int getDefaultMaxBroadcastSubscribers() {
return defaultMaxBroadcastSubscribers;
}
/**
* Set the default max-subscribers for all broadcasting (pub-sub) channels that don't have the
* attribute set. Default {@link Integer#MAX_VALUE}.
* @param defaultMaxBroadcastSubscribers
*/
public void setDefaultMaxBroadcastSubscribers(int defaultMaxBroadcastSubscribers) {
this.defaultMaxBroadcastSubscribers = defaultMaxBroadcastSubscribers;
}
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.beanFactory, "'beanFactory' must not be null");
if (!autoCreate){
@@ -111,10 +77,7 @@ final class ChannelInitializer implements BeanFactoryAware, InitializingBean {
if (this.logger.isDebugEnabled()){
this.logger.debug("Auto-creating channel '" + channelName + "' as DirectChannel");
}
RootBeanDefinition messageChannel = new RootBeanDefinition();
messageChannel.setBeanClass(DirectChannel.class);
BeanDefinitionHolder messageChannelHolder = new BeanDefinitionHolder(messageChannel, channelName);
BeanDefinitionReaderUtils.registerBeanDefinition(messageChannelHolder, (BeanDefinitionRegistry) this.beanFactory);
IntegrationNamespaceUtils.autoCreateDirectChannel(channelName, (BeanDefinitionRegistry) this.beanFactory);
}
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessor.java
View File

@@ -33,6 +33,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
/**
* A {@link BeanFactoryPostProcessor} implementation that provides default beans for the error handling and task
@@ -153,7 +154,8 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
}
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler");
schedulerBuilder.addPropertyValue("poolSize", 10);
String taskSchedulerPoolSizeExpression = IntegrationProperties.getExpressionFor(IntegrationProperties.TASKSCHEDULER_POOLSIZE);
schedulerBuilder.addPropertyValue("poolSize", taskSchedulerPoolSizeExpression);
schedulerBuilder.addPropertyValue("threadNamePrefix", "task-scheduler-");
schedulerBuilder.addPropertyValue("rejectedExecutionHandler", new CallerRunsPolicy());
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(

27
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -70,21 +70,6 @@ public abstract class IntegrationNamespaceUtils {
public static final String AUTO_STARTUP = "auto-startup";
public static final String PHASE = "phase";
/**
* Property name on ChannelInitializer used to configure the default max subscribers for
* unicast channels.
*/
public static String DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME = "defaultMaxUnicastSubscribers";
/**
* Property name on ChannelInitializer used to configure the default max subscribers for
* broadcast channels.
*/
public static String DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME = "defaultMaxBroadcastSubscribers";
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
@@ -507,10 +492,16 @@ public abstract class IntegrationNamespaceUtils {
parserContext.getReaderContext().error("The channel-adapter's 'id' attribute is required when no 'channel' "
+ "reference has been provided, because that 'id' would be used for the created channel.", element);
}
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
autoCreateDirectChannel(channelId, parserContext.getRegistry());
return channelId;
}
public static void autoCreateDirectChannel(String channelName, BeanDefinitionRegistry registry) {
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelName);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, registry);
}
}

14
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -18,6 +18,8 @@ package org.springframework.integration.config.xml;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
@@ -29,7 +31,6 @@ import org.springframework.integration.channel.RendezvousChannel;
import org.springframework.integration.store.MessageGroupQueue;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parser for the &lt;channel&gt; element.
@@ -38,6 +39,7 @@ import org.w3c.dom.Element;
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*/
public class PointToPointChannelParser extends AbstractChannelParser {
@@ -121,11 +123,6 @@ public class PointToPointChannelParser extends AbstractChannelParser {
else if (dispatcherElement == null) {
// configure the default DirectChannel with a RoundRobinLoadBalancingStrategy
builder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
String maxSubscribers = this.getDefaultMaxSubscribers(parserContext,
IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
if (maxSubscribers != null) {
builder.addPropertyValue("maxSubscribers", maxSubscribers);
}
}
else {
// configure either an ExecutorChannel or DirectChannel based on existence of 'task-executor'
@@ -144,8 +141,7 @@ public class PointToPointChannelParser extends AbstractChannelParser {
builder.addConstructorArgValue(null);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "failover");
this.setMaxSubscribersProperty(parserContext, builder, dispatcherElement,
IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, dispatcherElement, "max-subscribers");
}
return builder;
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java
View File

@@ -28,6 +28,7 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class PublishSubscribeChannelParser extends AbstractChannelParser {
@@ -42,8 +43,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-failures");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
this.setMaxSubscribersProperty(parserContext, builder, element,
IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "min-subscribers");
return builder;
}

22
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java
View File

@@ -49,9 +49,7 @@ public abstract class IntegrationContextUtils {
public static final String INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME = "integrationHeaderChannelRegistry";
public static final String INTEGRATION_PROPERTIES_BEAN_NAME = "integrationProperties";
private static final Properties EMPTY_PROPERTIES = new Properties();
public static final String INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME = "integrationGlobalProperties";
/**
* Return the {@link MetadataStore} bean whose name is "metadataStore".
@@ -113,21 +111,23 @@ public abstract class IntegrationContextUtils {
}
/**
* @return the global {@link IntegrationContextUtils#INTEGRATION_PROPERTIES_BEAN_NAME}
* @return the global {@link IntegrationContextUtils#INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME}
* bean from provided {@code #beanFactory}, which represents the merged
* properties values from all 'META-INF/spring.integration.default.properties'
* and 'META-INF/spring.integration.properties'.
* May return {@link IntegrationContextUtils#EMPTY_PROPERTIES} if there is no
* {@link IntegrationContextUtils#INTEGRATION_PROPERTIES_BEAN_NAME} bean within
* Or user-defined {@link Properties} bean.
* May return only {@link IntegrationProperties#defaults()} if there is no
* {@link IntegrationContextUtils#INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME} bean within
* provided {@code #beanFactory} or provided {@code #beanFactory} is null.
*/
public static Properties getIntegrationProperties(BeanFactory beanFactory) {
Properties properties = null;
Properties properties = new Properties();
properties.putAll(IntegrationProperties.defaults());
if (beanFactory != null) {
properties = getBeanOfType(beanFactory, INTEGRATION_PROPERTIES_BEAN_NAME, Properties.class);
}
if (properties == null) {
properties = EMPTY_PROPERTIES;
Properties userProperties = getBeanOfType(beanFactory, INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME, Properties.class);
if (userProperties != null) {
properties.putAll(userProperties);
}
}
return properties;
}

17
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -59,6 +60,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
*/
protected final Log logger = LogFactory.getLog(getClass());
private final ConversionService defaultConversionService = new DefaultConversionService();
private volatile String beanName;
private volatile String componentName;
@@ -67,6 +70,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
private volatile TaskScheduler taskScheduler;
private volatile Properties integrationProperties = IntegrationProperties.defaults();
private volatile ConversionService conversionService;
private volatile ApplicationContext applicationContext;
@@ -102,6 +107,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
public final void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
this.beanFactory = beanFactory;
this.integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
}
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@@ -176,7 +182,16 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
* @see IntegrationContextUtils#getIntegrationProperties
*/
protected Properties getIntegrationProperties() {
return IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
return this.integrationProperties;
}
/**
* @param key Integration property.
* @param tClass the class to convert a value of Integration property.
* @return the value of the Integration property converted to the provide type.
*/
protected <T> T getIntegrationProperty(String key, Class<T> tClass) {
return this.defaultConversionService.convert(this.integrationProperties.getProperty(key), tClass);
}
@Override

88
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java
View File

@@ -16,16 +16,96 @@
package org.springframework.integration.context;
import java.io.IOException;
import java.util.Properties;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
/**
* Convention Enumeration to represent keys from 'META-INF/spring.integration.properties'.
* Utility class to encapsulate infrastructure Integration properties constants and
* their default values from resources 'META-INF/spring.integration.default.properties'.
*
* @author Artem Bilan
* @since 3.0
*/
public interface IntegrationProperties {
public final class IntegrationProperties {
String LATE_REPLY_LOGGING_LEVEL = "messagingTemplate.lateReply.logging.level";
/**
* Specifies whether to allow create automatically {@link org.springframework.integration.channel.DirectChannel}
* beans for non-declared channels or not.
*/
public static final String CHANNELS_AUTOCREATE = "channels.autoCreate";
String CHANNELS_AUTOCREATE = "channels.autoCreate";
/**
* Specifies the value for {@link org.springframework.integration.dispatcher.UnicastingDispatcher#maxSubscribers}
* in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.ExecutorChannel}),
* if the attribute {@code max-subscribers} isn't configured on the channel component.
*/
public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = "channels.maxUnicastSubscribers";
/**
* Specifies the value for {@link org.springframework.integration.dispatcher.BroadcastingDispatcher#maxSubscribers}
* in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.PublishSubscribeChannel}),
* if the attribute {@code max-subscribers} isn't configured on the channel component.
*/
public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = "channels.maxBroadcastSubscribers";
/**
* Specifies the value of {@link org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler#poolSize}
* for {@code taskScheduler} bean initialized but Integration infrastructure.
* @see org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor#registerTaskScheduler
*/
public static final String TASKSCHEDULER_POOLSIZE = "taskScheduler.poolSize";
// TODO public static final String LATE_REPLY_LOGGING_LEVEL = "messagingTemplate.lateReply.logging.level";
private static Properties defaults;
static {
String resourcePattern = "classpath*:META-INF/spring.integration.default.properties";
try {
ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver(IntegrationProperties.class.getClassLoader());
Resource[] defaultResources = resourceResolver.getResources(resourcePattern);
PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
propertiesFactoryBean.setLocations(defaultResources);
propertiesFactoryBean.afterPropertiesSet();
defaults = propertiesFactoryBean.getObject();
}
catch (IOException e) {
throw new IllegalStateException("Can't load '" + resourcePattern + "' resources.", e);
}
}
/**
* @return {@link Properties} with default values for Integration properties
* from resources 'META-INF/spring.integration.default.properties'.
*/
public static Properties defaults() {
return defaults;
}
/**
* Build the bean property definition expression to resolve the value
* from Integration properties within the bean building phase.
*
* @param key the Integration property key.
* @return the bean property definition expression.
* @throws IllegalArgumentException if provided {@code key} isn't an Integration property.
*/
public static String getExpressionFor(String key) {
if (defaults.containsKey(key)) {
return "#{T(" + IntegrationContextUtils.class.getName() + ").getIntegrationProperties(beanFactory).getProperty('" + key + "')}";
}
else {
throw new IllegalArgumentException("The provided key [" + key + "] isn't the one of Integration properties: " + defaults.keySet());
}
}
private IntegrationProperties() {
}
}

4
spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties
View File

@@ -1,2 +1,4 @@
channels.autoCreate=true
messagingTemplate.lateReply.logging.level=warn
channels.maxUnicastSubscribers=0x7fffffff
channels.maxBroadcastSubscribers=0x7fffffff
taskScheduler.poolSize=10

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersDefaultConfigurationTests-context.xml
View File

@@ -7,6 +7,8 @@
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:service-activator input-channel="autoCreateChannel" expression="'foo'"/>
<int:channel id="defaultChannel" />
<int:channel id="defaultChannel2">

19
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersOverrideDefaultTests-context.xml
View File

@@ -1,15 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
<bean id="channelInitializer" class="org.springframework.integration.config.xml.ChannelInitializer">
<property name="autoCreate" value="true" />
<property name="defaultMaxUnicastSubscribers" value="456" />
<property name="defaultMaxBroadcastSubscribers" value="789" />
</bean>
<util:properties id="integrationGlobalProperties">
<prop key="channels.maxUnicastSubscribers">456</prop>
<prop key="channels.maxBroadcastSubscribers">789</prop>
</util:properties>
<import resource="DispatcherMaxSubscribersDefaultConfigurationTests-context.xml" />

21
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DispatcherMaxSubscribersTests.java
View File

@@ -23,11 +23,15 @@ import org.springframework.integration.test.util.TestUtils;
/**
* @author Gary Russell
* @author Artem Bilan
* @since 2.2
*
*/
public abstract class DispatcherMaxSubscribersTests {
@Autowired
private MessageChannel autoCreateChannel;
@Autowired
private MessageChannel defaultChannel;
@@ -54,20 +58,17 @@ public abstract class DispatcherMaxSubscribersTests {
}
protected void doTestUnicast(int val1, int val2, int val3, int val4, int val5) {
Integer defaultMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(defaultChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer autoCreateMax = TestUtils.getPropertyValue(autoCreateChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val1, autoCreateMax.intValue());
Integer defaultMax = TestUtils.getPropertyValue(defaultChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val1, defaultMax.intValue());
Integer defaultMax2 = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(defaultChannel2, "dispatcher"), "maxSubscribers", Integer.class);
Integer defaultMax2 = TestUtils.getPropertyValue(defaultChannel2, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val2, defaultMax2.intValue());
Integer explicitMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(explicitChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer explicitMax = TestUtils.getPropertyValue(explicitChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val3, explicitMax.intValue());
Integer execMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(executorChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer execMax = TestUtils.getPropertyValue(executorChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val4, execMax.intValue());
Integer explicitExecMax = TestUtils.getPropertyValue(
TestUtils.getPropertyValue(explicitExecutorChannel, "dispatcher"), "maxSubscribers", Integer.class);
Integer explicitExecMax = TestUtils.getPropertyValue(explicitExecutorChannel, "dispatcher.maxSubscribers", Integer.class);
assertEquals(val5, explicitExecMax.intValue());
}

14
spring-integration-core/src/test/java/org/springframework/integration/context/IntegrationContextTests.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.context;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.util.Properties;
@@ -26,6 +25,8 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -38,17 +39,22 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class IntegrationContextTests {
@Autowired
@Qualifier(IntegrationContextUtils.INTEGRATION_PROPERTIES_BEAN_NAME)
@Qualifier(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME)
private Properties integrationProperties;
@Autowired
@Qualifier("fooService")
private IntegrationObjectSupport serviceActivator;
@Autowired
private ThreadPoolTaskScheduler taskScheduler;
@Test
public void testIntegrationContextComponents() {
assertEquals("error", this.integrationProperties.get(IntegrationProperties.LATE_REPLY_LOGGING_LEVEL));
assertSame(this.integrationProperties, this.serviceActivator.getIntegrationProperties());
//TODO INT-3005 assertEquals("error", this.integrationProperties.get(IntegrationProperties.LATE_REPLY_LOGGING_LEVEL));
assertEquals("20", this.integrationProperties.get(IntegrationProperties.TASKSCHEDULER_POOLSIZE));
assertEquals(this.integrationProperties, this.serviceActivator.getIntegrationProperties());
assertEquals(20, TestUtils.getPropertyValue(this.taskScheduler, "poolSize"));
}
}

3
spring-integration-core/src/test/resources/META-INF/spring.integration.properties
View File

@@ -1,2 +1,5 @@
#channels.autoCreate=false
#channels.maxUnicastSubscribers=1
#channels.maxBroadcastSubscribers=1
messagingTemplate.lateReply.logging.level=error
taskScheduler.poolSize=20

4
spring-integration-event/src/main/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducer.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.event.inbound;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
@@ -31,7 +32,6 @@ import org.springframework.integration.endpoint.ExpressionMessageProducerSupport
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* An inbound Channel Adapter that implements {@link ApplicationListener} and
@@ -71,7 +71,7 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
*/
public void setEventTypes(Class<? extends ApplicationEvent>... eventTypes) {
Set<Class<? extends ApplicationEvent>> eventSet = new HashSet<Class<? extends ApplicationEvent>>(
CollectionUtils.<Class<? extends ApplicationEvent>> arrayToList(eventTypes));
Arrays.asList(eventTypes));
eventSet.remove(null);
this.eventTypes = (eventSet.size() > 0 ? eventSet : null);

1
spring-integration-file/src/test/java/org/springframework/integration/file/FileOutboundGatewayIntegrationTests.java
View File

@@ -39,7 +39,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

19
spring-integration-file/src/test/java/org/springframework/integration/file/recursive/FileInboundChannelAdapterWithRecursiveDirectoryTests.java
View File

@@ -16,9 +16,19 @@
package org.springframework.integration.file.recursive;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
@@ -27,15 +37,6 @@ import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
/**
* @author Iwein Fuld
* @author Gunnar Hillert

8
spring-integration-http/src/main/java/org/springframework/integration/http/converter/MultipartAwareFormHttpMessageConverter.java
View File

@@ -40,10 +40,11 @@ import org.springframework.web.multipart.MultipartFile;
* An {@link HttpMessageConverter} implementation that delegates to an instance of
* {@link XmlAwareFormHttpMessageConverter} while adding the capability to <i>read</i>
* <code>multipart/form-data</code> content in an HTTP request.
*
*
* @author Mark Fisher
* @since 2.0
*/
@SuppressWarnings("deprecation")
public class MultipartAwareFormHttpMessageConverter implements HttpMessageConverter<MultiValueMap<String, ?>> {
private volatile MultipartFileReader<?> multipartFileReader = new DefaultMultipartFileReader();
@@ -66,10 +67,12 @@ public class MultipartAwareFormHttpMessageConverter implements HttpMessageConver
this.multipartFileReader = multipartFileReader;
}
@Override
public List<MediaType> getSupportedMediaTypes() {
return this.wrappedConverter.getSupportedMediaTypes();
}
@Override
public boolean canRead(Class<?> clazz, MediaType mediaType) {
if (!(MultiValueMap.class.isAssignableFrom(clazz) || byte[].class.isAssignableFrom(clazz))) {
return false;
@@ -83,10 +86,12 @@ public class MultipartAwareFormHttpMessageConverter implements HttpMessageConver
}
}
@Override
public boolean canWrite(Class<?> clazz, MediaType mediaType) {
return this.wrappedConverter.canWrite(clazz, mediaType);
}
@Override
public MultiValueMap<String, ?> read(Class<? extends MultiValueMap<String, ?>> clazz,
HttpInputMessage inputMessage) throws IOException, HttpMessageNotReadableException {
@@ -118,6 +123,7 @@ public class MultipartAwareFormHttpMessageConverter implements HttpMessageConver
return resultMap;
}
@Override
public void write(MultiValueMap<String, ?> map, MediaType contentType, HttpOutputMessage outputMessage)
throws IOException, HttpMessageNotWritableException {
this.wrappedConverter.write(map, contentType, outputMessage);

7
spring-integration-http/src/main/java/org/springframework/integration/http/inbound/IntegrationRequestMappingHandlerMapping.java
View File

@@ -116,30 +116,37 @@ public final class IntegrationRequestMappingHandlerMapping extends RequestMappin
org.springframework.web.bind.annotation.RequestMapping requestMappingAnnotation =
new org.springframework.web.bind.annotation.RequestMapping() {
@Override
public String[] value() {
return requestMapping.getPathPatterns();
}
@Override
public RequestMethod[] method() {
return requestMapping.getRequestMethods();
}
@Override
public String[] params() {
return requestMapping.getParams();
}
@Override
public String[] headers() {
return requestMapping.getHeaders();
}
@Override
public String[] consumes() {
return requestMapping.getConsumes();
}
@Override
public String[] produces() {
return requestMapping.getProduces();
}
@Override
public Class<? extends Annotation> annotationType() {
return org.springframework.web.bind.annotation.RequestMapping.class;
}

9
spring-integration-http/src/main/java/org/springframework/integration/http/support/DefaultHttpHeaderMapper.java
View File

@@ -21,8 +21,8 @@ import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.text.DateFormat;
import java.text.MessageFormat;
import java.text.SimpleDateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -30,9 +30,9 @@ import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.Locale;
import java.util.TimeZone;
import org.apache.commons.logging.Log;
@@ -265,6 +265,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
private volatile String userDefinedHeaderPrefix = "X-";
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
@@ -326,6 +327,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
* Depending on which type of adapter is using this mapper, the HttpHeaders might be
* for an HTTP request (outbound adapter) or for an HTTP response (inbound adapter).
*/
@Override
public void fromHeaders(MessageHeaders headers, HttpHeaders target) {
if (logger.isDebugEnabled()){
logger.debug(MessageFormat.format("outboundHeaderNames={0}", CollectionUtils.arrayToList(outboundHeaderNames)));
@@ -356,6 +358,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
* Depending on which type of adapter is using this mapper, the HttpHeaders might be
* from an HTTP request (inbound adapter) or from an HTTP response (outbound adapter).
*/
@Override
public Map<String, Object> toHeaders(HttpHeaders source) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("inboundHeaderNames={0}", CollectionUtils.arrayToList(inboundHeaderNames)));
@@ -389,6 +392,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
return target;
}
@Override
public void afterPropertiesSet() throws Exception {
if (this.beanFactory != null){
this.conversionService = IntegrationContextUtils.getConversionService(this.beanFactory);
@@ -901,6 +905,7 @@ public class DefaultHttpHeaderMapper implements HeaderMapper<HttpHeaders>, BeanF
return source.getIfNoneMatch();
}
else if (IF_MODIFIED_SINCE.equalsIgnoreCase(name)) {
@SuppressWarnings("deprecation")
long modifiedSince = source.getIfNotModifiedSince();
return (modifiedSince > -1) ? modifiedSince : null;
}

1
spring-integration-http/src/test/java/org/springframework/integration/http/HttpProxyScenarioTests.java
View File

@@ -107,6 +107,7 @@ public class HttpProxyScenarioTests {
RestTemplate template = Mockito.spy(new RestTemplate());
Mockito.doAnswer(new Answer<ResponseEntity<?>>() {
@SuppressWarnings("deprecation")
@Override
public ResponseEntity<?> answer(InvocationOnMock invocation) throws Throwable {
URI uri = (URI) invocation.getArguments()[0];

6
spring-integration-http/src/test/java/org/springframework/integration/http/config/HttpOutboundChannelAdapterParserTests.java
View File

@@ -16,9 +16,9 @@
package org.springframework.integration.http.config;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -28,6 +28,7 @@ import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -167,7 +168,6 @@ public class HttpOutboundChannelAdapterParserTests {
}
@Test
@SuppressWarnings("uchecked")
public void withUrlAndTemplate() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(this.withUrlAndTemplate);
RestTemplate restTemplate =
@@ -257,10 +257,12 @@ public class HttpOutboundChannelAdapterParserTests {
public static class StubErrorHandler implements ResponseErrorHandler {
@Override
public boolean hasError(ClientHttpResponse response) throws IOException {
return false;
}
@Override
public void handleError(ClientHttpResponse response) throws IOException {
}
}

5
spring-integration-http/src/test/java/org/springframework/integration/http/support/DefaultHttpHeaderMapperFromMessageInboundTests.java
View File

@@ -18,8 +18,8 @@ package org.springframework.integration.http.support;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.net.URI;
import java.text.ParseException;
@@ -33,10 +33,10 @@ import java.util.Locale;
import java.util.Map;
import org.junit.Test;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.ConversionServiceFactory;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.http.HttpHeaders;
@@ -577,6 +577,7 @@ public class DefaultHttpHeaderMapperFromMessageInboundTests {
public static class TestClassConverter implements Converter<TestClass, String>{
@Override
public String convert(TestClass source) {
return "TestClass.class";
}

4
spring-integration-http/src/test/java/org/springframework/integration/http/support/DefaultHttpHeaderMapperFromMessageOutboundTests.java
View File

@@ -386,6 +386,7 @@ public class DefaultHttpHeaderMapperFromMessageOutboundTests {
// If-Modified-Since tests
@SuppressWarnings("deprecation")
@Test
public void validateIfModifiedSinceAsNumber() throws ParseException{
HeaderMapper<HttpHeaders> mapper = DefaultHttpHeaderMapper.outboundMapper();
@@ -399,6 +400,7 @@ public class DefaultHttpHeaderMapperFromMessageOutboundTests {
assertEquals(simpleDateFormat.parse("Thu, 01 Jan 1970 03:25:45 GMT").getTime(), headers.getIfNotModifiedSince());
}
@SuppressWarnings("deprecation")
@Test
public void validateIfModifiedSinceAsString() throws ParseException{
HeaderMapper<HttpHeaders> mapper = DefaultHttpHeaderMapper.outboundMapper();
@@ -411,6 +413,7 @@ public class DefaultHttpHeaderMapperFromMessageOutboundTests {
assertEquals(simpleDateFormat.parse("Thu, 01 Jan 1970 03:25:45 GMT").getTime(), headers.getIfNotModifiedSince());
}
@SuppressWarnings("deprecation")
@Test
public void validateIfModifiedSinceAsDate() throws ParseException{
HeaderMapper<HttpHeaders> mapper = DefaultHttpHeaderMapper.outboundMapper();
@@ -673,6 +676,7 @@ public class DefaultHttpHeaderMapperFromMessageOutboundTests {
assertEquals(0, messageHeaders.size());
}
@SuppressWarnings("deprecation")
public void testInt2995IfModifiedSince() throws Exception{
Date ifModifiedSince = new Date();
SimpleDateFormat dateFormat = new SimpleDateFormat("EEE MMM dd HH:mm:ss yyyy", Locale.US);

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.integration.ip.tcp.connection;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
@@ -47,10 +48,11 @@ public class TcpConnectionEventListeningMessageProducer extends MessageProducerS
public void setEventTypes(Class<? extends TcpConnectionEvent>[] eventTypes) {
Assert.notEmpty(eventTypes, "at least one event type is required");
Set<Class<? extends TcpConnectionEvent>> eventTypeSet = new HashSet<Class<? extends TcpConnectionEvent>>();
eventTypeSet.addAll(CollectionUtils.<Class<TcpConnectionEvent>> arrayToList(eventTypes));
eventTypeSet.addAll(Arrays.asList(eventTypes));
this.eventTypes = eventTypeSet;
}
@Override
public void onApplicationEvent(TcpConnectionEvent event) {
if (this.isRunning()) {
if (CollectionUtils.isEmpty(this.eventTypes)) {

6
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -31,6 +31,7 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
@@ -308,6 +309,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
return null;
}
@SuppressWarnings("deprecation")
@Override
@ManagedAttribute
public long getMessageCount() {
@@ -365,6 +367,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
final String groupKey = getKey(groupId);
final String messageId = getKey(message.getHeaders().getId());
@SuppressWarnings("deprecation")
boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(Query.GROUP_EXISTS), groupKey, region) < 1;
final Timestamp updatedDate = new Timestamp(System.currentTimeMillis());
@@ -402,18 +405,21 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
}
@SuppressWarnings("deprecation")
@Override
@ManagedAttribute
public int getMessageGroupCount() {
return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_GROUPS), region);
}
@SuppressWarnings("deprecation")
@Override
@ManagedAttribute
public int getMessageCountForAllMessageGroups() {
return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUPS), region);
}
@SuppressWarnings("deprecation")
@Override
@ManagedAttribute
public int messageGroupSize(Object groupId) {

13
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
View File

@@ -220,6 +220,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* Method not implemented.
* @throws UnsupportedOperationException
*/
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, final int sequenceNumber) {
throw new UnsupportedOperationException("Not implemented");
}
@@ -367,6 +368,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
*
* @throws Exception
*/
@Override
public void afterPropertiesSet() throws Exception {
Assert.state(jdbcTemplate != null, "A DataSource or JdbcTemplate must be provided");
Assert.notNull(this.channelMessageStoreQueryProvider, "A channelMessageStoreQueryProvider must be provided.");
@@ -391,6 +393,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* @param groupId the group id to store the message under
* @param message a message
*/
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
@@ -408,6 +411,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
final byte[] messageBytes = serializer.convert(result);
jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getCreateMessageQuery()), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){
logger.debug("Inserting message with id key=" + messageId);
@@ -427,6 +431,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* Method not implemented.
* @throws UnsupportedOperationException
*/
@Override
public void completeGroup(Object groupId) {
throw new UnsupportedOperationException("Not implemented");
}
@@ -524,6 +529,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
/**
* Not fully used. Only wraps the provided group id.
*/
@Override
public MessageGroup getMessageGroup(Object groupId) {
return new SimpleMessageGroup(groupId);
}
@@ -561,6 +567,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* Method not implemented.
* @throws UnsupportedOperationException
*/
@Override
public Iterator<MessageGroup> iterator() {
throw new UnsupportedOperationException("Not implemented");
}
@@ -569,6 +576,8 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* Returns the number of messages persisted for the specified channel id (groupId)
* and the specified region ({@link #setRegion(String)}).
*/
@Override
@SuppressWarnings("deprecation")
@ManagedAttribute
public int messageGroupSize(Object groupId) {
final String key = getKey(groupId);
@@ -579,6 +588,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* Polls the database for a new message that is persisted for the given
* group id which represents the channel identifier.
*/
@Override
public Message<?> pollMessageFromGroup(Object groupId) {
final String key = getKey(groupId);
@@ -600,6 +610,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
* @param messageToRemove The message to remove
*
*/
@Override
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
this.doRemoveMessageFromGroup(groupId, messageToRemove);
@@ -661,11 +672,13 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
/**
* Will remove all messages from the message channel.
*/
@Override
public void removeMessageGroup(Object groupId) {
final String groupKey = getKey(groupId);
jdbcTemplate.update(getQuery(channelMessageStoreQueryProvider.getDeleteMessageGroupQuery()), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){
logger.debug("Marking messages with group key=" + groupKey);

13
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java
View File

@@ -94,18 +94,22 @@ public class JdbcPollingChannelAdapterIntegrationTests {
"select * from item where status=:status");
adapter.setSelectSqlParameterSource(new SqlParameterSource() {
@Override
public boolean hasValue(String name) {
return "status".equals(name);
}
@Override
public Object getValue(String name) throws IllegalArgumentException {
return 2;
}
@Override
public String getTypeName(String name) {
return null;
}
@Override
public int getSqlType(String name) {
return Types.INTEGER;
}
@@ -162,12 +166,14 @@ public class JdbcPollingChannelAdapterIntegrationTests {
assertEquals("Wrong id", 1, item.getId());
assertEquals("Wrong status", 2, item.getStatus());
@SuppressWarnings("deprecation")
int countOfStatusTwo = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 2");
assertEquals(
"Status not updated incorect number of rows with status 2", 0,
countOfStatusTwo);
@SuppressWarnings("deprecation")
int countOfStatusTen = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 10");
assertEquals(
@@ -198,12 +204,14 @@ public class JdbcPollingChannelAdapterIntegrationTests {
assertEquals("Wrong id", 1, item.getId());
assertEquals("Wrong status", 2, item.getStatus());
@SuppressWarnings("deprecation")
int countOfStatusTwo = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 2");
assertEquals(
"Status not updated incorect number of rows with status 2", 0,
countOfStatusTwo);
@SuppressWarnings("deprecation")
int countOfStatusTen = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 10");
assertEquals(
@@ -237,12 +245,14 @@ public class JdbcPollingChannelAdapterIntegrationTests {
assertEquals("Wrong id", 2, item.getId());
assertEquals("Wrong status", 2, item.getStatus());
@SuppressWarnings("deprecation")
int countOfStatusTwo = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 2");
assertEquals(
"Status not updated incorect number of rows with status 2", 2,
countOfStatusTwo);
@SuppressWarnings("deprecation")
int countOfStatusTen = this.jdbcTemplate
.queryForInt("select count(*) from copy where status = 10");
assertEquals(
@@ -275,12 +285,14 @@ public class JdbcPollingChannelAdapterIntegrationTests {
assertEquals("Wrong id", 2, item.getId());
assertEquals("Wrong status", 2, item.getStatus());
@SuppressWarnings("deprecation")
int countOfStatusTwo = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 2");
assertEquals(
"Status not updated incorect number of rows with status 2", 0,
countOfStatusTwo);
@SuppressWarnings("deprecation")
int countOfStatusTen = this.jdbcTemplate
.queryForInt("select count(*) from item where status = 10");
assertEquals(
@@ -328,6 +340,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
private static class ItemRowMapper implements RowMapper<Item> {
@Override
public Item mapRow(ResultSet rs, int rowNum) throws SQLException {
Item item = new Item();
item.setId(rs.getInt(1));

7
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcOutboundGatewayWithNamespaceIntegrationTests.java
View File

@@ -31,15 +31,16 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.jdbc.storedproc.CreateUser;
import org.springframework.integration.jdbc.storedproc.User;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

3
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcOutboundGatewayParserTests.java
View File

@@ -25,10 +25,12 @@ import javax.sql.DataSource;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.jdbc.JdbcOutboundGateway;
@@ -38,7 +40,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.core.MessagingTemplate;
/**
* @author Dave Syer

1
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/StoredProcInvalidConfigsTests.java
View File

@@ -20,6 +20,7 @@ import java.io.ByteArrayInputStream;
import java.util.Properties;
import org.junit.Test;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;

10
spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
View File

@@ -20,9 +20,11 @@ import javax.jms.MessageListener;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
@@ -51,7 +53,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
private volatile boolean initialized;
private volatile int maxSubscribers = Integer.MAX_VALUE;
private volatile Integer maxSubscribers;
public SubscribableJmsChannel(AbstractMessageListenerContainer container, JmsTemplate jmsTemplate) {
super(jmsTemplate);
@@ -105,6 +107,12 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
this.dispatcher = unicastingDispatcher;
}
if (this.maxSubscribers == null) {
this.maxSubscribers = this.getIntegrationProperty(isPubSub ?
IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS :
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS,
Integer.class);
}
this.dispatcher.setMaxSubscribers(this.maxSubscribers);
}

8
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java
View File

@@ -18,12 +18,13 @@ package org.springframework.integration.jms.config;
import javax.jms.Session;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the 'channel' and 'publish-subscribe-channel' elements of the
@@ -58,12 +59,13 @@ public class JmsChannelParser extends AbstractChannelParser {
builder.addPropertyReference("connectionFactory", connectionFactory);
if ("channel".equals(element.getLocalName())) {
this.parseDestination(element, parserContext, builder, "queue");
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_UNICAST_SUBSCRIBERS_PROPERTY_NAME);
}
else if ("publish-subscribe-channel".equals(element.getLocalName())) {
this.parseDestination(element, parserContext, builder, "topic");
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
String containerType = element.getAttribute(CONTAINER_TYPE_ATTRIBUTE);
String containerClass = element.getAttribute(CONTAINER_CLASS_ATTRIBUTE);
if (!StringUtils.hasText(containerClass)) {

8
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -32,6 +32,7 @@ import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
@@ -61,6 +62,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
private final AbstractDispatcher dispatcher = new BroadcastingDispatcher(true);
private volatile Integer maxSubscribers;
private volatile boolean initialized;
// defaults
@@ -97,6 +100,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
* @param maxSubscribers
*/
public void setMaxSubscribers(int maxSubscribers) {
this.maxSubscribers = maxSubscribers;
this.dispatcher.setMaxSubscribers(maxSubscribers);
}
@@ -121,6 +125,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
return;
}
super.onInit();
if (this.maxSubscribers == null) {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
if (this.messageConverter == null){
this.messageConverter = new SimpleMessageConverter();
}

8
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisChannelParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -16,13 +16,14 @@
package org.springframework.integration.redis.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractChannelParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.redis.channel.SubscribableRedisChannel;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the 'channel' and 'publish-subscribe-channel' elements of the
@@ -52,7 +53,8 @@ public class RedisChannelParser extends AbstractChannelParser {
// The following 2 attributes should be added once configurable on the RedisMessageListenerContainer
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
this.setMaxSubscribersProperty(parserContext, builder, element, IntegrationNamespaceUtils.DEFAULT_MAX_BROADCAST_SUBSCRIBERS_PROPERTY_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-subscribers");
return builder;
}

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests.java
View File

@@ -30,8 +30,6 @@ import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.expression.Expression;
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

5
spring-integration-security/src/test/java/org/springframework/integration/security/SecurityTestUtils.java
View File

@@ -16,12 +16,13 @@
package org.springframework.integration.security;
import java.util.Arrays;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.security.core.GrantedAuthority;
import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextImpl;
import org.springframework.util.CollectionUtils;
/**
* @author Jonas Partner
@@ -39,7 +40,7 @@ public class SecurityTestUtils {
authorities[i] = new SimpleGrantedAuthority(roles[i]);
}
authToken = new UsernamePasswordAuthenticationToken(username, password,
CollectionUtils.<GrantedAuthority> arrayToList(authorities));
Arrays.asList(authorities));
}
else {
authToken = new UsernamePasswordAuthenticationToken(username, password);

5
spring-integration-xml/src/main/java/org/springframework/integration/xml/selector/XmlValidatingMessageSelector.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.xml.selector;
import java.io.IOException;
import java.util.Arrays;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -31,7 +32,6 @@ import org.springframework.integration.xml.DefaultXmlPayloadConverter;
import org.springframework.integration.xml.XmlPayloadConverter;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.xml.validation.XmlValidator;
@@ -87,6 +87,7 @@ public class XmlValidatingMessageSelector implements MessageSelector {
this.converter = converter;
}
@Override
public boolean accept(Message<?> message) {
SAXParseException[] validationExceptions = null;
try {
@@ -100,7 +101,7 @@ public class XmlValidatingMessageSelector implements MessageSelector {
if (this.throwExceptionOnRejection) {
throw new MessageRejectedException(message, "Message was rejected due to XML Validation errors",
new AggregatedXmlMessageValidationException(
CollectionUtils.<Throwable> arrayToList(validationExceptions)));
Arrays.<Throwable> asList(validationExceptions)));
}
if (logger.isDebugEnabled()) {
logger.debug("Message was rejected due to XML Validation errors");

4
spring-integration-xml/src/test/java/org/springframework/integration/xml/source/DomSourceFactoryTests.java
View File

@@ -16,7 +16,9 @@
package org.springframework.integration.xml.source;
import static org.custommonkey.xmlunit.XMLAssert.*;
import static org.junit.Assert.assertNotNull;
import static org.custommonkey.xmlunit.XMLAssert.assertXMLEqual;
import static org.junit.Assert.assertEquals;
import java.io.StringReader;

2
src/reference/docbook/preface.xml
View File

@@ -32,7 +32,7 @@
<para>
The default dependency used by <emphasis>Spring Integration</emphasis>
<emphasis role="strong">3.0.0.RELEASE</emphasis> is
<emphasis>Spring Framework</emphasis> <emphasis role="strong">3.1.4.RELEASE</emphasis>.
<emphasis>Spring Framework</emphasis> <emphasis role="strong">3.2.x</emphasis>.
</para>
<para>
Generally, <emphasis>Spring Integration</emphasis> <emphasis role="strong">3.0.x</emphasis>

5
src/reference/docbook/whats-new.xml
View File

@@ -141,6 +141,7 @@
To achieve flexible Request Mapping configuration, Spring Integration provides the <code>&lt;request-mapping/&gt;</code>
sub-element for <code>&lt;http:inbound-channel-adapter/&gt;</code> and <code>&lt;http:inbound-gateway/&gt;</code>.
Both HTTP Inbound Endpoints are now fully based on the Request Mapping infrastructure that was introduced with Spring MVC 3.1.
For example, multiple paths are supported on a single inbound endpoint.
For more information see <xref linkend="http-namespace"/>.
</para>
</section>
@@ -643,7 +644,7 @@
</para>
</section>
<section id="3.0-redis">
<title>Redis Adapters Changers</title>
<title>Redis Adapter Changes</title>
<para>
<itemizedlist>
<listitem>
@@ -665,7 +666,7 @@
</para>
</section>
<section id="3.0-filelistfilter">
<title>Persistend File List Filters (file, (S)FTP)</title>
<title>Persistent File List Filters (file, (S)FTP)</title>
<para>
New <classname>FileListFilter</classname>s that use a persistent <classname>MetadataStore</classname> are
now available. These can be used to prevent duplicate files after a system restart. See