Alternate expiry algorithm for AbstractMessageGroupStore?

[matt.friedman]

I would like to be able to provide an alternate expiry procedure that is similar to the existing one in AbstractMessageGroupStore.

My goal is to have the message groups expire after a timeout, but only if the message group contains a message of a specific type. If not, I want to the message group to expire later when there exists in the group a message of that specific type.

I could extend SimpleMessageStore and override expireMessageGroups(); perhaps that is a good way to proceed. I also tried to see if a MessageGroupCallback would do the trick but I haven't been able to reason out how that would work.

Or maybe it makes sense to implement my own reaper component?

Can anyone suggest a good approach to achieve this?

Thanks,
Matt

[Cleric]

Hi!

MessageGroupCallback would do the trick but I haven't been able to reason out how that would work

Yes, it is a point to implement your requirements:

<bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore">
	<property name="expiryCallbacks" ref="expiryCallback"/>
</bean>
	
<bean id="expiryCallback" class="your.ExpiryCallbackImpl"/>
	
<bean id="reaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
	<property name="messageGroupStore" ref="messageStore"/>
	<property name="timeout" value="10"/>
</bean>

Where

public static class ExpiryCallbackImpl implements MessageGroupCallback {

	public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
             if (group.getMessages() and etc.) {
                messageGroupStore.removeMessageGroup(group.getGroupId());
             }
        }	
}

HTH,
Artem

[matt.friedman]

public static class ExpiryCallbackImpl implements MessageGroupCallback {

	public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
             if (group.getMessages() and etc.) {
                messageGroupStore.removeMessageGroup(group.getGroupId());
             }
        }	
}

So just to clarify for myself, if the code above is executed according to the if stmt the message group will be expired, whereas if not, the message group will NOT be expired?

In other words, unless I call messageGroupStore.removeMessageGroup(group.getGrou pId()); the group will not be removed from the message store and the group will not be expired?

Thanks,
Matt

[Cleric]

Yes, it is

[matt.friedman]

Sorry to be a bother but I'm not sure I'm getting this.

I have an aggregator that should aggregate objects A, B, C. Object A is required. B and C are optional. Sometimes B and C arrive before A.

I have a reaper that should expire objects given a timeout. However, if the timeout occurs and A has not yet arrived there is an exception in the aggregator implementation (because A is required)

When using the callback mechanism I find that the group is already expired and sent to the aggregator, so it doesn't seem to work for me this case.

I had another thought: what if I implement something in the aggregator that resends the messages if A is missing, to the channel (before the aggregator) after some additional delay? It is a sort of retry.

Is there a way to do this directly in SI, or does it seem reasonable to try it by using a procedure in the aggregator itself?

If there are other better ways I'm open to that too.

Thanks,
Matt

[Cleric]

Hmm...
Try to attain this with delayer or resequencer

[matt.friedman]

Indeed. I am learning and it finally occurred to me to do this with delayer. I delay and then reduce the sequenceSize by 1 until the objects can be aggregated. It works great now.

Thanks for being patient. I am learning. Cheers,

Matt