The protocol handler receives data from a transport handler and prepares it for the data handler or receives data from the data handler and prepares it for the transport handler. A simple protocol is a comma separated list (CSV).
To create a new protocol handler AbstractProtocolHandler must be extended (or IProtocolHandler implemented)
/** * Creates a new protocol handler * @param direction is this handler used in a source (IN) or in a sink (OUT) * @param access which kind of access pattern is supported ( PUSH, PULL, ROBUST_PUSH, ROBUST_PULL) * @param options set of options as key value pairs * @param dataHandler the data handler thats connected to the protocol handler * @return */ public IProtocolHandler<T> createInstance(ITransportDirection direction, IAccessPattern access, Map<String, String> options, IDataHandler<T> dataHandler);
The following method could be overwritten:
- open(): This method is called, when the query is started.
Important: When overwriting this method, getTransportHandler().open() must be called, too. - close(): This method is called, when der query is stopped.
Important: When overwriting this method, getTransportHandler().close() must be called, too.
Pullbased
- boolean hasNext(): must be overwritten, to state if a new element is available for processing
- T getNext(): must be overwritten to deliver the next element
- boolean isDone(): can be overwritten to state if a source will not deliver anymore elements
Pushbased
For pushbased access the methods from the Interface ITransportHandlerListener need to be overwritten:
/** * Is called when a new connection with the transport handler is established * @param caller */ void onConnect(ITransportHandler caller); /** * Is called when an existing connection to the transport handler is interrupted * @param caller */ void onDisonnect(ITransportHandler caller); /** * Implement this method to process the message * @param message as ByteBuffer */ void process(ByteBuffer message); /** * Implement this method to process the message * @param message as String Array */ void process(String[] message); /** * Implement this method to process the message * @param message as T */ void process(T m);
Typically, these methods are called from the underlying transport handler