Paralel processing in WSO2 EI 6.2

Question

I'm developing a healthcheck API that needs to call several endpoints. The idea is that, if any of those EPs fail, I have to capture the fault and incorporate that into the response. I thought about creating a main sequence that calls other leaf sequences, and each leaf sequence has it own fault sequence. This diagram should make things clearer:

           seqA (faultSsqA)
         /
mainSeq  - seqB (faultSeqB)
         \ seqC (faultSeqC)

If all the sequences process successfully, everything goes well and I'm able to process the response, however, if there's a fault within one of the leaf sequences, the corresponding faultsequence is executed, but I don't know how to return the flow of execution back to the main sequence. What I want is that, if a exception occurs in sequence B, fault sequence B is executed and then the execution flow goes back to the main sequence and then to sequence C. Even better it would be if I could execute all the leaf sequences in pararel, and just gather all the results in the end. I know that iterate mediator sort of do that, but from what I understand the semantics are not the same of what I want. Anyone has thoughts on this?

EDIT: so it seems the clone mediator is what I'm looking for, however I'm still not able to get it working the way I need to. Since I'll be connecting to different platforms, the response formats are distinct. Also, I need to treat situations such as timeouts, 500 response codes, etc. Therefore, I can't simply call an endpoint from each target - for each backend platform I've created a sequence that does all this logic, and populate some properties so that I can populate the final response ("platform A = up", etc). My expectation would be that all the target sequences would be processed in paralel

This is my main sequence. I use a clone mediator and target it to 2 different sequences, which do the actual invocation of the EP and treat the response (I'm using only 2 for now, but in the future will be many more). The sequences also update some variables to reflect the status of the backend platforms ("platform A = up", etc).

<sequence name="inSequence_healthCheck" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <clone id="healthCHeck">
        <target>
            <sequence>
                <sequence key="healthCheck_wacs"/>
            </sequence>
        </target>
        <target>
            <sequence>
                <sequence key="healthCheck_thesys"/>
            </sequence>
        </target>
    </clone>
    <loopback/>
</sequence>

This is one of the target sequences (it' very simple for now, but there should be more logic in it). Notice that I populate the property wacsStatus depending on the response.

<sequence name="healthCheck_wacs" onError="faultSequence_healthCheck_wacs" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <call>
        <endpoint key="gov:ClientEquipments/endpoints/WACS/wacs_healthCheck.endpoint.xml"/>
    </call>
    <switch source="$axis2:HTTP_SC">
        <case regex="2\d\d|4\d\d">
            <property name="wacsStatus" scope="default" type="BOOLEAN" value="true"/>
        </case>
        <default>
            <property name="wacsStatus" scope="default" type="BOOLEAN" value="false"/>
        </default>
    </switch>
</sequence>

And this is my outSequence, that does the aggregate:

<sequence name="outSequence_healthCheck" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <log>
        <property name="step" value="START outSequence_healthCheck"/>
        <property expression="$ctx:wacsStatus" name="wacsStatus"/>
        <property expression="$ctx:thesysStatus" name="thesysStatus"/>
        <property expression="$ctx:phStatus" name="phStatus"/>
        <property expression="$ctx:naStatus" name="naStatus"/>        
    </log>
  <property name="info" scope="default">
    <ns:Information xmlns:ns="http://wso2.com"/>
  </property>
    <aggregate id="healthCHeck">
        <completeCondition>
            <messageCount max="-1" min="2"/>
        </completeCondition>
        <onComplete enclosingElementProperty="info" expression="s11:Body/child::* | s12:Body/child::*" 
              xmlns:m0="http://services.samples" xmlns:s11="http://schemas.xmlsoap.org/soap/envelope/" 
              xmlns:s12="http://www.w3.org/2003/05/soap-envelope">
          <log level="full"/> 
        </onComplete>

    </aggregate>
          <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
            <payloadFactory media-type="json">
                <format>
            {
                "WACS" : "$1",
                "thesys" : "$2"
            }
                        </format>
                <args>
                    <arg evaluator="xml" expression="$ctx:wacsStatus"/>
                    <arg evaluator="xml" expression="$ctx:thesysStatus"/>
                </args>
            </payloadFactory>
            <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
            <property name="HTTP_SC" scope="axis2" type="STRING" value="200"/>
            <respond/>  
</sequence>

My expectation would be that the aggregate mediator would wait for all the target sequences to execute, by then all the properties would be properly populated and I could use the payloadFactory to properly format the response. However, from the logs I can see that, even though the aggregate does aggregate all the responses from the target sequences, the properties are not properly populated and therefore the response is not correct.

Thanks.

Pedro

Answer 1

You should try an iterate mediator/clone & aggregate mediator (scatter-gather pattern) construction. The iterate allows you to execute multiple sequences in paralel while the aggregate will collect the responses. You can have the separate sequences deal with fault handling and return either the expected response or some fault. You can then check the aggregated result for any faults.

Here are a few samples: Yenlo Blog WSO2 Architecture Team