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

Subscription is now immutable (INT-134).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-27 19:25:10 +00:00
parent 44cb3e17a1
commit 604b768625
8 changed files with 51 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/channel/DispatcherPolicy.java
View File

@@ -37,15 +37,15 @@ public class DispatcherPolicy {
private final boolean publishSubscribe;
private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
private volatile int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private int rejectionLimit = DEFAULT_REJECTION_LIMIT;
private volatile int rejectionLimit = DEFAULT_REJECTION_LIMIT;
private long retryInterval = DEFAULT_RETRY_INTERVAL;
private volatile long retryInterval = DEFAULT_RETRY_INTERVAL;
private boolean shouldFailOnRejectionLimit = true;
private volatile boolean shouldFailOnRejectionLimit = true;
public DispatcherPolicy() {

2
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -100,7 +100,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
if (!this.isInbound) {
RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
subscriptionDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel));
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(channel));
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
endpointDef.getPropertyValues().addPropertyValue("subscription", new RuntimeBeanReference(subscriptionBeanName));

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

@@ -36,6 +36,7 @@ import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
@@ -52,8 +53,6 @@ public class EndpointParser implements BeanDefinitionParser {
private static final String SUBSCRIPTION_PROPERTY = "subscription";
private static final String CHANNEL_NAME_PROPERTY = "channelName";
private static final String DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE = "default-output-channel";
private static final String DEFAULT_OUTPUT_CHANNEL_PROPERTY = "defaultOutputChannelName";
@@ -92,8 +91,6 @@ public class EndpointParser implements BeanDefinitionParser {
private static final String SCHEDULE_ELEMENT = "schedule";
private static final String SCHEDULE_PROPERTY = "schedule";
private static final String CONCURRENCY_ELEMENT = "concurrency";
private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy";
@@ -103,11 +100,8 @@ public class EndpointParser implements BeanDefinitionParser {
RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
endpointDef.setSource(parserContext.extractSource(element));
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
if (StringUtils.hasText(inputChannel)) {
subscriptionDef.getPropertyValues().addPropertyValue(CHANNEL_NAME_PROPERTY, inputChannel);
}
String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE);
Schedule schedule = null;
if (StringUtils.hasText(defaultOutputChannel)) {
endpointDef.getPropertyValues().addPropertyValue(DEFAULT_OUTPUT_CHANNEL_PROPERTY, defaultOutputChannel);
}
@@ -136,13 +130,20 @@ public class EndpointParser implements BeanDefinitionParser {
}
}
else if (SCHEDULE_ELEMENT.equals(localName)) {
this.parseSchedule((Element) child, subscriptionDef);
schedule = this.parseSchedule((Element) child);
}
}
}
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName));
if (StringUtils.hasText(inputChannel)) {
RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(inputChannel);
if (schedule != null) {
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(schedule);
}
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName));
}
if (selectors.size() > 0) {
endpointDef.getPropertyValues().addPropertyValue(SELECTORS_PROPERTY, selectors);
}
@@ -199,13 +200,13 @@ public class EndpointParser implements BeanDefinitionParser {
endpointDef.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy);
}
private void parseSchedule(Element scheduleElement, RootBeanDefinition subscriptionDef) {
private Schedule parseSchedule(Element scheduleElement) {
PollingSchedule schedule = new PollingSchedule(5);
String period = scheduleElement.getAttribute(PERIOD_ATTRIBUTE);
if (StringUtils.hasText(period)) {
schedule.setPeriod(Integer.parseInt(period));
}
subscriptionDef.getPropertyValues().addPropertyValue(SCHEDULE_PROPERTY, schedule);
return schedule;
}
private String parseHandlerAdapter(String handlerRef, String handlerMethod, ParserContext parserContext) {

7
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -131,10 +131,8 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
final DefaultMessageEndpoint endpoint) {
String channelName = annotation.input();
if (StringUtils.hasText(channelName)) {
Subscription subscription = new Subscription();
subscription.setChannelName(channelName);
Schedule schedule = new PollingSchedule(annotation.pollPeriod());
subscription.setSchedule(schedule);
Subscription subscription = new Subscription(channelName, schedule);
endpoint.setSubscription(subscription);
}
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
@@ -155,11 +153,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
Subscription subscription = new Subscription(channel);
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
subscription.setSchedule(schedule);
Subscription subscription = new Subscription(channel, schedule);
endpoint.setSubscription(subscription);
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -123,7 +123,7 @@
<xsd:element ref="selector" minOccurs="0" maxOccurs="unbounded"/>
<xsd:element ref="handler" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="input-channel" type="xsd:string"/>
<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
<xsd:attribute name="default-output-channel" type="xsd:string"/>
<xsd:attribute name="handler-ref" type="xsd:string"/>
<xsd:attribute name="handler-method" type="xsd:string"/>
@@ -209,11 +209,12 @@
<xsd:attribute name="queue-capacity" type="xsd:int"/>
<xsd:attribute name="keep-alive" type="xsd:int"/>
</xsd:complexType>
<xsd:element name="aggregator">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation> Defines an aggregating message handler
<xsd:documentation>
Defines an aggregating message handler.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>

40
spring-integration-core/src/main/java/org/springframework/integration/scheduling/Subscription.java
View File

@@ -17,30 +17,42 @@
package org.springframework.integration.scheduling;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.util.Assert;
/**
* Configuration metadata for activating a subscription.
* Configuration metadata for activating a subscription. Immutable.
*
* @author Mark Fisher
*/
public class Subscription {
private MessageChannel channel;
private final MessageChannel channel;
private String channelName;
private final String channelName;
private Schedule schedule;
private final Schedule schedule;
public Subscription() {
}
public Subscription(MessageChannel channel) {
this.channel = channel;
this(channel, null);
}
public Subscription(String channelName) {
this(channelName, null);
}
public Subscription(MessageChannel channel, Schedule schedule) {
Assert.notNull(channel, "'channel' must not be null");
this.channel = channel;
this.schedule = schedule;
this.channelName = this.channel.getName();
}
public Subscription(String channelName, Schedule schedule) {
Assert.notNull(channelName, "'channelName' must not be null");
this.channelName = channelName;
this.schedule = schedule;
this.channel = null;
}
@@ -48,24 +60,12 @@ public class Subscription {
return this.channel;
}
public void setChannel(MessageChannel channel) {
this.channel = channel;
}
public String getChannelName() {
return (this.channel != null) ? this.channel.getName() : this.channelName;
}
public void setChannelName(String channelName) {
this.channelName = channelName;
}
public Schedule getSchedule() {
return this.schedule;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
}

3
spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java
View File

@@ -47,8 +47,7 @@ public class DefaultTargetAdapterTests {
target.afterPropertiesSet();
DefaultTargetAdapter adapter = new DefaultTargetAdapter(target);
SimpleChannel channel = new SimpleChannel();
Subscription subscription = new Subscription();
subscription.setChannel(channel);
Subscription subscription = new Subscription(channel);
Message<String> message = new GenericMessage<String>("123", "testing");
channel.send(message);
assertNull(queue.poll());

7
spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java
View File

@@ -64,7 +64,6 @@ public class CorrelationIdTests {
assertEquals(message.getId(), reply.getHeader().getCorrelationId());
}
@Test
public void testCorrelationIdCopiedFromMessageCorrelationIdIfAvailable() {
Message<?> message = new StringMessage("messageId","test");
@@ -77,7 +76,7 @@ public class CorrelationIdTests {
assertEquals(message.getHeader().getCorrelationId(), reply.getHeader().getCorrelationId());
assertTrue(message.getHeader().getCorrelationId().equals(reply.getHeader().getCorrelationId()));
}
@Test
public void testCorrelationNotPassedIfAlreadySetByHandler() throws Exception {
Object correlationId = "123-ABC";
@@ -85,7 +84,7 @@ public class CorrelationIdTests {
message.getHeader().setCorrelationId(correlationId);
AbstractMessageHandlerAdapter<TestBean> adapter = new AbstractMessageHandlerAdapter<TestBean>() {
@Override
protected Object doHandle(Message message, SimpleMethodInvoker invoker) {
protected Object doHandle(Message<?> message, SimpleMethodInvoker<TestBean> invoker) {
Object result = invoker.invokeMethod(message.getPayload());
Message<?> resultMessage = new GenericMessage<Object>(result);
resultMessage.getHeader().setCorrelationId("456-XYZ");
@@ -104,7 +103,7 @@ public class CorrelationIdTests {
Message<?> message = new StringMessage("test");
AbstractMessageHandlerAdapter<TestBean> adapter = new AbstractMessageHandlerAdapter<TestBean>() {
@Override
protected Object doHandle(Message message, SimpleMethodInvoker invoker) {
protected Object doHandle(Message<?> message, SimpleMethodInvoker<TestBean> invoker) {
Object result = invoker.invokeMethod(message.getPayload());
Message<?> resultMessage = new GenericMessage<Object>(result);
resultMessage.getHeader().setCorrelationId("456-XYZ");