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

Simplifications

Browse Source
This commit is contained in:
Mark Fisher
2008-01-17 04:50:42 +00:00
parent b4828d4b08
commit 9b367c14ad
12 changed files with 58 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-integration-core/src/main/java/org/springframework/integration/SystemInterruptedException.java
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

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

1
spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java
View File

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

5
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -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.
*/

16
spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
View File

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

24
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandler.java
View File

@@ -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.
*

22
spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/message/Message.java
View File

@@ -29,8 +29,4 @@ public interface Message<T> {
T getPayload();
void lock();
void unlock();
}

61
spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java
View File

@@ -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 <i>correlationId</i>) as well as information that is
* relevant for specific messaging endpoints. For the latter, String values may
* be stored as <i>properties</i> 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 <i>correlationId</i>) as well as information
* that is relevant for specific messaging endpoints. For the latter, String
* values may be stored as <i>properties</i> and Object values may be stored as
* <i>attributes</i>.
*
* @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<String, Object> attributes = new ConcurrentHashMap<String, Object>();
/**
* Return the creation time of this message.
*/
public Date getTimestamp() {
return this.timestamp;
}
/**
* Return the expiration date for this message or <code>null</code> to
* indicate 'never expire'.
*/
public Date getExpiration() {
return this.expiration;
}
/**
* Set the expiration date for this message or <code>null</code> to
* indicate 'never expire'. The default is <code>null</code>.
*/
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 <code>null</code> to
* indicate 'never expire'. The default is <code>null</code>.
*/
public void setExpiration(Date expiration) {
this.expiration = expiration;
}
/**
* Return the expiration date for this message or <code>null</code> 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) {