Work in progress!
Although, there is already a large set of wrappers available for Odysseus, there is often the need to create a new specific adapter for a specific source. To ease the creation of new wrapper, we created an adapter framework.
There are two kinds of generic wrapper:
- GenericPull: In this approach, data is pulled from the source, i.e. the source is asked if it provides more information and then the information is delivered. A typical example is a file or a web source. The Odysseus scheduler initiates the pull request.
- GenericPush: In this approach, the source actively sends the data to Odysseus, so each time new data is available it will directly send through Odysseus. A typical example is a tcp server. No scheduler is needed in this case.
To avoid redundant coding, the wrappers are built from basic elements (handler), each representing a specific function in the wrapping process.
- Transport: Die physical bridge between the external systems and Odysseus is given by this handler. It is responsible for the communication, this could be e.g. a file access, a tcp client/server or a message bus handler.
- Protocol: While the transport handler handles the connections, the protocol handler is responsible for the interpretation of the given input coming from the transport handler. I.e. the protocol handler translates the incoming data into a format understandable by Odysseus
- DataHandler: Finally, the internal format can be defined. This is the data type, which is used by the internal operators of Odysseus. There are different kind of operators for different kinds of datatypes. DataHandler are e.g. Tuple or KeyValue.
In a source, the information is send from the transport handler, via the protocol handler to the datahandler.
The wrapper can also be used at sink side, i.e. to send data from Odysseus to other systems. In this case, the information is send from the datahandler via the protocol handler to the transport handler. Each handler can provide both ways, i.e. retrieving and sending of data but is not required to. So some handler may only be used in sources, while other only in sinks.
This document explains how to write new wrappers using this generic wrappers.
Creating a new Transport Handler
To create a new TransportHandler, the Interface ITransportHandler must be implemented or the class AbstractTransportHandler be extended.
Depending on the way, the handler works, different methods need to be implemented.
Independent of Push/Pull
public ITransportHandler createInstance(IProtocolHandler<?> protocolHandler, Map<String, String> options);
This method must return a new initialized transport handler. Typically, the constructor is called.
@Override
public ITransportHandler createInstance(
IProtocolHandler<?> protocolHandler, Map<String, String> options) {
return new FileHandler(protocolHandler, options);
}
The methode getName() must deliver a global unique transport handler name.
String getName();
Its a good was to use this a follows (again example of FileHandler):
public static final String NAME = "File";
@Override
public String getName() {
return NAME;
}
When implementing ITransportHandler, open and close need to be implemented.
Hint: In the following we will assume, that AbstractTransportHandler will be overwritten.
AbstractTransportHandler provides already default implementations that cannot be overwritten (its implementend in AbstractTransportHandlerDelegate):
final synchronized public void open() throws UnknownHostException,
IOException {
if (openCounter == 0) {
if (getExchangePattern() != null
&& (getExchangePattern().equals(
ITransportExchangePattern.InOnly)
|| getExchangePattern().equals(
ITransportExchangePattern.InOptionalOut) || getExchangePattern()
.equals(ITransportExchangePattern.InOut))) {
callOnMe.processInOpen();
}
if (getExchangePattern() != null
&& (getExchangePattern().equals(
ITransportExchangePattern.OutOnly)
|| getExchangePattern().equals(
ITransportExchangePattern.OutOptionalIn) || getExchangePattern()
.equals(ITransportExchangePattern.InOut))) {
callOnMe.processOutOpen();
}
}
openCounter++;
}
final synchronized public void close() throws IOException {
openCounter--;
if (openCounter == 0) {
if (getExchangePattern() != null
&& (getExchangePattern().equals(
ITransportExchangePattern.InOnly)
|| getExchangePattern().equals(
ITransportExchangePattern.InOptionalOut) || getExchangePattern()
.equals(ITransportExchangePattern.InOut))) {
callOnMe.processInClose();
}
if (getExchangePattern() != null
&& (getExchangePattern().equals(
ITransportExchangePattern.OutOnly)
|| getExchangePattern().equals(
ITransportExchangePattern.OutOptionalIn) || getExchangePattern()
.equals(ITransportExchangePattern.InOut))) {
callOnMe.processOutClose();
}
}
}
A TransportHandler can provide different exchange pattern. The handler must deliver the pattern when calling the following method:
public ITransportExchangePattern getExchangePattern();
Currently, the following values are available (TODO: Extend description of remaining pattern):
- InOnly: The handler can only be used as source.
- RobustInOnly
- InOut: The handler can be used as source and as sink.
- InOptionalOut
- OutOnly: The handler can only be used as sink.
- RobustOutOnly
- OutIn
- OutOptionalIn
AbstractTransportHandler calls according to the exchange pattern the corresponding methods:
- processInOpen()
- processOutOpen()
- processInClose()
- processOutClose()
In this methods the TransportHandler must open or close the connections. The "IN"-methods are called for sources, the "OUT" for sinks. When starting or stopping a query, open and close are called respectively.
In the following again the implemenations for the FileHandler
@Override
public void processInOpen() throws IOException {
if (!preload) {
final File file = new File(filename);
try {
in = new FileInputStream(file);
fireOnConnect();
} catch (Exception e) {
fireOnDisconnect();
throw e;
}
} else {
fis = new FileInputStream(filename);
FileChannel channel = fis.getChannel();
long size = channel.size();
double x = size / (double) Integer.MAX_VALUE;
int n = (int) Math.ceil(x);
ByteBuffer buffers[] = new ByteBuffer[n];
for (int i = 0; i < n; i++) {
buffers[i] = ByteBuffer.allocateDirect(Integer.MAX_VALUE);
channel.read(buffers[i]);
buffers[i].rewind();
}
in = createInputStream(buffers);
fireOnConnect();
}
}
@Override
public void processInClose() throws IOException {
super.processInClose();
if (fis != null) {
fis.close();
}
}
@Override
public void processOutOpen() throws IOException {
final File file = new File(filename);
try {
out = new FileOutputStream(file, append);
fireOnConnect();
} catch (Exception e) {
fireOnDisconnect();
throw e;
}
}
@Override
public void processOutClose() throws IOException {
fireOnDisconnect();
out.flush();
out.close();
}
Generic Pull
After the connection is inialized, the framework tries to retrieve data from the TransportHandler. To be generic, we decided to use an InputStream for sources and an OutputStream for sinks. So the following methods need to be overwritten (Remark: It its not necessary to implement both methods, if the TransportHandler e.g. should only be used for sources):
public InputStream getInputStream(); public OutputStream getOutputStream();
A typical implementation in FileHandler:
@Override
public InputStream getInputStream() {
return in;
}
@Override
public OutputStream getOutputStream() {
return out;
}
In some cases, sources deliver not an endless data stream. For such cases the method
public boolean isDone();
can be overwrittten.
GenericPush
In generic push szenarios for sources there is no method that can be overwritten because it depends on the transport type and e.g. libararies that receive data from external sources. The information that is read must be send to the corresponding transport handler. To simplify the process, AbstractTransportHandler(Delegate) provides the following methods that should be used:
public void fireProcess(ByteBuffer message) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
// TODO: flip() erases the contents of the message if
// it was already flipped or just created...
// In other words: This method expects that the byte buffer
// is not fully prepared
message.flip();
l.process(message);
}
}
public void fireProcess(T m) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.process(m);
}
}
public void fireProcess(String[] message) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.process(message);
}
}
public void fireOnConnect(ITransportHandler handler) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.onConnect(handler);
}
}
public void fireOnDisconnect(ITransportHandler handler) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.onDisonnect(handler);
}
}
The fireProcess methods can be used with ByteBuffers and String-Arrays or with a Generic. In the latter case, the corresponding ProtocolHandler must read this type, else a class cast exception will be thrown. Two additional methods are used to inform listener about connection states (connect and disconnect)
public void fireOnConnect(ITransportHandler handler) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.onConnect(handler);
}
}
public void fireOnDisconnect(ITransportHandler handler) {
for (ITransportHandlerListener<T> l : transportHandlerListener) {
l.onDisonnect(handler);
}
}
OSGi, Registering, Declarative service
Creating a new Protocol Handler