INT-3413 Emit Event When an Agg. Expires a Group

JIRA: https://jira.spring.io/browse/INT-3413

Add MessageGroupExpiredEvent with group metadata
such as message count, last modified, whether
the group was discarded or not.

Conflicts:
	spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

Polishing

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

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();
+	}
+
+}

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<MessageGroupExpiredEvent> expiryEvents = new ArrayList<MessageGroupExpiredEvent>();
 
 	@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