DZone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
Refcards Trend Reports
Events Video Library
Refcards
Trend Reports

Events

View Events Video Library

Related

  • From Code to Customer: Building Fault-Tolerant Microservices With Observability in Mind
  • Designing Fault-Tolerant Messaging Workflows Using State Machine Architecture
  • Mutation Testing: The Art of Deliberately Introducing Issues in Your Code
  • Architecting for Resilience: Strategies for Fault-Tolerant Systems

Trending

  • Stop Writing If-Else Spaghetti: Architecting Cleaner Java with the Strategy Pattern
  • 12 Factor Framework for Building Secure and Compliant Cloud Applications
  • Common REST Design Pattern
  • Why DDoS Protection Is an Architectural Decision for Developers

Iterate/Aggregate Fault Handling in WSO2 EI

In this article, see how to iterate/aggregate fault handling in WSO2 EI.

By 
Francisco Ribeiro user avatar
Francisco Ribeiro
·
Feb. 19, 20 · Tutorial
Likes (5)
Comment
Save
Tweet
Share
20.7K Views

Join the DZone community and get the full member experience.

Join For Free

In WSO2, we can implement the Splitter and Aggregator EIP using the Iterate and Aggregate mediators. With the Splitter pattern, we can split a message composed by different elements that need to be processed individually, and then we use the Aggregator pattern to aggregate the results of each individual call and then perform some processing over the aggregated results.

Happy Path Example

In a happy path example, all the requests and processing done inside the iterate mediator will occur with no failures, and the aggregate mediator will handle the results of all the requests made. We can see that in the proxy below:

XML
  




xxxxxxxxxx
1
65


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<proxy name="ProxyIterateAggregateFaultNotWorking" startOnLoad="true" transports="http https" xmlns="http://ws.apache.org/ns/synapse">
3 
    <target>
4 
        <inSequence>
5 
            <!-- This is to force going to fault sequence in case of SOAP Fault -->
6 
            <!-- Set this to force a fault in case of SOAPFault returned by the backend service -->
7 
            <payloadFactory media-type="xml">
8 
                <format>
9 
                    <payload xmlns="">
10 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
11 
                            <in>1</in>
12 
                        </echo:echoInt>
13 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
14 
                            <in>1</in>
15 
                        </echo:echoInt>
16 
                    </payload>
17 
                </format>
18 
                <args></args>
19 
            </payloadFactory>
20 
            <!-- 1. Iterate over the echoInt elements -->
21 
            <iterate expression="//echo:echoInt" xmlns:echo="http://echo.services.core.carbon.wso2.org">
22 
                <target>
23 
                    <sequence>
24 
                        <property name="FORCE_ERROR_ON_SOAP_FAULT" scope="default" type="STRING" value="true"></property>
25 
                        <header name="Action" scope="default" value="urn:echoInt"></header>
26 
                        <call>
27 
                            <endpoint>
28 
                                <address format="soap11" uri="http://localhost:8280/services/echo">
29 
                                    <suspendOnFailure>
30 
                                        <initialDuration>-1</initialDuration>
31 
                                        <progressionFactor>-1</progressionFactor>
32 
                                        <maximumDuration>0</maximumDuration>
33 
                                    </suspendOnFailure>
34 
                                    <markForSuspension>
35 
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
36 
                                    </markForSuspension>
37 
                                </address>
38 
                            </endpoint>
39 
                        </call>
40 
                    </sequence>
41 
                </target>
42 
            </iterate>
43 
            <property name="result" scope="default">
44 
                <result xmlns=""></result>
45 
            </property>
46 
            <aggregate>
47 
                <completeCondition>
48 
                    <messageCount max="-1" min="-1"></messageCount>
49 
                </completeCondition>
50 
                <onComplete enclosingElementProperty="result" expression="$body/*[1]">
51 
                    <log level="custom">
52 
                        <property name="ON Aggregate SEQ" value="faultSequence default"></property>
53 
                    </log>
54 
                    <respond></respond>
55 
                </onComplete>
56 
            </aggregate>
57 
        </inSequence>
58 
        <outSequence></outSequence>
59 
        <faultSequence>
60 
            <log level="custom">
61 
                <property name="ON FAULT SEQ" value="faultSequence default"></property>
62 
            </log>
63 
        </faultSequence>
64 
    </target>
65 
</proxy>



This proxy basically creates a payload contaning multiple echoInt requests, and then we iterate over each echoInt to make a request to the echo service. Inside the iterate, we set the Action header and make a request to the endpoint using the call mediator. We added the property FORCE_ERROR_ON_SOAP_FAULT so in case of a SOAPFault returned by the backend service, the flow will be redirected to the fault sequence. 

In this example, there is no error, so the aggregate mediator after the iterate will gather all the responses and aggregate them in a single message and respond back. We can see the response of this Proxy service in the xml below:

XML
  




xxxxxxxxxx
1


 
1 
 <result>
2 
    <ns:echoIntResponse xmlns:ns="http://echo.services.core.carbon.wso2.org">
3 
        <return>1</return>
4 
    </ns:echoIntResponse>
5 
    <ns:echoIntResponse xmlns:ns="http://echo.services.core.carbon.wso2.org">
6 
        <return>1</return>
7 
    </ns:echoIntResponse>
8 
</result>  



You may also want to read: Iterative Processing Using the For Each Scope in Mule

Fault Example

In the proxy below, we made a small change in the payloadFactory in order to force a soap fault from the backend:

XML
  




xxxxxxxxxx
1
13


 
1 
<payloadFactory media-type="xml">
2 
    <format>
3 
        <payload xmlns="">
4 
            <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
5 
                <in>1</in>
6 
            </echo:echoInt>
7 
            <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
8 
                <in>abc</in>
9 
            </echo:echoInt>
10 
        </payload>
11 
    </format>
12 
    <args></args>
13 
</payloadFactory>



We are passing the value ‘abc’ in the second echoInt element. With this example, when we try the proxy, we are not going to receive any response as the flow is redirected to the faultSequence, and we can see the log entry:

[2020-02-16 21:32:41,526] [EI-Core]  INFO - LogMediator ON FAULT SEQ = faultSequence default

The full proxy service code can be seen below:

XML
  




xxxxxxxxxx
1
65


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<proxy name="ProxyIterateAggregateFaultNotWorking" startOnLoad="true" transports="http https" xmlns="http://ws.apache.org/ns/synapse">
3 
    <target>
4 
        <inSequence>
5 
            <!-- This is to force going to fault sequence in case of SOAP Fault -->
6 
            <!-- Set this to force a fault in case of SOAPFault returned by the backend service -->
7 
            <payloadFactory media-type="xml">
8 
                <format>
9 
                    <payload xmlns="">
10 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
11 
                            <in>1</in>
12 
                        </echo:echoInt>
13 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
14 
                            <in>abc</in>
15 
                        </echo:echoInt>
16 
                    </payload>
17 
                </format>
18 
                <args></args>
19 
            </payloadFactory>
20 
            <!-- 1. Iterate over the echoInt elements -->
21 
            <iterate expression="//echo:echoInt" xmlns:echo="http://echo.services.core.carbon.wso2.org">
22 
                <target>
23 
                    <sequence>
24 
                        <property name="FORCE_ERROR_ON_SOAP_FAULT" scope="default" type="STRING" value="true"></property>
25 
                        <header name="Action" scope="default" value="urn:echoInt"></header>
26 
                        <call>
27 
                            <endpoint>
28 
                                <address format="soap11" uri="http://localhost:8280/services/echo">
29 
                                    <suspendOnFailure>
30 
                                        <initialDuration>-1</initialDuration>
31 
                                        <progressionFactor>-1</progressionFactor>
32 
                                        <maximumDuration>0</maximumDuration>
33 
                                    </suspendOnFailure>
34 
                                    <markForSuspension>
35 
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
36 
                                    </markForSuspension>
37 
                                </address>
38 
                            </endpoint>
39 
                        </call>
40 
                    </sequence>
41 
                </target>
42 
            </iterate>
43 
            <property name="result" scope="default">
44 
                <result xmlns=""></result>
45 
            </property>
46 
            <aggregate>
47 
                <completeCondition>
48 
                    <messageCount max="-1" min="-1"></messageCount>
49 
                </completeCondition>
50 
                <onComplete enclosingElementProperty="result" expression="$body/*[1]">
51 
                    <log level="custom">
52 
                        <property name="ON Aggregate SEQ" value="faultSequence default"></property>
53 
                    </log>
54 
                    <respond></respond>
55 
                </onComplete>
56 
            </aggregate>
57 
        </inSequence>
58 
        <outSequence></outSequence>
59 
        <faultSequence>
60 
            <log level="custom">
61 
                <property name="ON FAULT SEQ" value="faultSequence default"></property>
62 
            </log>
63 
        </faultSequence>
64 
    </target>
65 
</proxy>



Fault Handling With Iterate/Aggregate

In order to still have the aggregate to work when having a fault, we need to make some changes to proxy and make use of sequences. After the changes we are going to have the following artifacts:

  • The Proxy Service
  • Iterate sequence
  • Aggregate sequence
  • IterateFaultHandler sequence

The proxy service code can be seen below:

XML
  




xxxxxxxxxx
1
37


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<proxy name="ProxyIterateAggregateWorking" startOnLoad="true" transports="http https" xmlns="http://ws.apache.org/ns/synapse">
3 
    <target>
4 
        <inSequence>
5 
            <!-- This is to force going to fault sequence in case of SOAP Fault -->
6 
            <!-- Set this to force a fault in case of SOAPFault returned by the backend service -->
7 
            <payloadFactory media-type="xml">
8 
                <format>
9 
                    <payload xmlns="">
10 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
11 
                            <in>1</in>
12 
                        </echo:echoInt>
13 
                        <echo:echoInt xmlns:echo="http://echo.services.core.carbon.wso2.org">
14 
                            <in>abc</in>
15 
                        </echo:echoInt>
16 
                    </payload>
17 
                </format>
18 
                <args></args>
19 
            </payloadFactory>
20 
            <!-- 1. Iterate over the echoInt elements -->
21 
            <iterate expression="//echo:echoInt" xmlns:echo="http://echo.services.core.carbon.wso2.org">
22 
                <target>
23 
                    <sequence>
24 
                        <sequence key="IterateSequence"></sequence>
25 
                    </sequence>
26 
                </target>
27 
            </iterate>
28 
            <sequence key="AggregateSequence"></sequence>
29 
        </inSequence>
30 
        <outSequence></outSequence>
31 
        <faultSequence>
32 
            <log level="custom">
33 
                <property name="ON FAULT SEQ" value="faultSequence default"></property>
34 
            </log>
35 
        </faultSequence>
36 
    </target>
37 
</proxy>



The main difference from the previous proxy is that inside the iterate, we are using a predefined sequence instead of an anonymous sequence. And we have the AggregateSequence just after the iterate mediator.

The AggregateSequence just contains the code that we had previously inside the proxy:

XML
  




xxxxxxxxxx
1
17


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<sequence name="AggregateSequence" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
3 
    <property name="result" scope="default">
4 
        <result xmlns=""></result>
5 
    </property>
6 
    <aggregate>
7 
        <completeCondition>
8 
            <messageCount max="-1" min="-1"></messageCount>
9 
        </completeCondition>
10 
        <onComplete enclosingElementProperty="result" expression="$body/*[1]">
11 
            <log level="custom">
12 
                <property name="ON Aggregate SEQ" value="faultSequence default"></property>
13 
            </log>
14 
            <respond></respond>
15 
        </onComplete>
16 
    </aggregate>
17 
</sequence>



There is nothing special in that sequence.

Now let us see the IterateSequence:

XML
  




xxxxxxxxxx
1
20


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<sequence name="IterateSequence" onError="IterateFaultHandler" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
3 
    <!-- Set this to force a fault in case of SOAPFault returned by the backend service -->
4 
    <property name="FORCE_ERROR_ON_SOAP_FAULT" scope="default" type="STRING" value="true"></property>
5 
    <header name="Action" scope="default" value="urn:echoInt"></header>
6 
    <call>
7 
        <endpoint>
8 
            <address format="soap11" uri="http://localhost:8280/services/echo">
9 
                <suspendOnFailure>
10 
                    <initialDuration>-1</initialDuration>
11 
                    <progressionFactor>-1</progressionFactor>
12 
                    <maximumDuration>0</maximumDuration>
13 
                </suspendOnFailure>
14 
                <markForSuspension>
15 
                    <retriesBeforeSuspension>0</retriesBeforeSuspension>
16 
                </markForSuspension>
17 
            </address>
18 
        </endpoint>
19 
    </call>
20 
</sequence>



It contains the same code that we had in the proxy, but the tricky part is in the onError property of the sequence where we specify which sequence should be executed in case of errors in the sequence, that in this case is the IterateFaultHandler. So, in this case, when it tries to execute the echoInt with the abc value it will receive the SOAPFault and will redirect the flow to the IterateFaultHandler.

So, now let us see the IterateFaultHandler:

XML
  




xxxxxxxxxx
1
18


 
1 
<?xml version="1.0" encoding="UTF-8"?>
2 
<sequence name="IterateFaultHandler" onError="IterateFaultHandler" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
3 
    <log level="custom">
4 
        <property name="ON ITERATE FAULT SEQ" value="IterateFaultHandler"></property>
5 
    </log>
6 
    <payloadFactory media-type="xml">
7 
        <format>
8 
            <errorProcessing xmlns="">
9 
                <error>$1</error>
10 
            </errorProcessing>
11 
        </format>
12 
        <args>
13 
            <arg evaluator="xml" expression="//faultstring"></arg>
14 
        </args>
15 
    </payloadFactory>
16 
    <property name="RESPONSE" scope="default" type="STRING" value="true"></property>
17 
    <sequence key="AggregateSequence"></sequence>
18 
</sequence>



Inside this fault sequence, we create an error payload with the faultstring. Then we set the RESPONSE property to true, this indicates to the synapse engine that this message is in the response direction. And then we call the AggregateSequence. We need to specify the RESPONSE property because the aggregate mediator expects the message flow to be in the response direction.

So, now when we try the proxy it will receive a fault for the element with the invalid value and then redirect to the fault sequence, it will generate that error payload and calls the aggregate. We can see below the error log indicating that it executed the fault sequence and then the final payload:

[2020-02-16 21:55:48,155] [EI-Core]  INFO - LogMediator ON ITERATE FAULT SEQ = IterateFaultHandler

The response payload:

XML
  




xxxxxxxxxx
1


 
1 
<result>
2 
    <ns:echoIntResponse xmlns:ns="http://echo.services.core.carbon.wso2.org">
3 
        <return>1</return>
4 
    </ns:echoIntResponse>
5 
    <errorProcessing>
6 
        <error>Invalid value "abc" for element in</error>
7 
    </errorProcessing>
8 
</result>



With this approach, the aggregate is always executed and the flow finishes without any issues.

You may ask, “Why not call the AggregateSequence inside the Proxy faultSequence?” When we try using the AggregateSequence inside the faultSequence of the proxy, it will throw the error below and that does not happen when we use a named sequence for the fault handling:

Java
  




xxxxxxxxxx
1
25


 
1 
[2020-02-16 21:59:54,782] [EI-Core] ERROR - SequenceMediator Runtime error occurred while mediating the message
2 
java.util.EmptyStackException
3 
    at java.util.Stack.peek(Stack.java:102)
4 
    at org.apache.synapse.mediators.eip.aggregator.AggregateMediator.mediate(AggregateMediator.java:302)
5 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:109)
6 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:71)
7 
    at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158)
8 
    at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:214)
9 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:109)
10 
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:71)
11 
    at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158)
12 
    at org.apache.synapse.mediators.MediatorFaultHandler.onFault(MediatorFaultHandler.java:96)
13 
    at org.apache.synapse.FaultHandler.handleFault(FaultHandler.java:53)
14 
    at org.apache.synapse.endpoints.AbstractEndpoint.invokeNextFaultHandler(AbstractEndpoint.java:735)
15 
    at org.apache.synapse.endpoints.AbstractEndpoint.onFault(AbstractEndpoint.java:550)
16 
    at org.apache.synapse.endpoints.AddressEndpoint.onFault(AddressEndpoint.java:46)
17 
    at org.apache.synapse.FaultHandler.handleFault(FaultHandler.java:101)
18 
    at org.apache.synapse.core.axis2.SynapseCallbackReceiver.handleMessage(SynapseCallbackReceiver.java:527)
19 
    at org.apache.synapse.core.axis2.SynapseCallbackReceiver.receive(SynapseCallbackReceiver.java:195)
20 
    at org.apache.axis2.engine.AxisEngine.receive(AxisEngine.java:180)
21 
    at org.apache.synapse.transport.passthru.ClientWorker.run(ClientWorker.java:265)
22 
    at org.apache.axis2.transport.base.threads.NativeWorkerPool$1.run(NativeWorkerPool.java:172)
23 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
24 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
25 
    at java.lang.Thread.run(Thread.java:748)



I hope this helps! See you in the next post.

Further Reading

Aggregate Millions of Database Rows in a Spring Controller

Fault (technology)

Published at DZone with permission of Francisco Ribeiro. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

  • From Code to Customer: Building Fault-Tolerant Microservices With Observability in Mind
  • Designing Fault-Tolerant Messaging Workflows Using State Machine Architecture
  • Mutation Testing: The Art of Deliberately Introducing Issues in Your Code
  • Architecting for Resilience: Strategies for Fault-Tolerant Systems

Partner Resources

×
Comments

The likes didn't load as expected. Please refresh the page and try again.

  • RSS
  • X
  • Facebook

ABOUT US

  • About DZone
  • Support and feedback
  • Community research

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • Become a Contributor
  • Core Program
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 3343 Perimeter Hill Drive
  • Suite 215
  • Nashville, TN 37211
  • [email protected]

Let's be friends:

  • RSS
  • X
  • Facebook