A New Approach for the Remoting API and Protocol Implementations
I'm going to describe here a new approach for JBoss Remoting, based on information contained in several wiki pages (specifically, building on prior articles by Ovidiu Feodorov, Tom Elrod, and Ron Sigal), forum threads, and JIRA bugs that I've been evaluating.
This approach revolves around several fundamental changes to the basic design of Remoting. I will list them here in no particular order. This list is not exhaustive but it should cover all the major points.
JCA Integration
Remoting 3 should be fully integrated with JCA when used within a container. A client Session should be viewable as a JCA resource that can support local and distributed transactions. Inbound Remoting Sessions should be able to take advantage of the message and transaction inflow capabilities of JCA to provide the ability to port transactions from a remote client.
At the same time, Remoting 3 should be usable on a standalone basis. It should be possible (easy, even) for a user to configure communications between two containers, two standalone applications, or a container and a standalone application, including transaction support.
Endpoint
An Endpoint is the basis for communications. It can represent a server that listens for new clients, and automatically constructs Sessions for them, or it can be a client itself that generate a Session to interact with a configured server. In addition, an Endpoint has a shutdown method that is used to end all Sessions associated with it.
An Endpoint also contains a ConcurrentMap that holds information about the Endpoint, and may also be used by applications.
Endpoint API:
public interface Endpoint {
ConcurrentMap<Object, Object> getEndpointMap();
void shutdown();
boolean isShutdown();
Session openSession() throws UnsupportedOperationException;
Session openSession(ClassLoader classLoader) throws UnsupportedOperationException;
}
Session
A Session instance represents an association with one or more other Session instances that may exist in the same or different virtual machines. There may be one or more network connections to implement the association. Alternatively, the association may be implemented by way of UDP sockets, or a Pipe, an SSL session (over one or more network connections) or even a serial port for that matter. The implementation details are abstracted behind the Session interface.
A Session would also contain a ConcurrentMap that would contain information about the Session, and could also be used by applications to attach state information. This map would have a key type of Object, so that predefined session information (such as local and remote endpoint IP address and TCP/UDP port, or authentication data) can be defined at well-known keys without risk of namespace collision (by using enums, for example). Specifically, it should be guaranteed to the user that Remoting will never itself use Strings as keys for metainformation.
The Session can be used to create new Requests and Messages, and can be used to send Messages to the remote endpoint. The Session may also be used to create new Contexts for request-based communication (see below).
Sessions may have an associated ClassLoader which is used to load classes from the remote endpoint.
Session API:
public interface Session extends Closeable {
void close() throws IOException;
boolean isOpen();
void send(Message message) throws IOException;
Message createMessage(Object identifier, Object body);
Request createRequest(Object identifier, Object body);
ConcurrentMap<Object, Object> getSessionMap();
ListenerMapper getListenerMapper();
Endpoint getEndpoint();
ClassLoader getClassLoader();
Context createContext() throws IOException;
}
Context
A Context is used to facilitate request-reply communications with a remote endpoint. The Context also carries a transaction context. Since it is possible to have multiple Contexts per Session, many transactions may be concurrently executing over one Session.
Using a Context, it is possible to send a request without blocking, receiving an asynchronous notification when it completes. In this way it is possible to trigger multiple concurrent executions within the same Context; however some endpoints (particularly endpoints supporting JCA) will not be able to execute tasks concurrently. Such endpoints may elect to execute the requests serially instead. NOTE: It may be better to simply forbid concurrent execution in a Context. To be determined. NOTE: It is yet to be determined what the semantics of transaction completion are in the presence of outstanding asynchronous requests.
Asynchronous invocations are facilitated using the FutureReply interface, which can be used to monitor, cancel, or get the result of an ongoing operation. In addition, it provides a facility to asynchronously notify the caller when an operation completes. The FutureReply extends the standard java.util.concurrent.Future interface, with the addition of one method that can be used to register a completion handler.
FutureReply API:
public interface FutureReply extends Future<Reply> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
Reply get() throws InterruptedException, CancellationException, ExecutionException;
Reply get(long timeout, TimeUnit unit) throws InterruptedException, CancellationException, ExecutionException, TimeoutException;
FutureReply setCompletionNotifier(RequestCompletionHandler handler);
}
The transaction state of a Context is controlled in one of two ways: either using an XAResource or directly by way of the begin()/commit()/rollback() methods. Some implementations may support nested transactions; however, XAResource may not be used to control such transactions.
It is possible to acquire a local Context that is used to invoke operations on the local side of a Session. This may be used on the listening side of a Session to create composite operations.
Like Session and Endpoint, Context carries a ConcurrentMap that stores metainformation about the Context instance, and can also be used by applications to store state.
Context API:
public interface Context extends Closeable {
void close() throws IOException;
boolean isOpen();
Session getSession();
Reply invoke(Request request) throws IOException, ExecutionException, InterruptedException;
FutureReply send(Request request) throws IOException;
Context getLocalContext() throws IOException;
ConcurrentMap<Object, Object> getContextMap();
void begin() throws IOException;
void commit() throws IOException;
void rollback() throws IOException;
boolean isXASupported();
XAResource getXAResource() throws IOException;
}
Exceptions
Remoting would shift to using more standard JDK exceptions. RemotingException would extend IOException.
Messages and Requests
There are two basic types of activity that can be initiated with a Session: sending messages, and sending requests. A "message" is defined here as being a one-way transmission between associated Session objects. A "request" is a two-way transmission involving a Request and a Reply.
A Session is simply an association; it does not imply tranditional "server" or "client" semantics. The same interface is used on both "ends". Therefore a message or request may be initiated from a client or a server.
Transmission of a Message inherently implies no specific delivery guarantee by default. Varying levels of guaranteed delivery could also be provided.
Since a Message carries no transaction context, it can be sent directly over a Session. A Request must be sent over a Context.
Both messages and requests have similar structure. They consist of an Object identifier (used to identify the operation that is to be invoked on the remote side), and an Object message body. The message body may be any object. Any part of the message body that is a type that is identified as a stream (see below) will automatically be handled as a stream by the serialization code.
Like the entities described above, every message and request has a ConcurrentMap associated with it, which may be used to associate metadata with a single message or request.
Nested Requests
Removed - Streams provide all the functionality of nested requests, and are quite a bit simpler as well.
Streams
A stream is defined as a mechanism by which data can be moved from one endpoint to another a little at a time. Several basic stream types are supported, including InputStream and OutputStream, Readers and Writers, perhaps some XML types like XMLInputStream and XMLOutputStream, and, more critically, object streams, using Iterator, or one of two new interfaces:
public interface ObjectSource<T> extends Closeable {
boolean hasNext();
T next() throws IOException;
void close() throws IOException;
}
public interface ObjectSink<T> extends Closable {
void accept(T object) throws IOException;
void close() throws IOException;
}
These interfaces would allow for streaming of arbitrary objects.
Finally a standard stream type should be defined to enable long-running operations to keep the client updated with the current progress of that operation:
public interface ProgressStream {
void update(String operationTitle, int unitsDone, int totalUnits);
}
This stream type can be used, for example, to update a progress meter in a Swing application.
The user may associate zero or more streams with a request. The remote side would be able to use the given streams serially or concurrently. The stream data would be pushed or pulled on demand over the socket.
A pluggable stream serializer should be introduced to support additional stream types.
The possible applications of data streaming are numerous. Here's a few:
A swing application has a list of database results. An asynchronous invocation is used to initiate a database query, and an ObjectSink implementation which updates the list is associated with the request. The results are fed back asynchronously through the ObjectSink, which adds rows to the user display. The operation can be canceled with a Cancel button, which would simply call the cancel() method on the Future. The ObjectSink is closed by the Session when the operation completes or is canceled.
A billing system needs to process thousands or millions of billing records from external sources. An invocation initiates the billing process; an ObjectSource is associated with the request. The billing system pulls records at the maximum pace that it can process them; perhaps even distributing them among many worker threads.
An invoicing printing system needs to read invoice records and merge them into one large document stream that can be sent to a large-scale printing system. The invoicing system can act as a client to a billing system, and send a request providing an ObjectSink that will accept the invoice records. It can then translate the records into PostScript or a similar printing language, and stream it out to the printer. In this way, files gigabytes in size can be handled without the corresponding memory or disk overhead.
Point-to-Point versus Point-to-Multipoint
A request with a reply is the typical usage scenario of Remoting until now. Such a usage typically requires a point-to-point connection with guaranteed delivery.
A one-way request mechanism (which, by the way, may not require any sort of delivery guarantee) is also useful in such a connection; however, it also opens the door to a new type of association - point-to-multipoint association.
A Session could be created that provides (one-way only) delivery to a datagram socket associated with a multicast group or broadcast address, for example. In this way, messages can be delivered to multiple remote nodes at the same time, with potentially very high throughput. Such a Session would not support Contexts, since two-way communication would not be possible in this case.
Listener
I propose a greatly simplified invocation handler mechanism, consisting of two handler interfaces - one for asynchronous messages, and one for requests:
public interface MessageListener {
void handleMessage(Context context, Message message) throws Exception;
}
public interface RequestListener {
void handleRequest(ReplyContext context, ReceivedRequest request, OutboundReply reply) throws Exception;
boolean isInterruptable();
}
Message listeners may or may not make use of the provided Context to get more information about the sender, or to send message back to the originator.
Request listeners are expected to use the provided ReplyContext to send the reply back to the client. This may be done during the execution of the request handler, or it may be done at any later time. If the request handler throws an exception, it will be relayed back to the client immediately. Otherwise, an exception may be associated with the reply, in which case all reply attachments are cleared.
A request listener may declare whether it is interruptable. If so, invoking the FutureReply.cancel() method on the requesting side may cause the local side to interrupt the request listener.
The ReplyContext simply extends Context with an additional method to send a reply:
public interface ReplyContext extends Context {
void send(OutboundReply reply) throws IOException;
}
Remoting 3 Wire Protocols
The Remoting 3 standard wire protocol, JRPP (JBoss Remoting Point-to-Point), would be a TCP-based transport, which supports the standard authentication/encryption methods described below. In addition, the transport would allow for connection sharing (in other words, all sessions between the same two endpoints would be able to share a single connection).
The URI scheme for JRPP is "jrpp".
Also, an HTTP-based protocol should be supported.
Security - Authentication and Encryption
SASL is the standard authentication mechanism that should be used when possible with Remoting 3. In addition, at least one mechanism should be introduced (such as SRP) that allows authentication and can also generate an encryption key as a side-effect, thus allowing a secure connection without key management. In addition, it should be possible for the client side of a connection to authenticate the server.
SSL should also be configurable, but such configuration should be considered optional due to the complexity of certificate and key management.
APIs and SPIs
The Remoting API will be abstracted from the implementation. Part of the API includes an SPI for registering request and message listeners and stream serializers.
For a prototype version of this API, see: http://anonsvn.jboss.org/repos/sandbox/david.lloyd/remoting3/api-proto
Backwards Compatibility
Wire-level backwards compatibility with prior versions could be easily achieved by providing transports to implement each existing protocol.
API-level backwards compatibility could be implemented by providing JARs for each supported version of Remoting that exposes the corresponding behavior, using the Remoting 3 API internally, if it was desired to reduce the number of maintained versions of Remoting.
References
Comments