Scaling up Mule with Async Request Handling/Continuations

AhcProcessor(1).java

package org.ossandme;

...

// Naive implementation
public class AhcProcessor extends AbstractInterceptingMessageProcessor {

    @Override
    public MuleEvent process(final MuleEvent event) throws MuleException {
        ...    
    }

}

AhcProcessor(2).java

package org.ossandme;

...

// Naive implementation
public class AhcProcessor extends AbstractInterceptingMessageProcessor {

    @Override
    public MuleEvent process(final MuleEvent event) throws MuleException {
        final CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
        final HttpGet request = new HttpGet("http://www.google.com/");

        httpclient.start();

        httpclient.execute(request, new FutureCallback<HttpResponse>() {

            public void completed(final HttpResponse response) {
                try {
                    InputStream input = response.getEntity().getContent();
                    ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
                    IOUtils.copy(input, byteOut);
                    event.setMessage(new DefaultMuleMessage(byteOut.toString(), event.getMessage().getMuleContext()));
                    final MuleEvent lastEvent = processNext(event);
                    lastEvent.getReplyToHandler().processReplyTo(lastEvent, lastEvent.getMessage(), event.getReplyToDestination());
                } catch (Exception e) {
                    throw new RuntimeException();
                }
            }

            public void failed(final Exception ex) {
                System.out.println(request.getRequestLine() + "->" + ex);
            }

            public void cancelled() {
                System.out.println(request.getRequestLine() + " cancelled");
            }

        });

        ...
    }

}

AhcProcessor(3).java

package org.ossandme;

...

// Naive implementation
public class AhcProcessor extends AbstractInterceptingMessageProcessor {

    @Override
    public MuleEvent process(final MuleEvent event) throws MuleException {
        final CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
        final HttpGet request = new HttpGet("http://www.google.com/");

        httpclient.start();

        httpclient.execute(request, new FutureCallback<HttpResponse>() {

            public void completed(final HttpResponse response) {
                try {
                    InputStream input = response.getEntity().getContent();
                    ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
                    IOUtils.copy(input, byteOut);
                    event.setMessage(new DefaultMuleMessage(byteOut.toString(), event.getMessage().getMuleContext()));
                    final MuleEvent lastEvent = processNext(event);
                    lastEvent.getReplyToHandler().processReplyTo(lastEvent, lastEvent.getMessage(), event.getReplyToDestination());
                } catch (Exception e) {
                    throw new RuntimeException();
                }
            }

            public void failed(final Exception ex) {
                System.out.println(request.getRequestLine() + "->" + ex);
            }

            public void cancelled() {
                System.out.println(request.getRequestLine() + " cancelled");
            }

        });

        return null;
    }

}

mule-config(1).xml

<mule ...>
 
    <flow name="...">
        <http:inbound-endpoint address="..." exchange-pattern="request-response"/>
        <async>
            <http:outbound-endpoint address="..." exchange-pattern="request-response"/>
        </async>
        ...
    </flow>

</mule>

mule-config(2).xml

<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      ...
      xmlns:jetty="http://www.mulesoft.org/schema/mule/jetty"
      xsi:schemaLocation="
        ...
        http://www.mulesoft.org/schema/mule/jetty http://www.mulesoft.org/schema/mule/jetty/current/mule-jetty.xsd">

    <jetty:connector name="test" useContinuations="true"/>

    <flow name="main">
        <jetty:inbound-endpoint address="http://localhost:8081" exchange-pattern="request-response"/>
        ...
    </flow>
</mule>

mule-config(3).xml

<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      ...
      xmlns:jetty="http://www.mulesoft.org/schema/mule/jetty"
      xsi:schemaLocation="
        ...
        http://www.mulesoft.org/schema/mule/jetty http://www.mulesoft.org/schema/mule/jetty/current/mule-jetty.xsd">

    <jetty:connector name="test" useContinuations="true"/>

    <flow name="main">
        <jetty:inbound-endpoint address="http://localhost:8081" exchange-pattern="request-response"/>
        <custom-processor class="org.ossandme.AhcProcessor"/>
        <logger level="INFO" message="Reply from service received" />
        ...
    </flow>
</mule>