Hasitha's Tech Blessings

Done by server Worker thread. 

• When "send mediator" is used, FlexibleMEPClient send out the message from synapse layer down to Axis2 layer. 
• This call lands onto the HTTP sender mentioned in axis2.xml - by defalut PasthrougHTTPSender. It has invike() method. It is hit. 
• PasthrougHTTPSender passes the messageContext to deliveryAgent to send message. 
• Message is added to a queue here until a connection is free to send this up. 

    /**
     * This Map holds the messages that need to be delivered. But at the moment maximum
     * number of connections to the host:pair is being used. So these messages has to wait
     * until a new connection is available.
     */
    private Map<HttpRoute, Queue<MessageContext>> waitingMessages =
            new ConcurrentHashMap<HttpRoute, Queue<MessageContext>>();

serverHandler Thread returns here to pick up next one. 

msgContext.getOperationContext().setProperty(
                Constants.RESPONSE_WRITTEN, Constants.VALUE_TRUE);
return InvocationResponse.CONTINUE;

Done by IO Reactor threads 

• Above messages in waiting queue is written to the wire using IO Reactoer threads. Target handler’s outputReady method is called. 

WSO2 ESB uses Apache Synapse as its mediation engine. Synapse has a layered architecture to mediate messages through it.

Following is a high level component diagram of Synapse design.

This post walks you through main steps done in synapse to mediate messages. It is more of a code reference that gives a high level understanding of message flow so that one can debug the code and understand.

Contents

Where is the code?

Proxy Configuration Used for Debugging 

Inflow of a message

    Entry point

    Creating synapse mc out of axis2 mc

    Check if message is coming from Inbound Endpoint 

    Set service level log (per proxy logs)

    Honour isSoapConversionRequired parameter

    Execute synapse handlers

    Check for mandatory sequence

    Start mediating message with SequenceMediator

    Building the message

    Iterate through mediators and exit where necessary

    Forwarding message to the Back-end in in-sequence

    Send message to endpoint (send mediator debug)

    Axis2SynapseEnvironment.send() Method (to send out any message from synapse)

    Axis2Sender.sendOn(endpoint, synCtx) method (send message to an endpoint)

    Axis2FlexibleMEPClient.send() method 

    Register the callback for response

    Write to wire so that message goes to BE service

Outflow of a message

    SynapseCallbackReceiver

    handleMessage() method

    Inject message to synapse environment (Axis2SynapseEnvironment)

    Using send mediator in out sequence 

    Dispatching message to Axis2 to send back

Where is the code?

You can clone the code from GitHub from here. Let us look at the message flow inside synapse core.
Project can be built using Maven.

Proxy Configuration Used for Debugging 

<?xml version="1.0" encoding="UTF-8"?>
<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="debugProxy"
       transports="http https"
       startOnLoad="true"
       trace="disable">
   <description/>
   <target inSequence="backend_call_seq">
      <outSequence>
         <send/>
      </outSequence>
      <faultSequence>
         <log>
            <property name="FAULT_OCCURRED" value="BBB"/>
         </log>
      </faultSequence>
   </target>
</proxy>

<?xml version="1.0" encoding="UTF-8"?>
<sequence xmlns="http://ws.apache.org/ns/synapse" name="backend_call_seq">
   <log level="full">
      <property name="INCOMING_MSG" value="AAA"/>
   </log>
   <send>
      <endpoint>
         <address uri="http://localhost:9000/services/SimpleStockQuoteService"/>
      </endpoint>
   </send>
</sequence>

Inflow of a message

Entry point

The entry points of the message into synapse engine from Axis2 layer are

1. ProxyServiceMessageReceiver - when a message hits a proxy service

2. SynapseMessageReceiver - when a message hits a ESB API/main sequence

In this example ProxyServiceMessageReceiver is hit as a proxy is invoked. 

public void receive(org.apache.axis2.context.MessageContext mc) throws AxisFault {..}

Creating synapse mc out of axis2 mc

Check if message is coming from Inbound Endpoint 

Iterate through mediators and exit where necessary

 handleMessage() method

Using send mediator in out sequence 

Dispatching message to Axis2 to send back

package mBeanServer;

public class Customer {

    private String name;
    private String address;
    private int age;
    private long amountPaid;
    private boolean isMarried;

    public Customer(String name, String address, int age, long amountPaid, boolean isMarried) {
        this.name = name;
        this.address = address;
        this.age = age;
        this.amountPaid = amountPaid;
        this.isMarried = isMarried;
    }

    public String getName() {
        return name;
    }

    public String getAddress() {
        return address;
    }

    public int getAge() {
        return age;
    }

    public long getAmountPaid() {
        return amountPaid;
    }

    public boolean isMarried() {
        return isMarried;
    }

    public String toString() {
        return "name: " + name
        + "address:" + address
        + "age:" + age
        + "amount paid:" + amountPaid
        + "is married: " + isMarried;
    }
}

Create a Object factory to create customers

package mBeanServer;

import java.util.ArrayList;
import java.util.List;

public class CustomerFactory {

    public static List generate() {

        Customer customer1 = new Customer("Hasitha", "Kandy, Sri Lanka", 30, 2000, false);
        Customer customer2 = new Customer("Asanka", "Bandarawatte, Kaluthara, Sri Lanka,", 27, 3000, false);
        Customer customer3 = new Customer("Perera", "Colombo, Sri Lanka", 25, 400, false);

        List customers = new ArrayList<>();
        customers.add(customer1);
        customers.add(customer2);
        customers.add(customer3);

        return customers;
    }
}

Create an interface for MBean

package ;
public interface YMBean {...}

package ;
public class Y implements YMBean {...}

package mBeanServer;

import javax.management.MBeanException;
import javax.management.openmbean.CompositeData;

public interface CustomerInfoProviderMBean {

    String NAME = "name";
    String ADDRESS = "address";
    String AGE = "age";
    String AMOUNT_PAID = "amountPaid";
    String IS_MARRIED = "isMarried";

    CompositeData[] getCustomerInformation() throws MBeanException;

}

Implement MBean provider interface

Create CompositeDataSupport objects and add to CompositeData[]

    @Override
    public CompositeData[] getCustomerInformation() throws MBeanException {
        List compositeDataList = new ArrayList<>();
        List listOfCustomers = CustomerFactory.generate();
        try {
            for (Customer customer : listOfCustomers) {
                Object[] itemValues = new Object[]{customer.getName(),
                        customer.getAddress(),
                        customer.getAge(),
                        customer.getAmountPaid(),
                        customer.isMarried()};
                CompositeDataSupport support = new CompositeDataSupport(getCustomerDataType(), customerAttributeNames(),
                        itemValues);
                compositeDataList.add(support);
            }
        } catch (OpenDataException e) {
            throw new MBeanException(e, "Error occurred when getting customer information via JMX");
        }
        return compositeDataList.toArray(new CompositeData[compositeDataList.size()]);
    }

getCustomerDataType(), customerAttributeNames()

• Likewise, if application retained objects permanently in memory over long run, and always keep to occupy a memory near to heap limit, GC happens frequently. 
• As objects keep memory (due to some memory leak, or a cache not cleared for a longer time by application),  the amount of memory GC can recover keeps getting low in every GC cycle. 
• There, if load (usually a high load) put to the server creates temporary objects in memory so fast that GC operations cannot sweep them out, GC overhead limit exceeded error will be thrown. 

(Java) VisualVM  + Visual GC

1.  Run wso2 server by [HOME]/bin wso2server.sh script. 
2. Use "jps" command to find "virtual machine ID"
3. Use that virtual machine ID with jstat command as below

• -XX:+PrintGCDetails (enabled by default)
• -XX:+PrintGCTimeStamps
• -XX:+PrintHeapAtGC 
• -XX:+PrintGCDateStamps (from JDK 6 update 4)

[GC [<collector>: <starting occupancy1> -> <ending occupancy1>, <pause time1> secs] <starting occupancy3> -> <ending occupancy3>, <pause time3> secs]

{Heap before GC invocations=12 (full 0):
 PSYoungGen      total 272384K, used 238314K [0x00000007eaa80000, 0x0000000800000000, 0x0000000800000000)
  eden space 200704K, 100% used [0x00000007eaa80000,0x00000007f6e80000,0x00000007f6e80000)
  from space 71680K, 52% used [0x00000007f6e80000,0x00000007f933a948,0x00000007fb480000)
  to   space 69632K, 0% used [0x00000007fbc00000,0x00000007fbc00000,0x0000000800000000)
 ParOldGen       total 341504K, used 128961K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 37% used [0x00000007c0000000,0x00000007c7df05b0,0x00000007d4d80000)
 PSPermGen       total 36352K, used 36237K [0x00000007b0000000, 0x00000007b2380000, 0x00000007c0000000)
  object space 36352K, 99% used [0x00000007b0000000,0x00000007b2363598,0x00000007b2380000)
2016-11-20T14:30:29.565-0530: 11.921: [GC [PSYoungGen: 238314K->39257K(270848K)] 367275K->170403K(612352K), 0.0190040 secs] [Times: user=0.06 sys=0.00, real=0.02 secs] 
Heap after GC invocations=12 (full 0):
 PSYoungGen      total 270848K, used 39257K [0x00000007eaa80000, 0x00000007ff900000, 0x0000000800000000)
  eden space 208384K, 0% used [0x00000007eaa80000,0x00000007eaa80000,0x00000007f7600000)
  from space 62464K, 62% used [0x00000007fbc00000,0x00000007fe256760,0x00000007ff900000)
  to   space 67072K, 0% used [0x00000007f7600000,0x00000007f7600000,0x00000007fb780000)
 ParOldGen       total 341504K, used 131145K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 38% used [0x00000007c0000000,0x00000007c80125b0,0x00000007d4d80000)
 PSPermGen       total 36352K, used 36237K [0x00000007b0000000, 0x00000007b2380000, 0x00000007c0000000)
  object space 36352K, 99% used [0x00000007b0000000,0x00000007b2363598,0x00000007b2380000)
}
{Heap before GC invocations=13 (full 0):
 PSYoungGen      total 270848K, used 247641K [0x00000007eaa80000, 0x00000007ff900000, 0x0000000800000000)
  eden space 208384K, 100% used [0x00000007eaa80000,0x00000007f7600000,0x00000007f7600000)
  from space 62464K, 62% used [0x00000007fbc00000,0x00000007fe256760,0x00000007ff900000)
  to   space 67072K, 0% used [0x00000007f7600000,0x00000007f7600000,0x00000007fb780000)
 ParOldGen       total 341504K, used 131145K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 38% used [0x00000007c0000000,0x00000007c80125b0,0x00000007d4d80000)
 PSPermGen       total 37376K, used 37048K [0x00000007b0000000, 0x00000007b2480000, 0x00000007c0000000)
  object space 37376K, 99% used [0x00000007b0000000,0x00000007b242e080,0x00000007b2480000)
2016-11-20T14:30:30.252-0530: 12.607: [GC [PSYoungGen: 247641K->14407K(275456K)] 378787K->145734K(616960K), 0.0191150 secs] [Times: user=0.03 sys=0.00, real=0.02 secs] 
Heap after GC invocations=13 (full 0):
 PSYoungGen      total 275456K, used 14407K [0x00000007eaa80000, 0x0000000800000000, 0x0000000800000000)
  eden space 208384K, 0% used [0x00000007eaa80000,0x00000007eaa80000,0x00000007f7600000)
  from space 67072K, 21% used [0x00000007f7600000,0x00000007f8411f40,0x00000007fb780000)
  to   space 69120K, 0% used [0x00000007fbc80000,0x00000007fbc80000,0x0000000800000000)
 ParOldGen       total 341504K, used 131327K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 38% used [0x00000007c0000000,0x00000007c803fc50,0x00000007d4d80000)
 PSPermGen       total 37376K, used 37048K [0x00000007b0000000, 0x00000007b2480000, 0x00000007c0000000)
  object space 37376K, 99% used [0x00000007b0000000,0x00000007b242e080,0x00000007b2480000)
}
{Heap before GC invocations=14 (full 0):
 PSYoungGen      total 275456K, used 222791K [0x00000007eaa80000, 0x0000000800000000, 0x0000000800000000)
  eden space 208384K, 100% used [0x00000007eaa80000,0x00000007f7600000,0x00000007f7600000)
  from space 67072K, 21% used [0x00000007f7600000,0x00000007f8411f40,0x00000007fb780000)
  to   space 69120K, 0% used [0x00000007fbc80000,0x00000007fbc80000,0x0000000800000000)
 ParOldGen       total 341504K, used 131327K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 38% used [0x00000007c0000000,0x00000007c803fc50,0x00000007d4d80000)
 PSPermGen       total 39936K, used 39471K [0x00000007b0000000, 0x00000007b2700000, 0x00000007c0000000)
  object space 39936K, 98% used [0x00000007b0000000,0x00000007b268bd60,0x00000007b2700000)
2016-11-20T14:30:30.924-0530: 13.280: [GC [PSYoungGen: 222791K->3696K(274432K)] 354118K->135555K(615936K), 0.0038710 secs] [Times: user=0.02 sys=0.00, real=0.00 secs] 
Heap after GC invocations=14 (full 0):
 PSYoungGen      total 274432K, used 3696K [0x00000007eaa80000, 0x0000000800000000, 0x0000000800000000)
  eden space 205312K, 0% used [0x00000007eaa80000,0x00000007eaa80000,0x00000007f7300000)
  from space 69120K, 5% used [0x00000007fbc80000,0x00000007fc01c390,0x0000000800000000)
  to   space 72192K, 0% used [0x00000007f7300000,0x00000007f7300000,0x00000007fb980000)
 ParOldGen       total 341504K, used 131859K [0x00000007c0000000, 0x00000007d4d80000, 0x00000007eaa80000)
  object space 341504K, 38% used [0x00000007c0000000,0x00000007c80c4c18,0x00000007d4d80000)
 PSPermGen       total 39936K, used 39471K [0x00000007b0000000, 0x00000007b2700000, 0x00000007c0000000)
  object space 39936K, 98% used [0x00000007b0000000,0x00000007b268bd60,0x00000007b2700000)
}

1. If above monitoring tools tells you that there is not enough memory to handle the load put to the server, there is an actual resource limitation. In that case you need to increase heap memory size allocated to JVM. The load might be creating too many temporary objects. And in long run some objects are kept in memory by application for some time, making the situation worse. Following is how we can update memory allocation using JVM parameters. Xms - for minimum limit. Xmx - for maximum limit
   
2.  If monitoring tells you that there is a memory leak keeping objects in memory in long run, preventing GC to recover memory (even running frequently when memory brim is reaching), need to fix the memory leak on the first place. 
3. If load is considerably low and there is enough memory allocated, need to revisit the code how temporary objects are created and used by the application. Choosing to reuse objects instead of creating new ones can reduce the number of times the garbage collector gets in the way.
4. GC tuning might be an option as well. Frequent GC from beginning might solve this particular problem given that there is no memory leak (-XX:-UseConcMarkSweepGC will be an option).