title: How best to implement a DynamicPoller in Spring Integration
body: <p>I have a flow that receives large messages (that land in an RDBMS table) so I can't process too many of these at a given time.  As such I'm throttling the processing using <code>&lt;int:poller max-messages-per-poll="" /&gt;</code>, and also with some <code>queues</code> with a <code>capacity</code> set like <code>&lt;int:queue capacity=""&gt;</code>.  I understand that multiple threads/transactions will participate in this flow and for the use-case this is acceptible.</p>
  
  <p>The query polling the DB takes some time to run and as such I don't want to run it more often than I need to.  Additionally the messages this flow receives tend to come in within "bursts", meaning it might get 1000 messages then not get any in for an hour.</p>
  
  <p>What I'd like to do is use a <code>dynamic-poller</code> that will poll infrequently (since as-noted the query is costly to run) unless I see that I got a burst of messages in which case I want to poll very frequently until all messages are processed.  For example if I have <code>&lt;int:poller max-messages-per-poll="100" /&gt;</code> and I know the poller just read in 100 messages, then chances are good that there are more messages in the RDBMS that need to be processed and I should poll again immediately after our processing has completed.</p>
  
  <p>I know Spring doesn't offer a way to modify a <code>trigger</code> to make it dynamic in nature and have already looked at the Spring Integration Ref “<a href="http://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html" rel="nofollow noreferrer">7.1.5 Change Polling Rate at Runtime</a>"
  and at the <code>dynamic-poller</code> sample project:  <a href="https://github.com/spring-projects/spring-integration-samples/tree/master/intermediate/dynamic-poller" rel="nofollow noreferrer">Dynamic Poller</a><br>
  That's a start but <strong>I really need the <code>poller</code> to change it's frequency based on current load.</strong><br>
  I might not be correct on this, but I thought perhaps Gary mentioned something like this would be interesting to implement in his talk on "<em><a href="http://www.infoq.com/presentations/Implementing-HA-Architectures-Spring-Integration" rel="nofollow noreferrer">Implementing High-Availability Architectures with Spring Integration</a></em>".<br>
  In any event writing a class(es) to change the <code>poller</code> frequency doesn't seem to be a big deal.  What is a bit more challenging is how to know when a poll has occurred that produced no results since nothing gets posted to the output channel.</p>
  
  <p>Some options I've considered:  </p>
  
  <ol>
  <li><p>Attach a <code>&lt;int:wire-tap channel="" /&gt;</code> to the <code>poller</code>'s channel which calls a <code>&lt;int:service-activator&gt;</code>.  Service activator examines number of messages and adjusts <code>poller</code>'s <code>period</code> on the <code>DynamicPeriodicTrigger</code>.<br>
  Problem is that this will never get called if no messages are received so once I adjust to poll more frequently that polling period will remain indefinately.  </p></li>
  <li><p>Same as #1 but add logic to <code>DynamicPeriodicTrigger</code> that will revert <code>period</code> back to the <code>initialDelay</code> after next trigger occurs or after certain period of time.  </p></li>
  <li><p>Use an <code>&lt;int:advice-chain&gt;</code> element within the <code>&lt;int:poller&gt;</code> element with a <code>MethodInterceptor</code> implementation.<br>
  Similar to what Artem suggests in this <a href="https://stackoverflow.com/questions/20501149/spring-integration-jdbcinbound-channel-adapter-set-max-rows-per-poll-dynamic">link</a>.
  While this allows me to get in front of the <code>receive</code> method, it does not grant me access to results of the <code>receive</code> method (which would give me the number of messages retrieved).  Note this appears to be confirmed by what Gary mentions on this <a href="http://forum.spring.io/forum/spring-projects/integration/126568-transactional-channel-message-stores-and-failed-messages" rel="nofollow noreferrer">link</a>.  </p>
  
  <blockquote>
    <p>The request handler advice chain is a special case; we had to take care to only advise the internal endpoint methods and not any downstream processing (on output channels).</p>
    
    <p>Advising pollers is simpler because we're advising the whole flow. As described in section "7.1.4 Namespace Support" subsection "AOP Advice chains", you simply create an advice by implementing the MethodInterceptor interface.</p>
    
    <p>See <code>SourcePollingChannelAdapterFactoryBeanTests.testAdviceChain()</code> for a very simple advice...  </p>
    
    <p>Code:<br>
    <code>adviceChain.add(new MethodInterceptor() {</code><br>
    <code>public Object invoke(MethodInvocation invocation) throws Throwable {</code><br>
    <code>adviceApplied.set(true);</code><br>
    <code>return invocation.proceed();</code><br>
    <code>}</code><br>
    <code>});</code><br>
    This simply is used to assert that the advice was called properly; a real advice would add code before and/or after the invocation.proceed().</p>
    
    <p>In effect, this advice advises all methods, but there is only one, (<code>Callable.call()</code>).</p>
  </blockquote></li>
  <li><p>Create an <code>AfterReturning</code> advice with a pointcut that looks for the <code>Message&lt;T&gt; receive()</code> method.  </p></li>
  <li><p>Clone the <code>JdbcPollingChannelAdapter</code> and add my hooks in that new class.  </p></li>
  <li><p>Perhaps what Gary suggests on this <a href="http://forum.spring.io/forum/spring-projects/integration/127410-poller-with-advice-chain" rel="nofollow noreferrer">link</a> would be useful but the "gist" link is no longer valid.</p></li>
  </ol>
  
  <p><strong>UPDATED:</strong><br>
  The option I ended up implementing was to use an <code>AfterReturningAdvice</code> that looked something like the following.<br>
  Original code:  </p>
  
  <pre><code>&lt;int-jdbc:inbound-channel-adapter id="jdbcInAdapter" 
      channel="inputChannel" data-source="myDataSource"
      query="SELECT column1, column2 from tableA"
      max-rows-per-poll="100"&gt;
      &lt;int:poller fixed-delay="10000"/&gt;
  &lt;/int-jdbc:inbound-channel-adapter&gt;
  </code></pre>
  
  <p>New code:  </p>
  
  <pre><code>&lt;bean id="jdbcDynamicTrigger" class="DynamicPeriodicTrigger"&gt;
      &lt;constructor-arg name="period" value="20000" /&gt;
  &lt;/bean&gt; 
  &lt;bean id="jdbcPollerMetaData" class="org.springframework.integration.scheduling.PollerMetadata"&gt;
      &lt;property name="maxMessagesPerPoll" value="1000"/&gt;
      &lt;property name="trigger" ref="jdbcDynamicTrigger"/&gt;
  &lt;/bean&gt;
  &lt;bean id="pollMoreFrequentlyForHighVolumePollingStrategy" class="springintegration.scheduling.PollMoreFrequentlyForHighVolumePollingStrategy"&gt;
      &lt;property name="newPeriod" value="1"/&gt;
      &lt;property name="adjustmentThreshold" value="100"/&gt;
      &lt;property name="pollerMetadata" ref="jdbcPollerMetaData"/&gt;
  &lt;/bean&gt; 
  &lt;aop:config&gt;
      &lt;aop:aspect ref="pollMoreFrequentlyForHighVolumePollingStrategy" &gt;
          &lt;aop:after-returning pointcut="bean(jdbcInAdapterBean) and execution(* *.receive(..))" method="afterPoll" returning="returnValue"/&gt;
      &lt;/aop:aspect&gt;   
  &lt;/aop:config&gt;   
  &lt;bean id="jdbcInAdapterBean" class="org.springframework.integration.jdbc.JdbcPollingChannelAdapter"&gt;
      &lt;constructor-arg ref="myDataSource" /&gt;
      &lt;constructor-arg value="SELECT column1, column2 from tableA" /&gt;
      &lt;property name="maxRowsPerPoll" value="100" /&gt;
  &lt;/bean&gt; 
  &lt;int:inbound-channel-adapter id="jdbcInAdapter" ref="jdbcInAdapterBean" 
      channel="inputChannel"
      auto-startup="false"&gt;
      &lt;int:poller ref="jdbcPollerMetaData" /&gt;
  &lt;/int:inbound-channel-adapter&gt;
  </code></pre>
  
  <p>I've done a bit more research on this and feel that Spring Integration perhaps could offer some hooks into the pollers so that developers can better customize them.<br>
  For more info see <a href="https://jira.spring.io/browse/INT-3633" rel="nofollow noreferrer">https://jira.spring.io/browse/INT-3633</a></p>
  
  <p>If that JIRA does not get implemented and someone is interested in the code I implemented add a comment to this and I'll make the code available on github or gist.</p>