<?xml version="1.0" encoding="UTF-8"?>
<section xmlns="http://docbook.org/ns/docbook" version="5.0" xml:id="aggregator"
    xmlns:xlink="http://www.w3.org/1999/xlink">
  <title>Aggregator</title>

  <section id="aggregator-introduction">
    <title>Introduction</title>

    <para>Basically a mirror-image of the Splitter, the Aggregator is a type
    of Message Handler that receives multiple Messages and combines them into
    a single Message. In fact, an Aggregator is often a downstream consumer in a
    pipeline that includes a Splitter.</para>

    <para>Technically, the Aggregator is more complex than a Splitter, because
    it is stateful as it must hold the Messages to be aggregated and determine
    when the complete group of Messages is ready to be aggregated. In order to do
    this it requires a <interfacename>MessageStore</interfacename>.</para>
  </section>

  <section id="aggregator-functionality">
    <title>Functionality</title>

    <para>The Aggregator combines a group of related messages, by correlating
    and storing them, until the group is deemed complete. At that point, the
    Aggregator will create a single message by processing the whole group, and
    will send the aggregated message as output.</para>

    <para>Implementing an Aggregator requires providing the logic 
    to perform the aggregation (i.e., the creation of a single message
    from many). Two related concepts are correlation and
    release.</para>

    <para>Correlation determines how messages are grouped for aggregation.
    In Spring Integration correlation is done by default based on the CORRELATION_ID message
    header. Messages with the same CORRELATION_ID will be grouped
    together. However, the correlation strategy may be customized to allow 
    other ways of specifying how the messages should be grouped together by
    implementing a <interfacename>CorrelationStrategy</interfacename> (see below).</para>

    <para>To determine the point at which a group of messages is ready to be processed, a
    <interfacename>ReleaseStrategy</interfacename> is consulted. 
    The default release strategy for the Aggregator will release a group when all 
    messages included in a sequence are present, based on the SEQUENCE_SIZE header.
    This default strategy may be overridden by providing a reference to a
    custom <interfacename>ReleaseStrategy</interfacename> implementation.</para>
  </section>

  <section id="aggregator-api">
    <title>Programming model</title>

    <para>The Aggregation API consists of a number of classes:</para>

    <itemizedlist>
      <listitem>
        <para>The interface <interfacename>MessageGroupProcessor</interfacename>, and
        its subclasses:
        <classname>MethodInvokingAggregatingMessageGroupProcessor</classname> and
        <classname>ExpressionEvaluatingMessageGroupProcessor</classname></para>
      </listitem>

      <listitem>
        <para>The <interfacename>ReleaseStrategy</interfacename> interface and its default
        implementation <classname>SequenceSizeReleaseStrategy</classname></para>
      </listitem>

      <listitem>
        <para>The <interfacename>CorrelationStrategy</interfacename> interface and its default
        implementation <classname>HeaderAttributeCorrelationStrategy</classname></para>
      </listitem>
    </itemizedlist>

    <section>
      <title>CorrelatingMessageHandler</title>

      <para>The <classname>CorrelatingMessageHandler</classname> is a
      <interfacename>MessageHandler</interfacename> implementation, encapsulating the common
      functionalities of an Aggregator (and other correlating use cases),
      which are: <itemizedlist>
          <listitem>
            <para>correlating messages into a group to be aggregated</para>
          </listitem>

          <listitem>
            <para>maintaining those messages in a <interfacename>MessageStore</interfacename> until the group
            can be released</para>
          </listitem>

          <listitem>
            <para>deciding when the group can be released</para>
          </listitem>

          <listitem>
            <para>aggregating the released group into a single message</para>
          </listitem>

          <listitem>
            <para>recognizing and responding to an expired group</para>
          </listitem>
        </itemizedlist> The responsibility of deciding how the messages should
      be grouped together is delegated to a <interfacename>CorrelationStrategy</interfacename>
      instance. The responsibility of deciding whether the message group can
      be released is delegated to a <interfacename>ReleaseStrategy</interfacename>
      instance.</para>

      <para>Here is a brief highlight of the base
      <classname>AbstractAggregatingMessageGroupProcessor</classname> (the
      responsibility of implementing the <code>aggregatePayloads</code> method is left to
      the developer):</para>

      <programlisting language="java"><![CDATA[public abstract class AbstractAggregatingMessageGroupProcessor
              implements MessageGroupProcessor {

    protected Map<String, Object> aggregateHeaders(MessageGroup group) {
        // default implementation exists
    }

  	protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);

}]]></programlisting>

       The <interfacename>CorrelationStrategy</interfacename> is owned by the 

      <classname>CorrelatingMessageHandler</classname>

       and it has a default value based on the CORRELATION_ID message header: 

      <programlisting language="java"><![CDATA[
    public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
                   CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
     ...
      this.correlationStrategy = correlationStrategy == null ?
        new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID) : correlationStrategy;
      this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy;
      ...
  }
]]></programlisting>

      <para>As for actual processing of the message group, the default implementation is the
      <classname>DefaultAggregatingMessageGroupProcessor</classname>. It creates a
      single Message whose payload is a List of the payloads received for a
      given group. This works well for simple Scatter Gather implementations with either a Splitter, Publish
      Subscribe Channel, or Recipient List Router upstream.</para>

      <note>
        <para>When using a Publish Subscribe Channel or Recipient List Router
        in this type of scenario, be sure to enable the flag to
        <code>apply-sequence</code>. That will add the necessary
        headers (CORRELATION_ID, SEQUENCE_NUMBER and SEQUENCE_SIZE). That
        behavior is enabled by default for Splitters in Spring Integration,
        but it is not enabled for the Publish Subscribe Channel or Recipient
        List Router because those components may be used in a variety of
        contexts in which these headers are not necessary.</para>
      </note>

      <para>When implementing a specific aggregator strategy for an application,
      a developer can extend
      <classname>AbstractAggregatingMessageGroupProcessor</classname> and implement the
      <code>aggregatePayloads</code> method. However, there are better solutions, less 
      coupled to the API, for implementing the aggregation logic which can be configured easily
      either through XML or through annotations.</para>
 
      <para>In general, any POJO can implement the
      aggregation algorithm if it provides a method that
      accepts a single <interfacename>java.util.List</interfacename> as an argument 
      (parameterized lists are supported as well). This method will be invoked for aggregating
      messages as follows:</para>

      <itemizedlist>
        <listitem>
          <para>if the argument is a <interfacename>java.util.List&lt;T&gt;</interfacename>, and the
          parameter type T is assignable to <interfacename>Message</interfacename>, then the whole list of
          messages accumulated for aggregation will be sent to the aggregator</para>
        </listitem>

        <listitem>
          <para>if the argument is a non-parameterized <interfacename>java.util.List</interfacename> or the
          parameter type is not assignable to <interfacename>Message</interfacename>, then the method will
          receive the payloads of the accumulated messages</para>
        </listitem>

        <listitem>
          <para>if the return type is not assignable to <interfacename>Message</interfacename>, then it will
          be treated as the payload for a Message that will be created
          automatically by the framework.</para>
        </listitem>
      </itemizedlist>

      <note>
        <para>In the interest of code simplicity, and promoting best practices
        such as low coupling, testability, etc., the preferred way of
        implementing the aggregation logic is through a POJO, and using the
        XML or annotation support for configuring it in the application.</para>
      </note>
       
    </section>

    <section>
      <title>ReleaseStrategy</title>

      <para>The <interfacename>ReleaseStrategy</interfacename> interface is defined as
      follows:</para>

      <programlisting language="java"><![CDATA[public interface ReleaseStrategy {

  boolean canRelease(MessageGroup group);

}]]></programlisting>

      <para>In general, any POJO can implement the
      completion decision logic if it provides a method that accepts a single
      <interfacename>java.util.List</interfacename> as an argument (parameterized lists
      are supported as well), and returns a boolean value. This method will be
      invoked after the arrival of each new message, to decide whether the group
      is complete or not, as follows:</para>

      <itemizedlist>
        <listitem>
          <para>if the argument is a <interfacename>java.util.List&lt;T&gt;</interfacename>, and the
          parameter type T is assignable to <interfacename>Message</interfacename>, then the whole list of
          messages accumulated in the group will be sent to the method</para>
        </listitem>

        <listitem>
          <para>if the argument is a non-parametrized <interfacename>java.util.List</interfacename> or the
          parameter type is not assignable to <interfacename>Message</interfacename>, then the method will
          receive the payloads of the accumulated messages</para>
        </listitem>

        <listitem>
          <para>the method must return true if the message group is ready for
          aggregation, and false otherwise.</para>
        </listitem>
      </itemizedlist>
      
      For example:
      
      <programlisting language="java"><![CDATA[public class MyReleaseStrategy {

    @ReleaseStrategy 
    public boolean canMessagesBeReleased(List<Message<?>>) {...}
}]]></programlisting>

<programlisting language="java"><![CDATA[public class MyReleaseStrategy {

    @ReleaseStrategy 
    public boolean canMessagesBeReleased(List<String>) {...}
}]]></programlisting>

<para>
As you can see based on the above signatures, the POJO-based Release Strategy will be passed a <classname>Collection</classname> of unmarked Messages 
if you need access to the whole <classname>Message</classname> or <classname>Collection</classname> of payload objects if the type parameter is 
anything other than <classname>Message</classname>. Typically this would satisfy the majority of use cases. However if 
for some reason you need to access the full <classname>MessageGroup</classname> - which contains <code>unmarked</code> and <code>marked</code> Messages -
then you should simply provide an implementation of the <classname>ReleaseStrategy</classname> interface.
</para>

      <para>When the group is released for aggregation, all its unmarked
      messages are processed and then marked so they will not be processed
      again. If the group is also complete (i.e. if all messages from a
      sequence have arrived or if there is no sequence defined), then the group
      is removed from the message store. Partial sequences can be released, in
      which case the next time the <interfacename>ReleaseStrategy</interfacename> is called it
      will be presented with a group containing marked messages (already
      processed) and unmarked messages (potentially a new partial
      sequence).</para>

      <para>Spring Integration provides an out-of-the box implementation for
      <interfacename>ReleaseStrategy</interfacename>, the
      <classname>SequenceSizeReleaseStrategy</classname>. This implementation consults the
      SEQUENCE_NUMBER and SEQUENCE_SIZE headers of each arriving message to decide
      when a message group is complete and ready to be aggregated. As shown
      above, it is also the default strategy.</para>
    </section>

    <section>
      <title>CorrelationStrategy</title>

      <para>The <interfacename>CorrelationStrategy</interfacename> interface is defined as
      follows:</para>

      <programlisting language="java"><![CDATA[public interface CorrelationStrategy {

  Object getCorrelationKey(Message<?> message);

}]]></programlisting>

      <para>The method returns an Object which represents the correlation key
      used for associating the message with a message group. The key must satisfy the
      criteria used for a key in a Map with respect to the implementation of
      equals() and hashCode().</para>

      <para>In general, any POJO can implement the
      correlation logic, and the rules for mapping a message to a
      method's argument (or arguments) are the same as for a
      <interfacename>ServiceActivator</interfacename> (including support for @Header
      annotations). The method must return a value, and the value must not be
      <code>null</code>.</para>

      <para>Spring Integration provides an out-of-the box implementation for
      <interfacename>CorrelationStrategy</interfacename>, the
      <classname>HeaderAttributeCorrelationStrategy</classname>. This implementation
      returns the value of one of the message headers (whose name is specified
      by a constructor argument) as the correlation key. By default, the
      correlation strategy is a <classname>HeaderAttributeCorrelationStrategy</classname> returning
      the value of the CORRELATION_ID header attribute. If you have a custom header name
      you would like to use for correlation, then simply configure that on an instance of
      <classname>HeaderAttributeCorrelationStrategy</classname> and provide that as a
      reference for the Aggregator's correlation-strategy.</para>
    </section>
  </section>

	<section id="aggregator-config">
		<title>Configuring Aggregator</title>
		
		<section id="aggregator-xml">
    <title>Configuring an Aggregator with XML</title>

    <para>Spring Integration supports the configuration of an aggregator via
    XML through the &lt;aggregator/&gt; element. Below you can see an example
    of an aggregator.</para>

    <programlisting lang="xml"><![CDATA[<channel id="inputChannel"/>


<int:aggregator id="myAggregator" ]]><co id="aggxml01" /><![CDATA[
		auto-startup="true" ]]><co id="aggxml02" /><![CDATA[
		input-channel="inputChannel" ]]><co id="aggxml03" /><![CDATA[
		output-channel="outputChannel" ]]><co id="aggxml04" /><![CDATA[
		discard-channel="throwAwayChannel" ]]><co id="aggxml05" /><![CDATA[
		message-store="persistentMessageStore" ]]><co id="aggxml06" /><![CDATA[
		order="1" ]]><co id="aggxml07" /><![CDATA[
		send-partial-result-on-expiry="false" ]]><co id="aggxml08" /><![CDATA[
		send-timeout="1000" ]]><co id="aggxml09" /><![CDATA[
		
		correlation-strategy="correlationStrategyBean" ]]><co id="aggxml10" /><![CDATA[
		correlation-strategy-method="correlate" ]]><co id="aggxml11" /><![CDATA[
		
		ref="aggregatorBean" ]]><co id="aggxml12" /><![CDATA[
		method="aggregate" ]]><co id="aggxml13" /><![CDATA[
			
		release-strategy="releaseStrategyBean" ]]><co id="aggxml14" /><![CDATA[
		release-strategy-method="release"/> ]]><co id="aggxml15" /><![CDATA[


<int:channel id="outputChannel"/>

<int:channel id="throwAwayChannel"/>

<bean id="persistentMessageStore" class="org.springframework.integration.jdbc.JdbcMessageStore">
	<constructor-arg ref="dataSource"/>
</bean>

<bean id="aggregatorBean" class="sample.PojoAggregator"/>

<bean id="releaseStrategyBean" class="sample.PojoReleaseStrategy"/>

<bean id="correlationStrategyBean" class="sample.PojoCorrelationStrategy"/>]]></programlisting>

    <calloutlist>
      <callout arearefs="aggxml01">
        <para>The id of the aggregator is
        <emphasis>0ptional</emphasis>.</para>
      </callout>

      <callout arearefs="aggxml02">
        <para>Lifecycle attribute signaling if aggregator should be started during Application Context startup.
        <emphasis>Optional (default is 'true')</emphasis>.</para>
      </callout>
      
      <callout arearefs="aggxml03">
        <para>The channel from which where aggregator will receive messages.
        <emphasis>Required</emphasis>.</para>
      </callout>

      <callout arearefs="aggxml04">
        <para>The channel to which the aggregator will send the aggregation
        results. <emphasis>Optional (because incoming messages can specify a
        reply channel themselves via 'replyChannel' Message Header)</emphasis>.</para>
      </callout>

      <callout arearefs="aggxml05">
        <para>The channel to which the aggregator will send the messages that
        timed out (if <code>send-partial-result-on-expiry</code> is
        <emphasis>false</emphasis>). <emphasis>Optional</emphasis>.</para>
      </callout>

      <callout arearefs="aggxml06">
        <para>A reference to a <code>MessageGroupStore</code> used
        to store groups of messages under their correlation key until they are
        complete. <emphasis>Optional</emphasis>, by default a volatile
        in-memory store.</para>
      </callout>
      
       <callout arearefs="aggxml07">
        <para>Order of this aggregator when more than one handle is subscribed to the same DirectChannel
        (use for load balancing purposes). 
        <emphasis>Optional</emphasis>.</para>
      </callout>
      
      <callout arearefs="aggxml08">
        <para>Indicates if partially aggregated messages should be released when their storage time has expired
        (see <code>MessageGroupStore.expireMessageGroups(long)</code>).
        <emphasis>Optional</emphasis>.</para>
      </callout>
      
       <callout arearefs="aggxml09">
        <para>The timeout interval for sending the aggregated messages to the output or
        reply channel. <emphasis>Optional</emphasis>.</para>
      </callout>
     
       <callout arearefs="aggxml10">
        <para>A reference to a bean that implements the message correlation (grouping) 
        algorithm. The bean can be an implementation of the <interfacename>CorrelationStrategy</interfacename> 
        interface or a POJO. In the latter case the correlation-strategy-method attribute must be defined
        as well. <emphasis>Optional (by default, the aggregator will use
        the CORRELATION_ID header) </emphasis>.</para>
      </callout>
      
      <callout arearefs="aggxml11">
        <para>A method defined on the bean referenced by
        <code>correlation-strategy</code>, that implements the
        correlation decision algorithm. <emphasis>Optional, with
        restrictions (requires <code>correlation-strategy</code> to be
        present).</emphasis></para>
      </callout>
    
      <callout arearefs="aggxml12">
        <para>A reference to a bean defined in the application context. The bean must implement the aggregation logic
    as described above. <emphasis>Optional (by default the list of aggregated Messages will become a 
    payload of the output message).</emphasis></para>
      </callout>
      <callout arearefs="aggxml13">
        <para>A method defined on the bean referenced by <code>ref</code>,
        that implements the message aggregation
        algorithm. <emphasis>Optional, depends on <code>ref</code> attribute being defined.</emphasis></para>
      </callout>
      
      <callout arearefs="aggxml14">
        <para>A reference to a bean that implements the release strategy.
        The bean can be an implementation of the <interfacename>ReleaseStrategy</interfacename> interface
        or a POJO. In the latter case the release-strategy-method
        attribute must be defined as well. <emphasis>Optional (by default, the
        aggregator will use the SEQUENCE_SIZE header attribute)</emphasis>.</para>
      </callout>

      <callout arearefs="aggxml15">
        <para>A method defined on the bean referenced by
        <code>release-strategy</code>, that implements the
        completion decision algorithm. <emphasis>Optional, with
        restrictions (requires <code>release-strategy</code> to be
        present).</emphasis></para>
      </callout>

    </calloutlist>

    <para>Using a <code>ref</code> attribute is generally recommended if a custom
    aggregator handler implementation may be referenced in other
    <code>&lt;aggregator&gt;</code> definitions. However if a custom
    aggregator implementation is only being used by a single
    definition of the <code>&lt;aggregator&gt;</code>, you can use an inner
    bean definition (starting with version 1.0.3) to configure the aggregation
    POJO within the <code>&lt;aggregator&gt;</code> element:
    <programlisting language="xml"><![CDATA[<aggregator input-channel="input" method="sum" output-channel="output">
    <beans:bean class="org.foo.PojoAggregator"/>
</aggregator>]]></programlisting></para>

    <note>
      <para>Using both a <code>ref</code> attribute and an inner bean definition in the
      same <code>&lt;aggregator&gt;</code> configuration is not allowed, as it
      creates an ambiguous condition. In such cases, an Exception will be
      thrown.</para>
    </note>

    <para>An example implementation of the aggregator bean looks as
    follows:</para>

    <programlisting language="java"><![CDATA[public class PojoAggregator {

  public Long add(List<Long> results) {
    long total = 0l;
    for (long partialResult: results) {
      total += partialResult;
    }
    return total;
  }

}]]></programlisting>

    <para>An implementation of the completion strategy bean for the example
    above may be as follows:</para>

    <para><programlisting language="java"><![CDATA[public class PojoReleaseStrategy {
...
  public boolean canRelease(List<Long> numbers) {
    int sum = 0;
    for (long number: numbers) {
      sum += number;
    }
    return sum >= maxValue;
  }
}]]></programlisting> <note>
        <para>Wherever it makes sense, the release strategy method and the
        aggregator method can be combined in a single bean.</para>
      </note></para>

    <para>An implementation of the correlation strategy bean for the example
    above may be as follows:</para>

    <para><programlisting language="java"><![CDATA[public class PojoCorrelationStrategy {
...
  public Long groupNumbersByLastDigit(Long number) {
    return number % 10;
  }
}]]></programlisting></para>

    <para>For example, this aggregator would group numbers by some criterion
    (in our case the remainder after dividing by 10) and will hold the group
    until the sum of the numbers provided by the payloads exceeds a
    certain value.</para>

    <note>
      <para>Wherever it makes sense, the release strategy method, correlation
      strategy method and the aggregator method can be combined in a single
      bean (all of them or any two).</para>
    </note>
    
    <para>
    	<emphasis>Aggregators and Spring Expression Language (SpEL)</emphasis>
    </para>
    
    <para>
    Since Spring Integration 2.0, the various strategies (correlation, release, and aggregation) may be handled with 
    <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html">SpEL</ulink> 
    which is recommended if the logic behind such <emphasis>release strategy</emphasis> is relatively simple.
    Let's say you have a legacy component that was designed to receive an array of objects. We know that the default release 
    strategy will assemble all aggregated messages in the List. So now we have two problems. First we need to extract
    individual messages from the list, and then we need to extract the payload of each message and assemble
    the array of objects (see code below).

	<programlisting language="java"><![CDATA[public String[] processRelease(List<Message<String>> messages){
		List<String> stringList = new ArrayList<String>();
		for (Message<String> message : messages) {
			stringList.add(message.getPayload());
		}
		return stringList.toArray(new String[]{});
}]]></programlisting>

	However, with SpEL such a requirement could actually be handled relatively easily with a
	one-line expression, thus sparing you from writing a custom class and configuring it as a bean.

	<programlisting language="xml"><![CDATA[<int:aggregator input-channel="aggChannel" 
	output-channel="replyChannel" 
	expression="#this.![payload].toArray()"/>]]></programlisting>

	In the above configuration we are using a <ulink 
	url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/expressions.html#d0e12113">Collection Projection</ulink> expression 
	to assemble a new collection from the payloads of all messages in the list and then transforming it to an Array, thus 
	achieving the same result as the java code above.
    </para>
    
    <para>
    The same expression-based approach can be applied when dealing with custom <emphasis>Release</emphasis> and 
    <emphasis>Correlation</emphasis> strategies.
    </para>
    <para>
	Instead of defining a bean for a custom <classname>CorrelationStrategy</classname> via 
    the <code>correlation-strategy</code> attribute, you can implement your simple correlation logic
    via a SpEL expression and configure it via the <code>correlation-strategy-expression</code> attribute.
	</para>
	<para>
	For example:

	<programlisting language="xml"><![CDATA[correlation-strategy-expression="payload.person.id"]]></programlisting>
	
	In the above example it is assumed that the payload has an attribute <code>person</code> with an <code>id</code> 
	which is going to be used to correlate messages.
	</para>
	<para>
    Likewise, for the <interfacename>ReleaseStrategy</interfacename> you can implement your release logic as 
    a SpEL expression and configure it via the <code>release-strategy-expression</code> attribute. 
    The only difference is that since ReleaseStrategy is passed the List of Messages, the root object in the SpEL
    evaluation context is the List itself. That List can be referenced as <code>#this</code> within the expression.
	</para>
	<para>
	For example:

	<programlisting language="xml"><![CDATA[release-strategy-expression="#this.size() gt 5"]]></programlisting>
	
	In this example the root object of the SpEL Evaluation Context is the
    <interfacename>MessageGroup</interfacename> itself, and you are simply stating 
    that as soon as there are more than 5 messages in this group, it should be released.
    </para>
  </section>

  <section id="aggregator-annotations">
    <title>Configuring an Aggregator with Annotations</title>

    <para>An aggregator configured using annotations would look like this.</para>

    <programlisting language="java"><![CDATA[public class Waiter {
  ... 

  @Aggregator ]]><co id="aggann" /><![CDATA[ 
  public Delivery aggregatingMethod(List<OrderItem> items) {
    ...
  }

  @ReleaseStrategy ]]><co id="agganncs" /><![CDATA[
  public boolean releaseChecker(List<Message<?>> messages) {
    ...
  }

  @CorrelationStrategy ]]><co id="agganncorrs" /><![CDATA[
  public String correlateBy(OrderItem item) {
    ...
  }

}]]></programlisting>

    <calloutlist>
      <callout arearefs="aggann">
        <para>An annotation indicating that this method shall be used as an
        aggregator. Must be specified if this class will be used as an
        aggregator.</para>
      </callout>

      <callout arearefs="agganncs">
        <para id="aggann2">An annotation indicating that this method shall be
        used as the release strategy of an aggregator. If not present on any
        method, the aggregator will use the
        SequenceSizeReleaseStrategy.</para>
      </callout>

      <callout arearefs="agganncorrs">
        <para id="agann3">An annotation indicating that this method shall be
        used as the correlation strategy of an aggregator. If no correlation
        strategy is indicated, the aggregator will use the
        HeaderAttributeCorrelationStrategy based on CORRELATION_ID.</para>
      </callout>
    </calloutlist>

    <para>All of the configuration options provided by the xml element are
    also available for the @Aggregator annotation.</para>

    <para>The aggregator can be either referenced explicitly from XML or, if
    the @MessageEndpoint is defined on the class, detected automatically
    through classpath scanning.</para>
  </section>
		
	</section>

  <section>
    <title id="reaper">Managing State in an Aggregator:
    MessageGroupStore</title>

    <para>Aggregator (and some other patterns in Spring Integration) is a
    stateful pattern that requires decisions to be made based on a group of
    messages that have arrived over a period of time, all with the same
    correlation key. The design of the interfaces in the stateful patterns
    (e.g. <interfacename>ReleaseStrategy</interfacename>) is driven by the principle
    that the components (whether defined by the framework or a user) should be able to remain stateless.
    All state is carried by the <interfacename>MessageGroup</interfacename> and its
    management is delegated to the
    <interfacename>MessageGroupStore</interfacename>.
    
    
    <programlisting><![CDATA[public interface MessageGroupStore {
   int getMessageCountForAllMessageGroups();

   int getMarkedMessageCountForAllMessageGroups();

   int getMessageGroupCount();

   MessageGroup getMessageGroup(Object groupId);

   MessageGroup addMessageToGroup(Object groupId, Message<?> message);

   MessageGroup markMessageGroup(MessageGroup group);

   MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove);

   MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark);

   void removeMessageGroup(Object groupId);

   void registerMessageGroupExpiryCallback(MessageGroupCallback callback);

   int expireMessageGroups(long timeout);
}]]></programlisting>
    
    For more information please refer to the
    <ulink url="http://static.springsource.org/spring-integration/api/org/springframework/integration/store/MessageGroupStore.html">JavaDoc</ulink>.
    </para>
    
    <para>The <interfacename>MessageGroupStore</interfacename> accumulates state
    information in <interfacename>MessageGroups</interfacename> while waiting for
    a release strategy to be triggered, and that event might not ever happen.
    So to prevent stale messages from lingering, and for volatile stores to
    provide a hook for cleaning up when the application shuts down, the
    <interfacename>MessageGroupStore</interfacename> allows the user to register
    callbacks to apply to its <interfacename>MessageGroups</interfacename> when they
    expire. The interface is very straightforward:</para>

    <programlisting><![CDATA[public interface MessageGroupCallback {

  void execute(MessageGroupStore messageGroupStore, MessageGroup group);

}]]></programlisting>

    <para>The callback has direct access to the store and the message group
    so it can manage the persistent state (e.g. by removing the group from the
    store entirely).</para>

    <para>The <interfacename>MessageGroupStore</interfacename> maintains a list of these callbacks
    which it applies, on demand, to all messages whose timestamp is earlier than a time
    supplied as a parameter (see the <methodname>registerMessageGroupExpiryCallback(..)</methodname>
    and <methodname>expireMessageGroups(..)</methodname> methods above).</para>

    <para>The <code>expireMessageGroups</code> method can be called with a timeout value.
    Any message older than the current time minus this value will be expired,
    and have the callbacks applied. Thus it is the user of the store that
    defines what is meant by message group "expiry".</para>

    <para>As a convenience for users, Spring Integration provides a wrapper
    for the message expiry in the form of a
    <classname>MessageGroupStoreReaper</classname>:</para>

    <programlisting><![CDATA[<bean id="reaper" class="org...MessageGroupStoreReaper">
  <property name="messageGroupStore" ref="messageStore"/>
  <property name="timeout" value="30000"/>
</bean>

<task:scheduled-tasks scheduler="scheduler">	
  <task:scheduled ref="reaper" method="run" fixed-rate="10000"/>
</task:scheduled-tasks>]]></programlisting>

    <para>The reaper is a <interfacename>Runnable</interfacename>, and all that is happening
    in the example above is that the message group store's expire method is being called
    once every 10 seconds. The timeout itself is 30 seconds.</para>

    <para>In addition to the reaper, the expiry callbacks are invoked when the application
    shuts down via a lifecycle callback in the <classname>CorrelatingMessageHandler</classname>.
    </para>

    <para>The <classname>CorrelatingMessageHandler</classname> registers its
    own expiry callback, and this is the link with the boolean flag
    <code>send-partial-result-on-expiry</code> in the XML configuration of the
    aggregator. If the flag is set to true, then when the expiry callback is
    invoked, any unmarked messages in groups that are not yet released can
    be sent on to the output channel.</para>
  </section>
</section>
