title: Aggregate Messages in Spring Cloud Stream
body: <p>I am new to spring cloud and looking to change our mono structure to micro services , this said at first what i am trying to do now is the following</p>
  
  <ol>
  <li>Receive Requests to invoke a web service (External System) from different sources. At any certain time this can be 1 request or up to 100K request.</li>
  <li>The external system support bulk , so it is better if i can aggregate the messages and send them in bulk . For example keep aggregating until either number threshold is reached (100 message) or time threshold 2 second is reached.</li>
  <li>Also if i received an error i want to back off exponentially </li>
  </ol>
  
  <p>My first idea is to create a Processor before my Sink that do the above aggregation .</p>
  
  <p>Is this the correct way of thinking in cloud computing or their is another path to go through ?</p>
  
  <hr>
  
  <p>Working Solution</p>
  
  <pre><code>@EnableBinding(Processor.class)
  class Configuration {
  
      @Autowired
      Processor processor;
  
  
      @ServiceActivator(inputChannel = Processor.INPUT)
      @Bean
      public MessageHandler aggregator() {
  
          AggregatingMessageHandler aggregatingMessageHandler =
                  new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
                          new SimpleMessageStore(10));
  
          //AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
          //aggregatorFactoryBean.setMessageStore();
          aggregatingMessageHandler.setOutputChannel(processor.output());
          //aggregatorFactoryBean.setDiscardChannel(processor.output());
          aggregatingMessageHandler.setSendPartialResultOnExpiry(true);
          aggregatingMessageHandler.setSendTimeout(1000L);
          aggregatingMessageHandler.setCorrelationStrategy(new  ExpressionEvaluatingCorrelationStrategy("'FOO'"));
          aggregatingMessageHandler.setReleaseStrategy(new MessageCountReleaseStrategy(3)); //ExpressionEvaluatingReleaseStrategy("size() == 5")
          aggregatingMessageHandler.setExpireGroupsUponCompletion(true);
          aggregatingMessageHandler.setGroupTimeoutExpression(new ValueExpression&lt;&gt;(3000L)); //size() ge 2 ? 5000 : -1
          aggregatingMessageHandler.setExpireGroupsUponTimeout(true);
          return aggregatingMessageHandler;
      }
  }
  </code></pre>