diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 9fec2f6612..bc7ef3ae94 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -31,6 +31,8 @@ import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -81,7 +83,8 @@ import org.springframework.util.StringUtils; * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler - implements MessageProducer, DisposableBean, IntegrationEvaluationContextAware { + implements MessageProducer, DisposableBean, IntegrationEvaluationContextAware, + ApplicationEventPublisherAware { private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class); @@ -123,6 +126,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH private EvaluationContext evaluationContext; + private volatile ApplicationEventPublisher applicationEventPublisher; + public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); @@ -197,6 +202,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH super.setTaskScheduler(taskScheduler); } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -572,6 +582,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH discardChannel.send(message); } } + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new MessageGroupExpiredEvent(this, correlationKey, group + .size(), new Date(group.getLastModified()) , new Date(), !sendPartialResultOnExpiry)); + } } protected void completeGroup(Object correlationKey, MessageGroup group) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupExpiredEvent.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupExpiredEvent.java new file mode 100644 index 0000000000..4aeed80bf7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupExpiredEvent.java @@ -0,0 +1,100 @@ +/* + * Copyright 2014 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.aggregator; + +import java.util.Date; + +import org.springframework.integration.event.IntegrationEvent; +import org.springframework.integration.support.context.NamedComponent; + +/** + * Event representing the expiration of a message group. + * + * @author Gary Russell + * @since 4.0.1 + */ +public class MessageGroupExpiredEvent extends IntegrationEvent { + + private static final long serialVersionUID = -7126221042599333919L; + + private final Object groupId; + + private final int messageCount; + + private final Date lastModified; + + private final Date expired; + + private final boolean discarded; + + public MessageGroupExpiredEvent(Object source, Object groupId, int messageCount, Date lastModified, Date expired, + boolean discarded) { + super(source); + this.groupId = groupId; + this.messageCount = messageCount; + this.lastModified = lastModified; + this.expired = expired; + this.discarded = discarded; + } + + public Object getGroupId() { + return groupId; + } + + public int getMessageCount() { + return messageCount; + } + + protected Date getLastModified() { + return lastModified; + } + + public Date getExpired() { + return expired; + } + + public boolean isDiscarded() { + return discarded; + } + + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + Object sourceName; + if (this.source instanceof NamedComponent) { + sourceName = ((NamedComponent) source).getComponentName(); + } + else { + sourceName = this.source.toString(); + } + builder.append("MessageGroupExpiredEvent [groupId=") + .append(this.groupId) + .append(", messageCount=") + .append(this.messageCount) + .append(", lastModified=") + .append(this.lastModified) + .append(", expiredAt=") + .append(this.expired) + .append(", discarded=") + .append(this.discarded) + .append(", source=") + .append(sourceName) + .append("]"); + return builder.toString(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 9b333ad107..cd87e3cba2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -13,12 +13,9 @@ package org.springframework.integration.aggregator; -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThat; -import static org.mockito.Mockito.mock; +import static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; import java.util.ArrayList; import java.util.Collection; @@ -32,6 +29,8 @@ import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.AbstractMessageHandler; @@ -61,12 +60,22 @@ public class AggregatorTests { private final SimpleMessageStore store = new SimpleMessageStore(50); + List expiryEvents = new ArrayList(); @Before public void configureAggregator() { this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store); this.aggregator.setBeanFactory(mock(BeanFactory.class)); + this.aggregator.setApplicationEventPublisher(new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + expiryEvents.add((MessageGroupExpiredEvent) event); + } + }); + this.aggregator.setBeanName("testAggregator"); this.aggregator.afterPropertiesSet(); + expiryEvents.clear(); } @Test @@ -98,13 +107,15 @@ public class AggregatorTests { for (int i=0; i < 120000; i++) { if (i % 10000 == 0) { stopwatch.stop(); - logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); stopwatch.start(); } handler.handleMessage(message); } stopwatch.stop(); - logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); } @Test @@ -144,7 +155,7 @@ public class AggregatorTests { } } - }; + } DirectChannel outputChannel = new DirectChannel(); CustomHandler handler = new CustomHandler(outputChannel); @@ -162,13 +173,15 @@ public class AggregatorTests { for (int i=0; i < 120000; i++) { if (i % 10000 == 0) { stopwatch.stop(); - logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() + + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); stopwatch.start(); } handler.handleMessage(message); } stopwatch.stop(); - logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); + logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() + + " (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)"); } @Test @@ -200,6 +213,11 @@ public class AggregatorTests { Message discardedMessage = discardChannel.receive(1000); assertNotNull("A message should have been discarded", discardedMessage); assertEquals(message, discardedMessage); + assertEquals(1, expiryEvents.size()); + assertSame(this.aggregator, expiryEvents.get(0).getSource()); + assertEquals("ABC", this.expiryEvents.get(0).getGroupId()); + assertEquals(1, this.expiryEvents.get(0).getMessageCount()); + assertEquals(true, this.expiryEvents.get(0).isDiscarded()); } @Test @@ -214,6 +232,11 @@ public class AggregatorTests { Message reply = replyChannel.receive(1000); assertNotNull("A reply message should have been received", reply); assertEquals(15, reply.getPayload()); + assertEquals(1, expiryEvents.size()); + assertSame(this.aggregator, expiryEvents.get(0).getSource()); + assertEquals("ABC", this.expiryEvents.get(0).getGroupId()); + assertEquals(2, this.expiryEvents.get(0).getMessageCount()); + assertEquals(false, this.expiryEvents.get(0).isDiscarded()); } @Test