Remote Streams in RHQ

The agent/server communication layer in RHQ provides rich, bi-directional communication that is highly configurable, performant, and fault tolerant. And as a developer it has another feature of which I am quite fond - I rarely have to think about it. It just works.

Recently I started working on a new feature that involves streaming potentially large files from agent to server. This work has led me to look under the hood of the comm layer to an extent. The comm layer allows for high-level APIs between server and agent. Consider the following example:

public interface ConfigurationServerService {
    ...
    @Asynchronous(guaranteedDelivery = true)
    void persistUpdatedResourceConfiguration(int resourceId, Configuration resourceConfiguration);
}

The agent calls persistUpdateResourceConfiguration when it has detected a resource configuration change that has occurred outside of RHQ. The @Asynchronous annotation tells the communication layer that the remote method call from agent to server can be performed asynchronously. There are no special stubs or proxies that I have to worry about to use this remote API. It is all nicely tucked away in the communication layer.

Several posts could be devoted to discussing RHQ's communication layer but back to my current work of streaming large files. I needed to put in place a remote API on the server so that the agent can upload files. You might consider something like the following as an initial approach:

// Remote API exposed by RHQ server to stream files from agent to server
void uploadFile(byte[] data);

The problem with this approach is that it involves loading the file contents into memory. File sizes could easily exceed several hundred megabytes in size resulting in substantial memory usage that would be impractical. The RHQ agent is finely tuned to keep a low foot print in terms of memory usage as well as CPU utilization. When reading the contents of a large file that is too big to fit into memory, java.io.InputStream is commonly used. With the RHQ communication layer, I am able to expose an API like the following,

// Remote API exposed by RHQ server to stream files from agent to server
void uploadFile(InputStream stream);

With this API, the agent passes an InputStream object to the server. Keep in mind though that none of Java's standard InputStream classes implement Serializable which is a requirement for using objects with a remote invocation framework like RMI or JBoss Remoting. Fortunately for me RHQ provides the RemoteInputStream class which extends java.io.InputStream. The Javadocs from that class state,

This is an input stream that actually pulls down the stream data from a remote server. Note that this extends InputStream so it can be used as any normal stream object; however, all methods are overridden to actually delegate the methods to the remote stream.

When the agent wants to upload a file, it calls uploadFile passing a RemoteInputStream object. The server can then read from the input stream just as it would any other input stream unbeknownst to it that the bytes are being streamed over the wire.

While I find myself impressed with RemoteInputStream, it gets even better. I wanted to read from the stream asynchronously. When the agent calls uploadFile, instead of reading from the stream in the thread handling the request, I fire off a message to a JMS queue to free up the thread to service other agent requests. I am able to pass the RemoteInputStream object in a JMS message and have a Message Driven Bean then read from the stream to upload the file from the agent.

This level of abstraction along with the performance, fault tolerance, and stability characteristics of the agent/server communication layer makes it one of those hidden gems you do not really appreciate until you have to look under the hood so to speak. And rarely if ever do I find myself having to look under the hood because... it just works. Lastly, I should point out that there is a RemoteOutputStream class that compliments the RemoteInputStream class.