This article demonstrates the capability of the UltraESB to split messages and aggregate responses, and in particular highlights the capability to stream chunks of responses aggregated as-soon-as-they-become-available; while handling premature timeouts thus preventing the remote client from receiving a partial response on a timeout.
Note: This is an advanced example solving a complex scenario, and was driven by a user request for a proof-of-concept.
Scenario - Problem definition
The scenario simulates multiple concurrent users issuing JSON requests of the form given below, with a request to multiply two variables v1 and v2, along with a text message:
There are three backend service endpoints servicing such JSON requests, but each expecting a slightly different message format. Thus each such message by a user must be cloned and transformed into a format suitable to each destination backend systems In this example, the format expected by the backend services are:
Each backend JSON service responds to such a request within a random delay of 1200ms as follows, adding a "result" as v1 * v2:
These responses now needs to be aggregated and sent back to the client as one single message, where each <part i> for i=1,2,3 denotes a response as shown above.
We first consider aggregation without a timeout, and then move onto the more advanced scenario, where we aggregate responses received within 1000 ms, and handle premature timeouts correctly with partial - but valid and complete results.
Aggregation without timeouts
First let us consider the aggregation of the responses without a timeout. This expects all three services to respond in-order to send a complete response back to the client. The UltraESB sample configuration # 211 and the Unit test case JsonLoadTest using JUnitPerf load tests this configuration with multiple concurrent users.
Simple in-memory aggregation
When using simple in-memory aggregation, the responses received are collected until all expected responses are received. When the collection completes, a new response message is injected into the outSequence for processing; with each of the responses aggregated as its attachments.
Here is the definition of the proxy service that handles simple in-memory aggregation. Refer to the sample #211 for the complete and annotated configuration.
The merge-and-aggregate-proxy will send a received JSON message to three endpoints (json1, json2 and json3 - defined in the configuration) after cloning and transforming the message into three distinct payloads compatible with each endpoint. The sample uses a simple custom Java transformation SampleJsonTransformation.transformRequest() using the Jackson JSON library to convert a request to the form expected by the backend service.
All responses are received at the outSequence, where the InMemoryMessageAggregator bean is handed the response messages received. The InMemoryMessageAggregator is configured to aggregate 3 messages, and pushes the aggregated message - once complete - again into the outSequence, marked with a special property 'MERGED'
A merged message holds each individual response received as an attachment, and these attachments are then merged into a single payload using another simple Java class JsonMerge.aggregateAsJsonResult() to combine the parts into a single JSON response as { "merged": [ <part 1>, <part 2>, <part 3> ]}. The merged response is sent as one single response back to the original client
Streaming aggregation
When using streaming aggregation, the responses received are sent back to the client as "chunks" of a larger single response. The HTTP protocol streaming support allows the client to receive these chunks transparently as-soon-as-they-become available, without waiting until all three messages are received and aggregated. Using an advanced streaming parser such as the Jackson parser for JSON, a client is able to say update a UI as more data becomes available. To simulate a real-world scenario, the backend JSON services are configured to respond with a random delay of less than 1200ms.
Here is the definition of the proxy service that handles streaming aggregation. Refer to the sample #211 for the complete and annotated configuration.
The stream-and-aggregate-proxy will send a received JSON message to the three endpoints (json1, json2 and json3 - defined in the configuration) after cloning and transforming the message into three distinct payloads compatible with each endpoint. The sample uses the same custom Java transformation SampleJsonTransformation.transformRequest() using the Jackson JSON library to convert a request to the form expected by the backend service.
All responses are received at the outSequence, where the StreamingAggregator bean is handed the response message received. The StreamingAggregator is configured to aggregate 3 messages, and marks each new message as "part <i> of <n>" where "i" is the current message position of the total of "n" messages.
The streaming support of the UltraESB used now can send each of these parts "as soon as they become available" as chunks of a larger aggregated response. Thus part 1 is streamed to the client composed as :
and each subsequent part (i.e. in this example, part 2) is streamed as
and the last part as
to form the complete JSON response to the client. Note that we inject custom Strings between message parts sent, so that they will collectively form a valid larger JSON response to the client reading the response. If we are dealing with an XML payload, a similar gluing-up-of-parts could be performed as required.
Aggregation with timeouts
Next we consider the aggregation of the responses with a timeout - set to one second. Note that the backend services are configured to respond with a random delay less than 1200ms, so that a load test will create aggregated responses with zero, one, two and three part response distribution. The configurations are fixed to respond back after completion, or after the timeout - irrespective of the number of responses actually received from the backend services, and is able to return an empty response as { "merged" : []} if no responses has yet been received.
Simple in-memory aggregation
When using simple in-memory aggregation, the responses received are collected until all expected responses are received or a timeout occurs. When the collection completes or on timeout, a new response message is injected for processing, with each of the responses aggregated as its attachments.
Here is the definition of the proxy service that handles simple in-memory aggregation. Refer to the sample #212 for the complete and annotated configuration.
To facilitate timeouts, the MessageAggregator.beginTimer() method must be invoked from the inSequence for the original incoming message, so that the aggregator can correctly trigger completion after the expected duration from that moment.
On the outSequence, each response received is handed to the message aggregator. Once the timeout expires, the InMemoryMessageAggregator fires with whatever responses collected thus far, and the JsonMerge.aggregateAsJsonResult() method aggregates these messages and sends back to the client
Streaming aggregation
When using streaming aggregation with timeouts, the responses received are sent back to the client as "chunks" of one single response to the client, while considering the timeout, to close a premature response as a valid and correct JSON response expected by the client.
Here is the definition of the proxy service that handles streaming aggregation with a timeout. Refer to the sample #212 for the complete and annotated configuration.
To facilitate timeouts, the MessageAggregator.beginTimer() must be invoked from the inSequence for the original incoming message, so that the aggregator can correctly trigger completion after the expected duration from that moment.
On the outSequence, each response received is handed to the message aggregator, that stamps it as 'EXPIRED' if the aggregation group to which the response belongs has expired. Such messages are dropped.
When the StreamingAggregator is configured with a timeout, it triggers an event after expiration of this time from the beginTimer() call for each aggregation group. Thus, a premature completion will trigger the StreamingAggregator to inject a trigger message marked as "part <i> of <n>", with n set to -1 for such a timeout, signaling immediate closure of the response, and indicating current position expected.
Thus a premature close at part1 will send an empty response as:
While a premature close at part 2, or 3 will send a response as:
to effectively close the premature response as a valid JSON message when one or two parts has been already sent as
Testing the samples in 5 minutes
The configuration defines three external endpoints (json1, json2, json3) with which the proxy services interact €" and in this simple example, we've made them all point at the same sample JSON service hosted in the test environment at: http://localhost:9000/service/JsonServlet
To try the sample service, you could use the AdroitLogic ToolBox, or the UltraESB unit test cases JsonLoadTest OR JsonTimeoutLoadTest which could be executed through your IDE.
Testing aggregation without timeouts
To test the sample 211 without timeouts, start the UltraESB as "./ultraesb.sh -sample 211" from the bin directory of your installation. Next start the ToolBox as "./toolbox.sh" from the bin directory of your installation. [Note: You may use the .bat variants on a Windows environment]
From the ToolBox GUI, select File->New->JettyServer and press "Start Jetty" to start the sample JSON services on port 9000. Now select File->New->Http/s Client and specify the URL http://localhost:9000/service/JsonServlet, the content type as "application/json" and the payload as follows to test the backend service.
You should see a response as shown below
Now send the following request payload to the proxy services as described below.
Lets first send the request to the merge-and-aggregate-proxy by selecting the URL as http://localhost:8280/service/merge-and-aggregate-proxy .You will receive a response as follows:
Note that the response received aggregates responses from three endpoints. Next send a request to the stream-and-aggregate-proxy by selecting the URL as http://localhost:8280/service/stream-and-aggregate-proxy. You will see a similar response with an aggregated message.
To test the proxies with timeouts, start sample 212 as "./ultraesb.sh -sample 212" from the bin directory of your installation, and use the TookBox.
Similarly send the same request to the two proxy services at the same proxy service URLs. You will sometimes receive a response with three parts, and sometimes with two, one or none as shown below - depending on how many responses from the backend were collected before the timeout in 1000ms.
Using TCPDump to see the streaming in action
To verify the streaming of the response with the stream-and-aggregate-proxy, on a Unix or Linux system perform a tcpdump over port 8280. [Note: On a Windows system Wireshark or similar tools will allow such a dump]. The following is an ASCII tcpdump trace revealing the streaming in action
Note that the different parts were returned via separate TCP messages as they became available. In contrast a trace with the merge-and-aggregate-proxy will reveal the complete response being returned as one TCP layer response as follows.