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

INT-330: BufferingMessageHandler -> CorrelatingMessageHandler and MessagesProcessor -> MessageGroupProcessor

Browse Source
This commit is contained in:
Iwein Fuld
2009-11-26 18:16:45 +00:00
parent 0ce4b2ebb7
commit 94c9c4d4fe
12 changed files with 45 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,13 +39,13 @@ import java.util.concurrent.locks.ReentrantLock;
*
* @author Iwein Fuld
*/
public class BufferingMessageHandler extends AbstractMessageHandler implements Lifecycle {
public class CorrelatingMessageHandler extends AbstractMessageHandler implements Lifecycle {
private MessageStore store = new SimpleMessageStore(100);
private final CorrelationStrategy correlationStrategy;
private final IdTracker tracker = new IdTracker();
private final CompletionStrategy completionStrategy;
private MessagesProcessor outputProcessor;
private MessageGroupProcessor outputProcessor;
private MessageChannel outputChannel;
private volatile MessageChannel discardChannel = new NullChannel();
private TaskScheduler taskScheduler;
@@ -57,10 +57,10 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
private volatile boolean sendPartialResultOnTimeout;
private ChannelResolver channelResolver;
public BufferingMessageHandler(MessageStore store,
public CorrelatingMessageHandler(MessageStore store,
CorrelationStrategy correlationStrategy,
CompletionStrategy completionStrategy,
MessagesProcessor processor) {
MessageGroupProcessor processor) {
Assert.notNull(store);
Assert.notNull(correlationStrategy);
Assert.notNull(completionStrategy);
@@ -71,8 +71,8 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
this.outputProcessor = processor;
}
public BufferingMessageHandler(MessageStore store,
MessagesProcessor processor) {
public CorrelatingMessageHandler(MessageStore store,
MessageGroupProcessor processor) {
this(store, new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID),
new SequenceSizeCompletionStrategy(), processor);

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java
View File

@@ -27,7 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
/**
* @author Iwein Fuld
*/
public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessagesProcessor {
public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessageGroupProcessor {
private final ConcurrentMap<Object, AtomicInteger> nextMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
private volatile DefaultResequencerStrategies.SequenceNumberComparator sequenceSizeComparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java
View File

@@ -9,7 +9,7 @@ import java.util.Collection;
/**
* @author Iwein Fuld
*/
public interface MessagesProcessor {
public interface MessageGroupProcessor {
void processAndSend(Object correlationKey,
Collection<Message<?>> messagesUpForProcessing,

13
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java
View File

@@ -3,12 +3,11 @@ package org.springframework.integration.aggregator;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
public class MethodInvokingAggregatorFactoryBean implements
FactoryBean<BufferingMessageHandler> , InitializingBean{
FactoryBean<CorrelatingMessageHandler> , InitializingBean{
private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE;
@@ -19,7 +18,7 @@ public class MethodInvokingAggregatorFactoryBean implements
private CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
private MessagesProcessor processor;
private MessageGroupProcessor processor;
private Object target;
@@ -33,13 +32,13 @@ public class MethodInvokingAggregatorFactoryBean implements
// build processor
}
public BufferingMessageHandler getObject() throws Exception {
return new BufferingMessageHandler(store, correlationStrategy,
public CorrelatingMessageHandler getObject() throws Exception {
return new CorrelatingMessageHandler(store, correlationStrategy,
completionStrategy, processor);
}
public Class<? extends BufferingMessageHandler> getObjectType() {
return BufferingMessageHandler.class;
public Class<? extends CorrelatingMessageHandler> getObjectType() {
return CorrelatingMessageHandler.class;
}
public boolean isSingleton() {

4
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -11,14 +11,14 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.*;
public class MethodInvokingMessagesProcessor implements MessagesProcessor {
public class MethodInvokingMessageGroupProcessor implements MessageGroupProcessor {
private final Object target;
private final Method method;
private final MessageListMethodAdapter adapter;
public MethodInvokingMessagesProcessor(Object target) {
public MethodInvokingMessageGroupProcessor(Object target) {
this.target = target;
this.method = selectMethodFrom(target);
this.adapter = new MessageListMethodAdapter(target, method);

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -5,7 +5,7 @@ import org.springframework.integration.core.MessageChannel;
import java.util.Collection;
public class PassThroughMessagesProcessor implements MessagesProcessor {
public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
for (Message<?> message : messagesUpForProcessing) {

6
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java
View File

@@ -15,11 +15,11 @@ public class BufferingMessageHandlerIntegrationTest {
private CorrelationStrategy correlationStrategy;
private MessageStore store = new SimpleMessageStore(100);
private MessageChannel outputChannel = mock(MessageChannel.class);
private MessagesProcessor processor = new PassThroughMessagesProcessor();
// private BufferingMessageHandler customizedHandler = new BufferingMessageHandler(
private MessageGroupProcessor processor = new PassThroughMessageGroupProcessor();
// private CorrelatingMessageHandler customizedHandler = new CorrelatingMessageHandler(
// store, correlationStrategy, completionStrategy, processor,
// outputChannel);
private BufferingMessageHandler defaultHandler = new BufferingMessageHandler(
private CorrelatingMessageHandler defaultHandler = new CorrelatingMessageHandler(
store, processor);
@Before public void setupHandler(){

6
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java
View File

@@ -37,7 +37,7 @@ import static org.mockito.Mockito.*;
@RunWith(MockitoJUnitRunner.class)
public class BufferingMessageHandlerTest {
private BufferingMessageHandler buffer;
private CorrelatingMessageHandler buffer;
@Mock
private MessageStore store;
@Mock
@@ -45,13 +45,13 @@ public class BufferingMessageHandlerTest {
@Mock
private CompletionStrategy completionStrategy;
@Mock
private MessagesProcessor processor;
private MessageGroupProcessor processor;
@Mock
private MessageChannel outputChannel;
@Before
public void initializeSubject() {
buffer = new BufferingMessageHandler(store, correlationStrategy,
buffer = new CorrelatingMessageHandler(store, correlationStrategy,
completionStrategy, processor);
buffer.setOutputChannel(outputChannel);
}

10
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -24,7 +24,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class MethodInvokingMessagesProcessorTests {
public class MethodInvokingMessageGroupProcessorTests {
@Mock
private BufferedMessagesCallback processedCallback;
@@ -61,7 +61,7 @@ public class MethodInvokingMessagesProcessorTests {
@Test
public void shouldFindAnnotatedAggregatorMethod() throws Exception {
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(
new AnnotatedAggregatorMethod());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
.forClass(Message.class);
@@ -86,7 +86,7 @@ public class MethodInvokingMessagesProcessorTests {
@Test
public void shouldFindSimpleAggregatorMethod() throws Exception {
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(
new SimpleAggregator());
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
.forClass(Message.class);
@@ -119,7 +119,7 @@ public class MethodInvokingMessagesProcessorTests {
@Test
public void shouldFindFittingMethodAmongMultipleUnanotated() {
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(
new UnnanotatedAggregator()
);
@@ -150,7 +150,7 @@ public class MethodInvokingMessagesProcessorTests {
@Test
public void shouldFindFittingMethodAmongMultipleWithAnnotatedParameters() {
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(
new AnnotatedParametersAggregator()
);

25
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewAggregatorEndpointTests.java
View File

@@ -16,36 +16,25 @@
package org.springframework.integration.aggregator;
import org.junit.After;
import static org.junit.Assert.*;
import static org.hamcrest.CoreMatchers.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.Ignore;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.aggregator.BufferingMessageHandler;
import org.springframework.integration.aggregator.MessagesProcessor;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.MessageGroupProcessor;
import org.springframework.integration.aggregator.BufferedMessagesCallback;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.config.StubTaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.test.annotation.Repeat;
import java.util.Collection;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ScheduledFuture;
/**
* @author Mark Fisher
@@ -54,11 +43,11 @@ import java.util.concurrent.ScheduledFuture;
*/
public class NewAggregatorEndpointTests {
private BufferingMessageHandler aggregator;
private CorrelatingMessageHandler aggregator;
@Before
public void configureAggregator() {
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
}
@Test
@@ -236,7 +225,7 @@ public class NewAggregatorEndpointTests {
@Test
public void testNullReturningAggregator() throws InterruptedException {
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
@@ -262,7 +251,7 @@ public class NewAggregatorEndpointTests {
return builder.build();
}
private class MultiplyingProcessor implements MessagesProcessor {
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing,
MessageChannel outputChannel, BufferedMessagesCallback processedCallback
) {
@@ -279,7 +268,7 @@ public class NewAggregatorEndpointTests {
}
}
private class NullReturningMessageProcessor implements MessagesProcessor {
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
//noop
}

15
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java
View File

@@ -31,12 +31,11 @@ import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.aggregator.BufferingMessageHandler;
import org.springframework.integration.aggregator.MessagesProcessor;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.MessageGroupProcessor;
import org.springframework.integration.aggregator.BufferedMessagesCallback;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.annotation.Repeat;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
@@ -53,7 +52,7 @@ public class NewConcurrentAggregatorEndpointTests {
private ThreadPoolTaskScheduler taskScheduler;
private BufferingMessageHandler aggregator;
private CorrelatingMessageHandler aggregator;
@Before
public void configureAggregator() {
@@ -61,7 +60,7 @@ public class NewConcurrentAggregatorEndpointTests {
this.taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
this.taskScheduler.afterPropertiesSet();
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
this.aggregator.setTaskScheduler(this.taskScheduler);
}
@@ -243,7 +242,7 @@ public class NewConcurrentAggregatorEndpointTests {
@Test
public void testNullReturningAggregator() throws InterruptedException {
this.aggregator.start();
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
this.aggregator = new CorrelatingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
this.aggregator.setTaskScheduler(this.taskScheduler);
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
@@ -320,7 +319,7 @@ public class NewConcurrentAggregatorEndpointTests {
if (this.aggregator != null) this.aggregator.stop();
}
private class MultiplyingProcessor implements MessagesProcessor {
private class MultiplyingProcessor implements MessageGroupProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing,
MessageChannel outputChannel, BufferedMessagesCallback processedCallback
) {
@@ -337,7 +336,7 @@ public class NewConcurrentAggregatorEndpointTests {
}
}
private class NullReturningMessageProcessor implements MessagesProcessor {
private class NullReturningMessageProcessor implements MessageGroupProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
//noop
}

7
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java
View File

@@ -22,7 +22,7 @@ import static org.hamcrest.CoreMatchers.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.Ignore;
import org.junit.matchers.JUnitMatchers;
import static org.junit.matchers.JUnitMatchers.*;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
@@ -33,7 +33,6 @@ import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import java.util.Arrays;
import static java.util.Arrays.*;
/**
@@ -43,7 +42,7 @@ import static java.util.Arrays.*;
*/
public class NewResequencerTests {
private BufferingMessageHandler resequencer;
private CorrelatingMessageHandler resequencer;
private ThreadPoolTaskScheduler taskScheduler;
@@ -53,7 +52,7 @@ public class NewResequencerTests {
public void configureResequencer() {
this.resequencerStrategies = new DefaultResequencerStrategies();
MessageStore store = new SimpleMessageStore(30);
this.resequencer = new BufferingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies);
this.resequencer = new CorrelatingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies);
this.taskScheduler = TestUtils.createTaskScheduler(10);
this.resequencer.setTaskScheduler(taskScheduler);
this.taskScheduler.afterPropertiesSet();