Announcement Announcement Module
No announcement yet.
Conditional Poller Page Title Module
Move Remove Collapse
Conversation Detail Module
  • Filter
  • Time
  • Show
Clear All
new posts

  • Conditional Poller

    Hi, everyone!

    I want to poll one message at a time from a queue channel (the work queue) and then stop polling until a certain condition becomes true.
    The background for this requirement is that several poller instances poll from the work queue for load balancing reasons and each individual poller should only poll the next work item once the route behind the poller has finished processing the work item and all sub-messages that were created in the course of that processing.
    Is there any pattern or component for this, I did not come up with anything useful despite a lot of reading?

    Best regards and thanks in advance!

  • #2

    I try to explain what's going on.
    You can configure poller with "fixed-delay", so each poll task will be runed only after previous is complete.
    However this sequential behaviour works only if your polling task is single threaded: all you work is done at the same poller thread.

    From other side you always can control your polling endpoint via Control Bus:
    You start the polling task and send to the <control-bus> a Message with command to stop your polling endpoint.
    And further you should find where to send similar message with command to start that endpoint.



    • #3
      Hi, Artem

      Thank you for your quick reply.
      I forgot to explain that the further processing after the poller is asynchronous, so the "fixed-delay" solution will not work. Also it should help to know that I use a xml configuration for the spring integration pipeline.

      The Control Bus solution sounds promising. So I will give it a try.


      • #4
        Hi, everyone

        Sorry for the late relpy. Finally I got the whole thing running using a Control Bus like Cleric suggested.
        Since I did not find many xml snippets using a Control Bus, I wanted to share my solution.
        I start to poll one item from my input queue channel and immediately stop the polling endpoint(in my case this is a recipient-list-router, but this could easily be a different component) by sending a message containing a stop command to the control-bus.
        I do so because I want to wait for the processing Channel to finish its work on the first item until I poll the next one.
            <int:channel id="inputQueue">
                <int:queue />
            <int:recipient-list-router id="jobActivator" input-channel="inputQueue">
                <int:poller max-messages-per-poll="1" fixed-delay="50" receive-timeout="10" task-executor="myTaskExecutor"/>        
                <int:recipient channel="processingChannel"/>
                <int:recipient channel="stopPolling"/>
            <int:transformer input-channel="stopPolling" output-channel="operationChannel" expression="'@jobActivator.stop()'"/>
            <int:channel id="operationChannel"/>
            <int:control-bus input-channel="operationChannel"/>
        When all the work is done by the processingChannel, I will start the polling endpoint (jobActivator) again so it can poll the next item.
        <int:transformer input-channel="startPolling" output-channel="operationChannel" expression="'@jobActivator.start()'"/>
        I hope that this post is somewhat useful. If you have any doubts, questions or a better solution for this problem feel free to respond.
        Best regards, Magnus