Creating a custom protocol transport

This topic describes how to write a custom protocol implementation which can be used with the Connection API.

Extending connectivity

The connection infrastructure includes a variety of protocol implementations but there may be scenarios in which additional forms of connectivity are required. There are two alternative approaches that may be used to extend connectivity:

Write a simple transmit handler for use with the Event protocol.
Implement and registering a new protocol implementation using the Transport SPI.

The first approach is designed to simplify the custom implementation of an individual connection destination but the Transport SPI should be used for implementing reusable protocol implementations. The advantages of writing a new Transport implementation include:

Implementing a new protocol which can be named and configured under /config/connection/destinations.
Automatic selection of the Transport when the name of the protocol is a valid URL scheme name.
Support for non-blocking protocol implementations in which the request and response phases are carried out at different times.
The ability to replace the default implementation of one of the supplied protocols, such as HTTP or SMTP.

This article outlines how to implement a simple Transport for the ftp protocol and register it with the connection infrastructure so it may be invoked using the Connection API. See the Javadoc SPI reference for more information.

The Transport interface

Custom protocol implementations must implement the zero.core.connection.transports.Transport interface. This interface has three methods which must be implemented.

void connect()
void respond()
void discard()

In addition, a Transport will normally need to provide a public no-argument constructor in order to allow the connection infrastructure to create a new instance for each request.

The logic required in each method will depend on the protocol you are implementing and the way you choose to implement it. There are three basic approaches to writing a Transport:

Simple blocking transport: A simple blocking transport carries out all the activities necessary to send the request and open the response in its connect() method. The implementation of respond() in a simple blocking transport will normally be an empty method.
Non-blocking transport: A non-blocking transport separates the activities of sending the request and opening the response. The connect() method opens connection and completes the activities required to send the request but does not wait for the response. The respond() method waits for the response and opens it for the application.
OutputStream transport: An OutputStream transport is an advanced form of a non-blocking transport that can also provide the application with an OutputStream into which the request body can be written without local buffering.

The discard() method can be used in advanced scenarios to allow a Transport to release resources it may have allocated. Most Transport implementations will have an empty discard() method.

The connection context

The connect() and respond() methods interact directly with the /connection zone of the GlobalContext. Under normal operation, connect() will read the contents of the /connection/request/ keys to send the request and either connect() or respond() will populate the /connection/response keys with the response.

Details of the outbound request can be found using the following keys:

/connection/request/target: Destination resource name.
/connection/request/operation: Operation name.
/connection/request/headers/ header_name: List of values for request header.
/connection/request/protocol/_name: Name of the protocol requested by the application.
/connection/request/protocol/ parameter_name: Protocol configuration value.
/connection/request/body: Request body.

The object type of the request body will depend on the body supplied by the caller of the Connection API. Note that if the caller requested a Connection.getRequestBodyOutputStream() and the Transport does not support OutputStream processing, then /connection/request/body will reference an InputStream from which the request body can be read.

Details of the response can be passed back to the application using the following keys:

/connection/response/headers/ header_name: List of values for response header.
/connection/response/status: Response status String.
/connection/response/body: Response body.

The response body may be any object type, including InputStream and Reader. It is good practice to include a Content-Type response header with a charset value where known, to allow the Connection API to convert between character and binary data as the application requests.

Implementations of the discard() method must not reference the GlobalContext.

Reading protocol configuration

Protocol configuration can be applied to the connection request using various mechanism, as described in Configuring protocols. By the time the Transport implementation is invoked, the protocol configuration will have been made available under keys beginning with /connection/request/protocol/.

For example, consider the following destination configuration:

/config/connection/destinations += {
    "testDestination" : {
        "connection" : {
            "protocol" : "testProtocol",
            "config" : {
                "prop1" : "value1",
                "prop2" : "value2",
                "map1" : {
                    "prop3" : "value3",
                    "prop4" : "value4"
                }
            }
        }
    }
}

This configuration associates the protocol called "testProtocol" with the destination, "testDestination", and supplies some protocol configuration. When the Transport is invoked, it will be able to find this configuration in the GlobalContext as follows:

/connection/request/protocol/_name = testProtocol
/connection/request/protocol/prop1 = value1
/connection/request/protocol/prop2 = value2
/connection/request/protocol/map1/prop3 = value3
/connection/request/protocol/map1/prop4 = value4

For more information about destination configuration, see Configuring destinations.

Exception handling

Either connect() or respond() can signal an exception situation by placing an instance of Exception in /connection/response/body. The Exception contained in /connection/response/body can then be examined or replaced any configured connection handlers.

Implementations of connect() and respond() should be designed to use this mechanism to signal exceptions rather than throwing a RuntimeException. However, if an unexpected RuntimeException is thrown during connect() or respond() processing, it will be caught by the connection infrastructure and placed in /connection/response/body on behalf of the Transport implementation.

If /connection/response/body contains an Exception after the connection handlers have completed their processing, it will eventually be thrown to the application when it uses the Connection API. The Exception will be wrapped in an ConnectionException if it is not an instance of ConnectionException or IOException.

Discarding resources

Implementations of the Transport interface must not assume that the respond() method will always be called. For example, the respond() method will not be called if the connect() method signals an exception condition or if the application does not attempt to read the response.

In some cases, it may be necessary for a Transport implementation to allocate a resource during connect() processing and maintain a instance variable reference to it for use in respond(). The implementation should release such resources before the completion of respond() or before the completion of connect() if connect() sets an Exception.

Even though it cannot be guaranteed that the application will examine the response and so cause respond() to be called, a Transport implementation can ensure any resources are released through the implementation of the discard() method.

The discard() method will called after application request processing has completed, regardless of whether or not respond() was previously called. The discard() must not access the GlobalContext.

Example transport implementations

Writing a simple blocking transport

To demonstrate how to write a simple blocking transport, consider a protocol that might be used to access a simple in-memory stack. In this section we write a Transport that implements two operations, GET and PUT, and that interacts with the request and response body elements.

The example StackTransport class implements the Transport interface and uses a static variable Stack myStack to implement the in-memory stack. The first method of the Transport interface that we must implement is connect():

public void connect() {
    try {
        switch (Connection.toOperation(GlobalContext.zget(GlobalContextURIs.Connection.Request.operation))) {
            case GET:
                GlobalContext.zput(GlobalContextURIs.Connection.Response.body, myStack.pop());
                break;
            case PUT:
                myStack.push(GlobalContext.zget(GlobalContextURIs.Connection.Request.body));
                break;
            default:
                throw new Exception("Unsupported operation for StackTransport");
        }
    } catch (Exception e) {
        GlobalContext.zput(GlobalContextURIs.Connection.Response.body, e);
    }
}

The following is a summary of what this connect() method is doing:

The /connection/request/operation key is read from the GlobalContext and validated. The Connection.toOperation() method is used for convenience; to convert the operation name to a valid Connection.Operation value.
The appropriate action for the current operation is performed:
- For GET, a value is popped from myStack and placed in /connection/response/body.
- For PUT, the value of /connection/request/body is pushed onto myStack.
In any exceptional situation arises, an Exception is placed in /connection/response/body.

Being a simple blocking transport, StackTransport carries out all of its request and response processing in the connect() method, so the respond() method can be an empty implementation.

public void respond() {
    // Nothing to do as all processing is carried out in connect()
}

Here is the remainder of the StackTranport implementation:

package examples.transport;

import java.util.*;
import zero.core.context.GlobalContext;
import zero.core.context.GlobalContextURIs;
import zero.core.connection.Connection;
import zero.core.connection.transports.Transport;

public class StackTransport implements Transport {
        
    private static Stack<Object> myStack = new Stack<Object>();
        
    // connect() and respond() implementations go here
        
    public void discard() {
        // Nothing to do
    }
}

This completes the implementation of our simplistic StackTransport example but before it can be used with the Connection API, it must be registered as a protocol transport with the connection infrastructure.

Writing a non-blocking transport

When writing a non-blocking transport, we must separate the operation of the protocol into connection and response phases and implement the logic for each in the connect() and respond() method.

To demonstrate these concerns we will consider a simple implementation of the ftp protocol, using the ftp support provided in java.net.URLConnection. The implementation of SimpleFtpTransport needs to:

Open the URLConnection in connect() and if the operation is PUT send the request body using it.
If the operation is GET, use the open URLConnection to obtain the response InputStream in respond().

Here is the implementation of connect():

public void connect() {
    try {
        URL url = new URL((String) GlobalContext.zget(GlobalContextURIs.Connection.Request.target));
        urlConnection = url.openConnection();
        operation = Connection.toOperation(GlobalContext.zget(GlobalContextURIs.Connection.Request.operation));
        switch (operation) {
            case GET:
                // nothing to do here>
                break;
            case PUT:
                OutputStream outputStream = urlConnection.getOutputStream();
                try {
                    Object body = GlobalContext.zget(
                        GlobalContextURIs.Connection.Request.body, null);
                    DataUtilities.writeToOutputStream(outputStream, body, null);
                } finally {
                    outputStream.close();
                }
                break;
            default:
                throw new Exception("Unsupported operation for SimpleFtpTransport");
        }
    } catch (Exception e) {
        GlobalContext.zput(GlobalContextURIs.Connection.Response.body, e);
    }
}

There is a summary of what this connect() method is doing:

The /connection/request/target key is read from the GlobalContext and a URLConnection is opened.
The operation is read from /connection/request/operation and if it is PUT then the data specified in /connection/request/body is written to the OutputStream obtained from the URLConnection (using a utility method in zero.core.connection.utils.DataUtilities).
In any exceptional situations arise, an Exception is placed in /connection/response/body. If this happens, then the connection infrastructure will not call the corresponding respond() method.

Here is the corresponding respond() method:

public void respond() {
    try {
        switch (operation) {
            case GET:
                GlobalContext.zput(GlobalContextURIs.Connection.Response.body,
                    urlConnection.getInputStream());
                break;
            case PUT:
                GlobalContext.zput(GlobalContextURIs.Connection.Response.body, null);
        }
    } catch (Exception e) {
        GlobalContext.zput(GlobalContextURIs.Connection.Response.body, e);
    }
}

For the GET operation, the respond() method opens the URLConnection input stream and puts it in /connection/response/body.

Here is the remainder of the SimpleFtpTransport implementation:

package examples.transport;

import java.net.*;
import java.io.*;
import zero.core.context.GlobalContext;
import zero.core.context.GlobalContextURIs;
import zero.core.connection.Connection;
import zero.core.connection.transports.Transport;
import zero.core.connection.utils.DataUtilities;

public class SimpleFtpTransport implements Transport {
        
    Connection.Operation operation;
    URLConnection urlConnection;
        
    // connect() and respond() implementations go here
        
    public void discard() {
        // Nothing to do
    }
        
}

Once again, the new Transport implementation must be registered as a protocol transport before it can be used with the Connection API.

It should be remembered that the SimpleFtpTransport example is intended to illustrate the principles of writing a non-blocking Transport implementation and may not be suitable for deployment in this form. In particular, it is worth noting the following limitations:

The simple implementation assumes that /connection/request/target will be a String in the form of a valid ftp: URL. It does not include any logic to handle cases when this is not true.
There is no consideration for conversion between binary and character data. PUT assumes the system encoding when it converts from character data (using DataUtilities.writeToOutputStream()) and GET does not annotate the response with any encoding information for the Connection API. One solution to this problem would be to make use of a Content-Type header for the PUT and GET operations in a similar manner to the File protocol.
The implementation uses instance variables to pass the operation and urlConnection values between connect() and respond(). This is safe provided the Transport is registered in a way that ensures a new instance is used for each request.

Writing a non-blocking OutputStream transport

A non-blocking protocol implementation can optionally provide the application with an instance of OutputStream in which the request body should be written. If the underlying protocol implementation uses an OutputStream you should consider adding this support to allow an application using Connection.getRequestOutputStream() to obtain access to the stream without buffering.

If a Transport does not provide direct OutputStream for the request body, it does not prevent an application from calling Connection.getRequestBodyOutputStream(). In this case, the connection infrastructure will present the application with an OutputStream buffer and wait for the application to call close(). Once the buffer is ready the Transport will be invoked and the request body supplied to it as an InputStream.

The SimpleFtpTransport example can be modified to provide OutputStream by making some simple changes to the connect() method:

@zero.core.connection.transports.TransportProperties(supportsRequestOutputStream=true)
public void connect() {
    try {
        URL url = new URL((String) GlobalContext.zget(GlobalContextURIs.Connection.Request.target));
        urlConnection = url.openConnection();
        operation = Connection.toOperation(GlobalContext.zget(GlobalContextURIs.Connection.Request.operation));
        switch (operation) {
            case GET:
                // nothing to do here
                break;
            case PUT:
                OutputStream outputStream = urlConnection.getOutputStream();
                Object body = GlobalContext.zget(GlobalContextURIs.Connection.Request.body, null);
                if (body == zero.core.connection.context.ConnectionContextConstants.OUTPUTSTREAM) {
                    GlobalContext.zput(GlobalContextURIs.Connection.Request.outputStream, outputStream);
                } else {
                    try {
                        DataUtilities.writeToOutputStream(outputStream, body, null);
                    } finally {
                        outputStream.close();
                    }
                }
                break;
            default:
                throw new Exception("Unsupported operation for SimpleFtpTransport");
        }
    } catch (Exception e) {
        GlobalContext.zput(GlobalContextURIs.Connection.Response.body, e);
    }
}

There are two small but important changes to note:

The connect() method is now annotated with the annotation, TransportProperties(supportsRequestOutputStream=true).
The value of the /connection/request/body is compared with ConnectionContextConstants.OUTPUTSTREAM using the equality (==) operator and,
- if equal, the OutputStream is placed in /connection/request/outputStream and connect() completes leaving the stream open.
- if not equal, the original non-blocking behavior is used; the request body is written to the OutputStream, which is then closed.

The TransportProperties(supportsRequestOutputStream=true) annotation is used to declare that the Transport is able to provide an OutputStream for the request body. If this annotation is not included, the /connection/request/body value will never be ConnectionContextConstants.OUTPUTSTREAM.

The connect() method is normally responsible for completing the work to send the request to the target. However, when connect() provides an OutputStream, this responsibility is deferred to the close() method of the OutputStream.

Registering a protocol Transport implementation

Once the protocol to be used the request has been finalized (see Configuring protocols), the connection infrastructure fires a getTransport event in order to discover a suitable Transport implementation. By providing a handler for the this event, the developer can determine the Transport implementation that should be used for requests that match a set of criteria.

When the getTransport event is fired, the GlobalContext contains the following information that may be used as conditions when registering the event hander.

/event/target
/event/operation
/event/protocol

Using GenericGetTransportHandler

The simplest way to provide a getTransport event handler is to use the supplied GenericGetTransportHandler implementation. This allows the Transport class to be named directly in zero.config as part of the handler configuration. The name of the selected implementation class is specified as a transport value in the event instanceData.

For example, to register SimpleFtpTransport for use with all requests for the ftp protocol, you can use the following configuration:

/config/handlers += [{
    "events" : "getTransport",
    "handler" : "zero.core.connection.transports.GenericGetTransportHandler.class",
    "conditions" : "/event/protocol =~ ftp",
    "instanceData" : {"transport" : "examples.transport.SimpleFtpTransport"}
}]

Using GenericGetTransportHandler implies that a new instance of the Transport will be automatically created for each connect() invocation, using the public no-argument constructor.

Writing a custom handler for getTransport

If the GenericGetTransportHandler does not offer sufficient flexibility you can write your getTransport handler. For example, you may want to further validate the /connection/request/ context before selecting the transport or select a specific Transport instance rather than allowing the connection infrastructure to create one using a public no-argument constructor.

If an onGetTransport() implementation wishes to determine the transport implementation to be used, it must set the /event/transport key. The value set must be either:

a Class which implements the Transport interface and which has a public, no-argument constructor, or
an instance of an object which implements the Transport interface.

In the former case, the connection infrastructure will create a new instance of the Class to handle the request. In the latter case, the Transport instance set in /event/transport will be used.

If multiple getTransport handlers are eligible for a given request, the first to set /event/transport will prevail. If /event/transport is not set by any handlers then the connection infrastructure will look for a default protocol implementation and return a ConnectionException if none is available.

Invoking the custom protocol Transport

Once the Transport implementation has been registered for a protocol, the connection infrastructure will invoke it for Connection requests using the protocol. For example, since ftp is a standard URL prefix, our SimpleFtpTransport can be invoked directly from the Connection API, with a resource target similar to ftp://userid:password@myhost.com/filename.txt;type=a.

For more information about how protocols are configured and selected, see Configuring protocols.