View Source

OpenHub framework allows to garant processing order of messages. This functionality garants that incoming messages with the same funnel_value will start processing in order by msgTimestamp (timestamp from source system) and next message won't start before previous message isn't finished.

Guaranteed message processing order takes all the following states into consideration: PROCESSING, WAITING, WAITING_FOR_RES, PARTLY_FAILED, FAILED and POSTPONED. FAILED state is used by default but can be excluded.

It can happen that first message failed and then all next messages will wait to start processing (message state will change to POSTPONED again and again).

There is new configuration parameter asynch.postponedIntervalWhenFailed that determines interval (in seconds) after that postponed messages will fail.

Comparison to msg-funnel

Msg-funnel component filters messages in one specific place of the processing but this gauranteed processing order functionality is for wholes routes at the beginning.

There are the following types of message filtering:

classic funnel (msg-funnel component) - filters processing messages (states PROCESSING, WAITING, WAITING_FOR_RES) by funnel_value in specific place of the route, most often before communication with external system to ensure that only one specific value at one time is send to it. For example telco provider can process one request of one specific subscriber (msisdn) at one moment only and we need to filter out requests to this system if there are more requests for one subscriber (msisdn).
classic funnel with guaranteed order (msg-funnel component) - component has optional parameter guaranteedOrder to turn on guaranteed order - messages are filtered out by funnel_value (it's same as classic funnel) and also msgTimestamp of the message is taken into consideration. Messages must be processed at specific funnel in order by msgTimestamp. PARTLY_FAILED, FAILED andPOSTPONED states are used together with processing states
guaranteed message processing order - this functionality that is about whole routes

How to use it?

Use org.cleverbus.api.asynch.AsynchRouteBuilder with methods withGuaranteedOrder() or withGuaranteedOrderWithoutFailed() (see Asynchronous messages for more details) or sets AsynchConstants.GUARANTEED_ORDER_HEADER or AsynchConstants.EXCLUDE_FAILED_HEADER to true when you create route definition manually.

private void createRouteForAsyncHelloRouteIn() throws FailedToCreateRouteException { 
	Expression funnelValueExpr = xpath("/h:asyncHelloRequest/h:name").namespaces(ns).stringResult();
    AsynchRouteBuilder.newInstance(ServiceEnum.HELLO, OPERATION_NAME,
            getInWsUri(new QName(SyncHelloRoute.HELLO_SERVICE_NS, "asyncHelloRequest")),
            new AsynchResponseProcessor() {
                @Override
                protected Object setCallbackResponse(CallbackResponse callbackResponse) {
                    AsyncHelloResponse res = new AsyncHelloResponse();
                    res.setConfirmAsyncHello(callbackResponse);
                    return res;
                }
            }, jaxb(AsyncHelloResponse.class))
			.withFunnelValue(funnelValueExpr)
            .withGuaranteedOrder()
            .build(this);
}