Java transport functions to implement

HTML Javadoc documentation for AbstractEventTransport and related classes is provided as part of the Apama documentation set. The Javadoc is located in the Apama installation’s doc\javadoc directory.

This topic includes the text of the Javadoc documentation for the functions a transport plug-in author needs to implement, in addition to some pointers on the idiomatic way in which such plug-ins are usually written.

The Constructor

/**
* Construct a new instance of AbstractEventTransport. All subclasses MUST
* provide a constructor with the same signature, which will be used by the
* IAF to create an instance of the transport class.
*
* The AbstractEventTransport implementation does nothing, but subclasses
* should make use of the arguments to initialize the transport.
*
* @param name The transport name, as specified in the IAF config file
* @param properties The transport property set specified in the IAF
* configuration file
* @param timestampConfig Timestamp recording/logging settings from the IAF
* configuration file
* @throws TransportException
*/
public AbstractEventTransport(String name,
EventTransportProperty[] properties
TimestampConfig timestampConfig)
throws TransportException

A typical constructor would create a logger using the plug-in name provided (see Logging from plug-ins in Java), make a call to the updateProperties method to deal with the initial property set passed in, and perform any other initialization operations required for the particular transport being developed.

updateProperties
/**
* Update the configuration of the transport. The transport may assume
* that stop(), flushUpstream() and flushDownstream() have all been called
* before this function is invoked. The recommended procedure for
* updating properties is to first compare the new property set with the
* existing stored properties - if there are no changes, no action should
* be taken.
*
* @param properties The new transport property set specified in the IAF
* configuration file
* @param TimestampConfig timestampConfig
* @throws TransportException
*/
abstract public void updateProperties(EventTransportProperty[] properties,
TimestampConfig timestampConfig)
throws TransportException;

The properties array contains an EventTransportProperty object for each plug-in property specified in the IAF configuration file (in order). The getName and getValue methods allow the plug-in to retrieve the name and value of each property as String objects.

addEventDecoder

/**
* Add a named event decoder to the set of decoders known to the
* transport. If the named decoder already exists, it should be
* replaced.
*
* @param name The name of the decoder to be added
* @param decoder The decoder object instance
* @throws TransportException
*/
abstract public void addEventDecoder(String name, EventDecoder decoder)
throws TransportException;

In an adapter in which multiple event codecs could be present, this function would usually be implemented by storing the <name, decoder> pair in a Java map, from which the EventDecoder could later be retrieved using a plug-in property (e.g. “decoderName”) to determine which of the plug-ins in the map should be used.

Alternatively, if this transport plug-in will only ever be used in an adapter with just one codec plug-in, this method can be implemented simply by storing the provided EventDecoder object in an instance field.

removeEventDecoder

/**
* Remove a named event decoder from the set of decoders known to the
* transport. If the named decoder does not exist, the function should do
* nothing.
*
* @param name The decoder to be removed
* @throws TransportException
*/
abstract public void removeEventDecoder(String name)
throws TransportException;

flushUpstream

/**
* Flush any pending transport events onto the transport. The transport may
* assume that the stop() function has been called before this function,
* so in many cases no action will be required to complete the flushing
* operation.
*
* @throws TransportException
*/
abstract public void flushUpstream() throws TransportException;

flushDownstream

/**
* Flush any pending transport events into the decoder. The transport may
* assume that the stop() function has been called before this function,
* so in many cases no action will be required to complete the flushing
* operation.
*
* @throws TransportException
*/
abstract public void flushDownstream() throws TransportException;

start

/**
* Start processing incoming data from the external transport.
*
* @throws TransportException
*/
abstract public void start() throws TransportException;

This is where a plug-in should establish a connection to its external message source/sink, often by starting a new thread to process or poll for new downstream messages. Events should not be sent to the correlator until the start method has been called.

stop

/**
* Stop processing incoming data from the external transport. New events
* arriving may be blocked, queued or simply dropped, but under no
* circumstances should any be sent into the adapter until the start()
* function is called.
*
* @throws TransportException
*/
abstract public void stop() throws TransportException;

Here, a plug-in may close existing connections and interrupt running threads to ensure that no more messages are passed to the event codec. Events should not be sent to the correlator after the stop method has returned. The stop method must wait for any other threads sending events to complete before the stop method returns.

cleanup/**
* Frees any resources allocated by the transport (useful for resources
* external to the JVM that were allocated in the constructor). The IAF
* guarantees to call this method exactly once.
*
* @throws TransportException
*/
abstract public void cleanup() throws TransportException;

This is where any heavy-weight threads or data structures used for interfacing with the external transport that do not get cleaned up when the plug-in is stopped should be destroyed.

getStatus

/**
* Return a TransportStatus class containing up-to-date status information
* for the transport.
*
* @return An immutable TransportStatus class containing status
* information.
*/
abstract public TransportStatus getStatus();

This method provides the statistics and status message displayed by the IAF Watch tool. A typical plug-in will continuously keep track of the number of messages sent upstream and downstream. Then, when getStatus is called, these message counts can simply be packaged up in a new TransportStatus object together with a String describing the current status of the plug-in (maximum length 1024 characters), and returned.

public TransportStatus getStatus()
{ String status = (started) ? "Status: Running" : "Status: Not running";
return new TransportStatus(status, totalReceived, totalSent);
}

getAPIVersion

/**
* Return the transport API version that the transport was built against.
* @return Must be EventTransport.API_VERSION.
*/
public abstract int getAPIVersion();

sendTransportEvent (send upstream)

/**
* Called by an event encoder to send an upstream message to the external
* transport.
*
* It is assumed that the encoder and transport share
* the same definition of the content of the event, so that the transport
* can effectively interpret the event.
*
* @param event An object representing the event to be sent by the
* transport, in a format shared by the encoder and transport.
* @param timestamps A TimestampSet representing the timestamps attached to
*
* @throws TransportException Thrown by the transport if any error occurs
* sending the message.
*/
public abstract void sendTransportEvent(Object event,
TimestampSet timestamps)
throws TransportException;

This is the method that a codec layer plug-in calls when it receives a translated upstream Apama event that needs to be sent on to an event transport for transmission to an external message sink.

Note that there are no guarantees about which threads might be used to call this method, so plug-in authors will need to consider any thread synchronization issues arising from use of shared data structures here.