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

Merge branch 'work'

Browse Source
This commit is contained in:
David Syer
2010-05-04 06:45:51 +00:00
parent 5dfa2d48f3
commit e9306a3347
6 changed files with 19 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -26,7 +26,6 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
@@ -111,7 +110,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
super.setTaskScheduler(taskScheduler);
}
// TODO: remove unused property setters
// TODO: INT-958 - remove unused property setters
public void setTimeout(long timeout) {
}
@@ -152,12 +151,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
}
if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
// TODO: strategise the treatment of overwritten correlation
message = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build();
}
// TODO: make the lock global?
// TODO: INT-1117 - make the lock global?
Object lock = getLock(correlationKey);
synchronized (lock) {
@@ -203,7 +197,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
// TODO: arrange for this to be called if user desires, e.g. periodically
// TODO: INT-958 - arrange for this to be called if user desires, e.g. periodically
public final boolean forceComplete(Object correlationKey) {
Object lock = getLock(correlationKey);

24
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -28,23 +28,27 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.store.MessageGroup;
/**
* This class implements all the strategy interfaces needed for a default
* resequencer.
* This class implements all the strategy interfaces needed for a default resequencer.
*
* @author Iwein Fuld
* @author Dave Syer
*
* @since 2.0
*/
public class Resequencer implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor {
public class Resequencer implements ReleaseStrategy, MessageGroupProcessor {
private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;
public Object getCorrelationKey(Message<?> message) {
// TODO: remove this (as its duplicating the default)
Object correlationKey = message.getHeaders().getCorrelationId();
return correlationKey;
/**
* Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be
* ordered they will be released, provided they all have sequence numbers greater than those already released.
*
* @param releasePartialSequences
*/
public void setReleasePartialSequences(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
public boolean canRelease(MessageGroup messages) {
@@ -69,14 +73,10 @@ public class Resequencer implements CorrelationStrategy, ReleaseStrategy, Messag
}
}
public void setReleasePartialSequences(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
private static class SequenceNumberComparator implements Comparator<Message<?>> {
public int compare(Message<?> o1, Message<?> o2) {
return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
}
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -78,7 +78,7 @@ public @interface Aggregator {
* may be recognized as belonging to an aggregate that has already completed
* or timed out
*/
// TODO: remove / deal with tracked id capacity
// TODO: INT-958 - remove / deal with tracked id capacity
int trackedCorrelationIdCapacity() default 42;
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -77,8 +77,6 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor").getBeanDefinition());
}
// TODO: expose message store as an XML attribute
if (innerHandlerDefinition != null) {
processorBuilder.addConstructorArgValue(innerHandlerDefinition);
} else {

5
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
View File

@@ -44,7 +44,6 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
// Message group processor
builder.addConstructorArgReference(processorRef);
// TODO: expose message store as an XML attribute
// Message store
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition());
@@ -55,9 +54,9 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
}
else {
// Correlation strategy
builder.addConstructorArgReference(processorRef);
builder.addConstructorArgValue(null);
}
// Completion strategy
// Release strategy
builder.addConstructorArgReference(processorRef);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -48,7 +48,7 @@ public class ResequencerTests {
@Before
public void configureResequencer() {
this.resequencer = new CorrelatingMessageHandler(processor, store, processor, processor);
this.resequencer = new CorrelatingMessageHandler(processor, store, null, processor);
}
@Test