Fuse ESB - Programing EIP Components - Asynchronous Processing

Overview

Producer endpoints normally follow a synchronous pattern when processing an exchange. When the preceding processor in a pipeline calls process() on a producer, the process() method blocks until a reply is received. In this case, the processor's thread remains blocked until the producer has completed the cycle of sending the request and receiving the reply.

Sometimes, however, you might prefer to decouple the preceding processor from the producer, so that the processor's thread is released immediately and the process() call does not block. In this case, you should implement the producer using an asynchronous pattern, which gives the preceding processor the option of invoking a non-blocking version of the process() method.

To give you an overview of the different implementation options, this section describes both the synchronous and the asynchronous patterns for implementing a producer endpoint.

Synchronous producer

Figure 5.6 shows an outline of a synchronous producer, where the preceding processor blocks until the producer has finished processing the exchange.

Figure 5.6. Synchronous Producer

The synchronous producer processes an exchange as follows:

The preceding processor in the pipeline calls the synchronous process() method on the producer to initiate synchronous processing. The synchronous process() method takes a single exchange argument.
In the body of the process() method, the producer sends the request (In message) to the endpoint.
If required by the exchange pattern, the producer waits for the reply (Out message) to arrive from the endpoint. This step can cause the process() method to block indefinitely. However, if the exchange pattern does not mandate a reply, the process() method can return immediately after sending the request.
When the process() method returns, the exchange object contains the reply from the synchronous call (an Out message message).

Asynchronous producer

Figure 5.7 shows an outline of an asynchronous producer, where the producer processes the exchange in a sub-thread, and the preceding processor is not blocked for any significant length of time.

Figure 5.7. Asynchronous Producer

The synchronous producer processes an exchange as follows:

Before the processor can call the asynchronous process() method, it must create an asynchronous callback object, which is responsible for processing the exchange on the return portion of the route. For the asynchronous callback, the processor must implement a class that inherits from the AsyncCallback interface.
The processor calls the asynchronous process() method on the producer to initiate asynchronous processing. The asynchronous process() method takes two arguments:
- an exchange object
- a synchronous callback object
In the body of the process() method, the producer creates a Runnable object that encapsulates the processing code. The producer then delegates the execution of this Runnable object to a sub-thread.
The asynchronous process() method returns, thereby freeing up the processor's thread. The exchange processing continues in a separate sub-thread.
The Runnable object sends the In message to the endpoint.
If required by the exchange pattern, the Runnable object waits for the reply (Out or Fault message) to arrive from the endpoint. The Runnable object remains blocked until the reply is received.
After the reply arrives, the Runnable object inserts the reply (Out message) into the exchange object and then calls done() on the asynchronous callback object. The asynchronous callback is then responsible for processing the reply message (executed in the sub-thread).