Introduction

I have long been intrigued with lightweight Java frameworks for configurable “processing pipelines”. A processing pipeline is an object that accepts input messages, processes each message and returns a response message representing the results of the request message. Along the way, processing pipelines use “interceptors” to intercept incoming and outgoing messages, and “transformers” to transform them. As you can see, processing pipelines are general purpose message (or event) processors.

What makes processing pipelines interesting is that they can be joined to each other (so that one pipeline processes the output messages from another), and that different “links” in the chain can be located remotely with respect to each other. For example, an order processing pipeline could have one pipeline that accepts orders, which feeds another that processes payment information. The payment-processing pipeline then feeds either an order-fulfillment pipeline (if the payment is processed successfully) or a decline-notification pipeline that notifies the purchaser of the problem with processing the payment information. The order-fulfillment pipeline finally feeds an order-completion pipeline that processes the payment information, generates an order for a delivery service (to pick up and deliver the order) and sends an email to the user notifying them that the order has been processed and providing them with tracking-information.

Processing pipelines are either “message processing” pipelines or “invocation pipelines”. The former simply delivers messages to some user-provided functionality that processes the message and generates an output message. The latter expects the input message to represent an action to be performed, the arguments for the action and some additional meta-data, locates an appropriate public method on a user-specified POJO, invokes this method with the given arguments, packages the returned value into a “response” message, and, finally, sends this response message on its way. The response message either encapsulates an “exception” (if the method threw an exception during method invocation) or a result.

Pipelines are built using “components”, which are logical “building blocks” consisting of an “input connector”, a “processing unit” and an “output connector”. A component reads messages from its input connectors, processes them through its processing unit, and sends processed messages through its output connector. Incoming messages read by a component are “request messages”, while outgoing, processed messages are “response messages”.

Processing pipelines are built by “chaining” one or more components in such a manner that messages are processed and passed from component to component, finally coming out from the processing pipeline from the output connector of the last component.

As an example, consider a processing pipeline that consists of three components (named A, B and C) that are joined to each other in the sequence A->B->C.

In other words, the output connector of A is connected to the input connector of B, and the output connector of B is connected to the input connector of C. A and C process only one type of message called “increment” that accepts an integer and adds one to it. B processes only one type of message named “addTwo” that accepts an integer and adds two to it.

Now, if a message requesting an “increment” action is written to A’s input connector with an input value of 1, A adds one to the value (resulting in a value of 2), and writes a message requesting the “addTwo” action, containing the value 2, to its output connector. This causes B to read the message, extract its value (which is 2), add two to this value, and write a message requesting an “increment” action, containing the value 4, to its output connector. This causes C to read the message, extract the value (which is now 4), and add one to this value, resulting in a new value of 5. Finally, C writes this value to its output connector, and the user reads the message (which now contains a value of 5) from the processing pipeline’s output connector. Note that A’s output message has its “action” identifier changed from “increment” to “addTwo” in this example. The mechanism which accomplishes this transformation is described later.

Messages

It is apparent that the concept of a “message” is central to components and processing pipelines. A message is modeled using the IMessage interface (for simplicity, package names are omitted):

public interface IMessage extends Serializable {
    /**
     * Returns the payload of this message.
     *
     * @return
     */
    Serializable getPayload();

    /**
     * Returns a {@link Map} containing all the properties associated with this
     * message.
     *
     * @return
     */
    Map<String, Serializable> getProperties();

    /**
     * Sets the given property, identified by its name and value.
     *
     * @param name
     * @param value
     */
    void setProperty(String name, Serializable value);
}

Thus, a message is modeled as a “payload” and (a logical map of) properties. The payload contains a representation of the “action” to be performed and any required parameters. The properties contain additional meta-data required to process the message (such as “quality of service” type information such as security-related information, etc.).

Messages are further modeled as “request” and “response” messages:

public interface IRequestMessage extends IMessage {
    /**
     * Returns a method-id (obtained via
     * {@link MethodUtil#toString(java.lang.reflect.Method)} indicating the
     * method to invoke.
     *
     * @return
     */
    String getMethod();

    /**
     * Returns an array of objects representing the arguments for the method to
     * be invoked (see {@link #getMethod()}.
     *
     * @return
     */
    Object[] getArguments();
}

and

public interface IResponseMessage extends IMessage {
    /**
     * Represents the value returned by the invoked method (see
     * {@link IRequestMessage#getMethod()}).
     *
     * @return
     */
    Object getResult();

    /**
     * Represents the remotely thrown exception (if any) by the invoked method
     * (see {@link IRequestMessage#getMethod()}).
     *
     * @return
     */
    Throwable getRemoteException();
}

where “request” messages represent a request to perform an action (such as would be provided to a component via its input connector) and “response” messages represent the result of performing the action on the request message, such as would be written out by a component via its output connector.

Components

As we have seen above, a component consists of three constituent parts, viz. an input connector, a message processor and an output connector.

Input connectors

Input connectors read request messages from specific “media”, such as TCP sockets, HTTP endpoints, JMS message queues, memory-based message queues, etc.. The exact type of media (or “transport”) that an input connector reads messages from are specified by the component that owns the input connector.

Output connectors

Similarly, output connectors write response messages to specific “media” (or “transports”) such as TCP sockets, HTTP endpoints, JMS or memory-based message queues, files or the console. Again, the exact type of transport is specified by the component which owns the output connector.

Message processing logic

Besides specifying the transports used by its input and output connectors, a component also specifies its message processing logic. Message processing logic consists of one or more business interfaces, a class implementing these interfaces, interceptors that intercept method invocations on instances of the implementation class and transformers that transform messages. Components specify message processing logic using the following items of information:

Business Services (Interfaces and implementations)

Business services consist of a list of Java interfaces (each of which represents a business service) and a Java class implementing these interfaces. In the following, the term POJI is used to describe the interfaces and POJO is used to describe the implementation class, respectively. Methods are invoked on instances of the POJO by setting up request messages with a) an identifier for the method to invoke and b) required arguments.

Interceptors

A list of “interceptors” that “intercept” message just before and just after invoking the requested method. Interceptors are represented by the following interface:

public interface IInterceptor {
    /**
     * Returns the name of this interceptor.
     *
     * @return
     */
    String getName();

    /**
     * Returns a {@link IRequestMessage} by modifying the original
     * {@link IRequestMessage}.
     *
     * <p>
     * The modified {@link IRequestMessage} is applied to the next
     * {@link IInterceptor} in the chain, or dispatched if this is the last
     * {@link IInterceptor} in the chain.
     * </p>
     *
     * @param request
     * @return
     */
    IMessage before(IMessage request);

    /**
     * Returns a {@link IResponseMessage} from the {@link IResponseMessage}
     * originally returned by dispatching a method.
     *
     * <p>
     * The modified {@link IInterceptor} is passed to the
     * {@link IInterceptor#after(IResponseMessage)} message of the next
     * {@link IInterceptor} in the chain, or returned to the caller if this is
     * the last {@link IInterceptor} in the chain.
     * </p>
     *
     * @param response
     * @return
     */
    IMessage after(IMessage response);

    final static List<IInterceptor> EMPTY_INTERCEPTORS_LIST = Collections
            .unmodifiableList(new ArrayList<IInterceptor>());
}

If more than one interceptor is specified, the interceptors are logically “chained” as follows. Assuming that two interceptors, X and Y are specified, the “before” method of X is called, then the “before” method of Y, and then, after the method invocation, the “after” method of Y is called, followed by the “after” method of X.

Transformers

Components can specify input and output transformers, where each transformer is modeled by the following interface:

public interface ITransformer {
    /**
     * Transforms the given <tt>message</tt> and returns the resulting message.
     *
     * @param message
     * @return
     */
    IMessage transform(IMessage message);

    static final List<ITransformer> EMPTY_TRANSFORMERS_LIST = Collections
            .unmodifiableList(new ArrayList<ITransformer>());
}

Input and output transformers are logically chained when more than one is specified: the “transform” method of each is called in sequence.

An example transformer is one that transforms the response message read from one component into a request message for another component.