How to check if processing is complete when using message store reapers?

[matt.friedman]

I need the application's main thread to sleep for some time while the reapers clear the remaining groups in the SimpleMessageStore.

I've come to the conclusion that when the message stores are empty I can consider the processing mostly complete. I may sleep for few more seconds just to be sure.

I am trying out some code that looks like this.

        while(referencedObjectsMessageStore.iterator().hasNext() || batchMessageStore.iterator().hasNext()) {
            System.out.println("Checking if the message stores are cleared...");
            Thread.sleep(1000);
        }

The system is end to end, that is, there is no reply channel. The results of the processing are written to a file.

Does this seems reasonable? Is there another approach?

Many Thanks,
Matt

[Gary Russell]

Actually, you can set expireOnDestroy to true on the reaper and he will expire all groups during context shutdown (context.close()).

[matt.friedman]

That is good to know but in this case I have a few different object streams that run sequentially and I want each one to start after the previous one has completed. So expireOnDestroy would only work for the final stream.

I do know that there is a facility for request-response but I'm not sure how that works; it may or may not apply to my situation, I'll have to see. After a quick look at the documentation the outbound-gateway concept doesn't seem to apply, so perhaps that wouldn't help me in this case? I currently use <outbound-channel-adapter channel="jsonStringChannel" ref="jsonStringHandler"/> to send the results to the bean that does the write() operations.

For now, checking the message stores for remaining items and then sleeping for a few more seconds seems to work.

But any other thoughts would be greatly appreciated.

Thanks,
Matt

[mgirard]

We had to do something similar to this - but decided to do something like a claim check type pattern (assuming you have some sort of split/aggregate route). This way if we ended up scaling processing to multiple machines, we could check for 'is done' on individual records regardless of where they went.

Basically, as a record began processing we created an "Activity Record" in the DB for that item (based on it's primary key) with a start time, then did the split and began processing the sub-items. Then we had an aggregator collecting all messages at the end of processing, grouping on that record id. Once the group was complete, it would flag the activity record complete. Or if the group had timed out, it would flag it as errored.

... -> s.a.:create activity record -> split ->queue -> (multiple process routes) -> aggregate -> s.a.:Mark activity complete/error -> ...

[matt.friedman]

Really interesting. We may need to scale in this manner as well. Thank you very much.

[mgirard]

The one thing I forgot to mention w/ this split/aggregate stuff is that if a split message fails somewhere, you have to remember to handle the error and pass something on to the aggregator - otherwise the group won't complete until it times out. I was reminded of it the hard way when I couldn't figure out why I was getting weird group behavior in testing.