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

  • Aggregator Issues

    Hi team, I have the following queries need your input. I am using the 2.1.0 version.
    1. It seems the aggregator would hangs up if there is no messages, even MessageGroupStoreReaper is configured, MessageGroupStoreReaper won�t take action if the first message wasn�t received?
    2. It is said that 'expire-groups-upon-completion' is the new feature of SI 2.1. Default is 'false'. And SI maintain a reference to a released MessageGroup which means that the aggregator thinks that all our messages coming after the first release are "Late Arrivals" and discards them. so setting 'expire-groups-upon-completion' to 'true' will expire MessageGroup after each release so new message with the same correlation id will create a new MessageGroup. However, it seems that even after the group is expired, the late arrivals can release with the same correlation id. I am just wondering if the group is expired, should the late arrivals be discarded directly instead of coming to the canRleased. As the requirement is meeted in this case, so the late arrivals is released finally.
    3.If the above two issues are encountered, then the method "aggregate" that implements the message aggregation algorithm seems not to be processed. If this is the case, how can I return any reasonable response to outputChannel or how to handle this case?
    Below is the aggregator revelant configuration. Please note that I implement my own TimeoutCountSequenceSizeReleaseStrategy to fulfil some function, but I think it shouldn't have effect on the above issues. Thanks in advance for your reply.

    <int:aggregator id="completelyDefinedAggregator"
    input-channel="aggregatorChannel" output-channel="wsOutChannel" ref="aggregatorBean"
    method="aggregate" release-strategy="timeout" message-store="messageStore"
    send-partial-result-on-expiry="false" />
    <bean id="timeout"
    class=" ReleaseStrategy">
    <constructor-arg name="threshold" value="2" />
    <constructor-arg name="timeout" value="10000" />
    <beans:bean id="aggregatorBean" class=""/>
    <beans:bean id="messageStore"
    class=" eMessageStore" />
    <beans:bean id="reaper"
    class=" geGroupStoreReaper">
    <beansroperty name="messageGroupStore" ref="messageStore" />
    <beansroperty name="timeout" value="15000" />
    <task:scheduled-tasks scheduler="scheduler">
    <task:scheduled ref="reaper" method="run" fixed-rate="5000" />
    <task:scheduler id="scheduler" pool-size="1" />

  • #2
    Please check out this sample. It demonstrates exactly what you you are asking
    Also, you may want to watch this screencast where I also talk about this particular case amongst others

    Let me know if you still have questions.


    • #3
      There is also a time-based variant of the aggregator demo, so please check that out as well


      • #4
        Hi oleg, Thanks for you quick reply~
        Actually, I have looked into the two demos you provided and applied it into my case. After some testing, I have the above queries. I would really appreciate your answering my queries above. For the screencast, I can't access to it before I come to office, also hope I can get the answer from the screencast when I am back in office tmr. Thanks a lot!