diff --git a/spring-integration-core/src/main/java/org/springframework/integration/SystemInterruptedException.java b/spring-integration-core/src/main/java/org/springframework/integration/SystemInterruptedException.java deleted file mode 100644 index 2d8f485bae..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/SystemInterruptedException.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2002-2007 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; - -/** - * An unchecked wrapper exception for InterruptedExceptions. - * - * @author Mark Fisher - */ -@SuppressWarnings("serial") -public class SystemInterruptedException extends MessagingException { - - public SystemInterruptedException(String message, Throwable cause) { - super(message, cause); - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index edd6691adf..a941bf600a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -303,7 +303,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } if (concurrencyPolicy != null) { - handler = new PooledMessageHandler(handler, concurrencyPolicy.getCoreConcurrency(), concurrencyPolicy.getMaxConcurrency()); + handler = new PooledMessageHandler(handler, concurrencyPolicy.getCoreSize(), concurrencyPolicy.getMaxSize()); } dispatcher.addHandler(handler, schedule); if (this.isRunning() && !dispatcher.isRunning()) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java index 3274df97d0..145c753d28 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -48,7 +48,6 @@ public class DefaultChannelRegistry implements ChannelRegistry { public void registerChannel(String name, MessageChannel channel) { Assert.notNull(name, "'name' must not be null"); Assert.notNull(channel, "'channel' must not be null"); - channel.setName(name); this.channels.put(name, channel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 9f0eeb0f7d..9ba0ad24d4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -25,11 +25,6 @@ import org.springframework.integration.message.Message; */ public interface MessageChannel { - /** - * Set the name of this channel. Must be unique within a context. - */ - void setName(String name); - /** * Return the name of this channel. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index 527c12b130..e8e3786dd9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -82,9 +82,9 @@ public class EndpointParser implements BeanDefinitionParser { private static final String CONCURRENCY_ELEMENT = "concurrency"; - private static final String CORE_CONCURRENCY_ATTRIBUTE = "core"; + private static final String CORE_SIZE_ATTRIBUTE = "core"; - private static final String MAX_CONCURRENCY_ATTRIBUTE = "max"; + private static final String MAX_SIZE_ATTRIBUTE = "max"; private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy"; @@ -162,13 +162,13 @@ public class EndpointParser implements BeanDefinitionParser { private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition subscriptionDefinition) { ConcurrencyPolicy policy = new ConcurrencyPolicy(); - String coreConcurrency = concurrencyElement.getAttribute(CORE_CONCURRENCY_ATTRIBUTE); - String maxConcurrency = concurrencyElement.getAttribute(MAX_CONCURRENCY_ATTRIBUTE); - if (StringUtils.hasText(coreConcurrency)) { - policy.setCoreConcurrency(Integer.parseInt(coreConcurrency)); + String coreSize = concurrencyElement.getAttribute(CORE_SIZE_ATTRIBUTE); + String maxSize = concurrencyElement.getAttribute(MAX_SIZE_ATTRIBUTE); + if (StringUtils.hasText(coreSize)) { + policy.setCoreSize(Integer.parseInt(coreSize)); } - if (StringUtils.hasText(maxConcurrency)) { - policy.setMaxConcurrency(Integer.parseInt(maxConcurrency)); + if (StringUtils.hasText(maxSize)) { + policy.setMaxSize(Integer.parseInt(maxSize)); } subscriptionDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java index b24e7d985c..8e72d69009 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java @@ -25,27 +25,27 @@ import org.springframework.util.Assert; */ public class ConcurrencyPolicy implements EndpointPolicy { - private int coreConcurrency; + private int coreSize; - private int maxConcurrency; + private int maxSize; - public int getCoreConcurrency() { - return this.coreConcurrency; + public int getCoreSize() { + return this.coreSize; } - public void setCoreConcurrency(int coreConcurrency) { - Assert.isTrue(coreConcurrency > 0, "'coreConcurrency' must be at least 1"); - this.coreConcurrency = coreConcurrency; + public void setCoreSize(int coreSize) { + Assert.isTrue(coreSize > 0, "'coreSize' must be at least 1"); + this.coreSize = coreSize; } - public int getMaxConcurrency() { - return this.maxConcurrency; + public int getMaxSize() { + return this.maxSize; } - public void setMaxConcurrency(int maxConcurrency) { - Assert.isTrue(maxConcurrency > 0, "'maxConcurrency' must be at least 1"); - this.maxConcurrency = maxConcurrency; + public void setMaxSize(int maxSize) { + Assert.isTrue(maxSize > 0, "'maxSize' must be at least 1"); + this.maxSize = maxSize; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandler.java index 13a87f02b7..1d345df7c5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandler.java @@ -19,7 +19,7 @@ package org.springframework.integration.handler; import org.springframework.integration.message.Message; /** - * Generic message handler interface. Typical implementations will translate + * Base message handler interface. Typical implementations will translate * between the generic Messages of the integration framework and the domain * objects that are passed-to and returned-from business components. * diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java index 24464a3cd9..3dadf39a90 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java @@ -16,9 +16,6 @@ package org.springframework.integration.message; -import java.util.concurrent.locks.ReentrantLock; - -import org.springframework.integration.SystemInterruptedException; import org.springframework.integration.util.RandomGuidUidGenerator; import org.springframework.integration.util.UidGenerator; import org.springframework.util.Assert; @@ -38,8 +35,6 @@ public class GenericMessage implements Message { private UidGenerator defaultUidGenerator = new RandomGuidUidGenerator(); - private ReentrantLock lock; - /** * Create a new message with the given id and payload. @@ -87,21 +82,4 @@ public class GenericMessage implements Message { this.payload = newPayload; } - public void lock() { - this.lock.lock(); - } - - public void lockInterruptibly() { - try { - lock.lockInterruptibly(); - } - catch (InterruptedException e) { - throw new SystemInterruptedException("Unable to obtain lock for message", e); - } - } - - public void unlock() { - lock.unlock(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java index bff4930903..7294978662 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java @@ -29,8 +29,4 @@ public interface Message { T getPayload(); - void lock(); - - void unlock(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java index 1458df107d..dabc5dd1d0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java @@ -24,24 +24,24 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; /** - * A holder for Message metadata. This includes information that is used by the - * messaging system (such correlationId) as well as information that is - * relevant for specific messaging endpoints. For the latter, String values may - * be stored as properties and Object values may be stored as + * A holder for Message metadata. This includes information that may be used by + * the messaging system (such as correlationId) as well as information + * that is relevant for specific messaging endpoints. For the latter, String + * values may be stored as properties and Object values may be stored as * attributes. * * @author Mark Fisher */ public class MessageHeader { - private Object correlationId; - - private String replyChannelName; - private Date timestamp = new Date(); private Date expiration; + private Object correlationId; + + private String replyChannelName; + private int sequenceNumber = 1; private int sequenceSize = 1; @@ -51,6 +51,29 @@ public class MessageHeader { private Map attributes = new ConcurrentHashMap(); + /** + * Return the creation time of this message. + */ + public Date getTimestamp() { + return this.timestamp; + } + + /** + * Return the expiration date for this message or null to + * indicate 'never expire'. + */ + public Date getExpiration() { + return this.expiration; + } + + /** + * Set the expiration date for this message or null to + * indicate 'never expire'. The default is null. + */ + public void setExpiration(Date expiration) { + this.expiration = expiration; + } + public Object getCorrelationId() { return this.correlationId; } @@ -71,26 +94,6 @@ public class MessageHeader { return this.sequenceNumber; } - public Date getTimestamp() { - return this.timestamp; - } - - /** - * Set the expiration date for this message or null to - * indicate 'never expire'. The default is null. - */ - public void setExpiration(Date expiration) { - this.expiration = expiration; - } - - /** - * Return the expiration date for this message or null to - * indicate 'never expire'. - */ - public Date getExpiration() { - return this.expiration; - } - public void setSequenceNumber(int sequenceNumber) { this.sequenceNumber = sequenceNumber; } @@ -116,7 +119,7 @@ public class MessageHeader { for (Object key : this.properties.keySet()) { propertyNames.add((String) key); } - return propertyNames; + return propertyNames; } public Object getAttribute(String key) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index 9566b4c805..b02b4f9c87 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -56,8 +56,8 @@ public class FixedDelayConsumerTests { PollingSchedule schedule = new PollingSchedule(10); schedule.setFixedRate(false); ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); - concurrencyPolicy.setCoreConcurrency(1); - concurrencyPolicy.setMaxConcurrency(1); + concurrencyPolicy.setCoreSize(1); + concurrencyPolicy.setMaxSize(1); Subscription subscription = new Subscription(); subscription.setSchedule(schedule); subscription.setChannelName("testChannel"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index 22cf2f9e3f..d19298f5d8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -82,8 +82,8 @@ public class FixedRateConsumerTests { PollingSchedule schedule = new PollingSchedule(5); schedule.setFixedRate(true); ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); - concurrencyPolicy.setCoreConcurrency(1); - concurrencyPolicy.setMaxConcurrency(1); + concurrencyPolicy.setCoreSize(1); + concurrencyPolicy.setMaxSize(1); Subscription subscription = new Subscription(); subscription.setChannelName("testChannel"); subscription.setSchedule(schedule);