com.devexperts.qd.qtp

Class AbstractQTPComposer

java.lang.Object

com.devexperts.qd.qtp.AbstractMessageVisitor

com.devexperts.qd.qtp.AbstractQTPComposer

All Implemented Interfaces:

DataVisitor, RecordSink, MessageVisitor, SubscriptionVisitor

Direct Known Subclasses:

BinaryQTPComposer

public abstract class AbstractQTPComposer
extends AbstractMessageVisitor
implements DataVisitor, SubscriptionVisitor, RecordSink

Base class for classes that compose QTP messages data into a byte stream in some format. This class has some common composing infrastructure for all formats. Descendants of this class implement particular formats. The output for this composer must be configured with setOutput(BufferedOutput) method immediately after construction.

Note, that all methods declared to throw IOException actually never do so if the underlying BufferedOutput is memory-only. They are declared to throw IOException only in order not to process it in every subclass but to process it in the outermost methods only, where they are wrapped in RuntimeQTPException if they happen.

Method naming convention

composeXXX methods are entry-level methods. The main method is compose(MessageProvider). All compose methods update the number of written bytes in stats.

visitXXX and writeXXX methods are helper methods that are invoked during composing. However, visitXXXData/Subscription methods can be invoked directly when stats are not needed. They update only per-record stats when appropriate.

This class is not thread-safe.

It can be reused to compose unrelated messages record descriptions are not being written (see AbstractQTPComposer(DataScheme, boolean) or if resetSession() is called to clean up its state.

See Also:

AbstractQTPParser

Field Summary

Fields

Modifier and Type	Field and Description
protected MessageType	currentMessageType
protected com.devexperts.io.ChunkedOutput	msg This is temporary buffer where message is composed into.
protected ProtocolOption.Set	optSet
protected DataScheme	scheme
protected QDStats	stats
protected boolean	writeEventTimeSequence
protected boolean	writeHeartbeat

Fields inherited from interface com.devexperts.qd.ng.RecordSink

VOID

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractQTPComposer(DataScheme scheme, boolean describeRecords) Constructs composer.

Method Summary

Modifier and Type	Method and Description
protected void	abortMessageAndRethrow(Throwable t)
void	append(RecordCursor cursor) Adds a record to this sink from the specified RecordCursor.
void	beginMessage(MessageType messageType) Begins new message, must be followed by endMessage().
protected void	beginRecord(DataRecord record)
boolean	compose(MessageProvider provider) Composes message from the corresponding message provider to output and updates number of written bytes in stats.
void	composeDescribeProtocol(ProtocolDescriptor descriptor) Composes describe protocol message to output and updates number of written bytes in stats.
void	composeEmptyHeartbeat() Composes empty heartbeat message to output and updates number of written bytes in stats.
void	composeHeartbeatMessage(HeartbeatPayload heartbeatPayload) Composes heartbeat message to output and updates number of written bytes in stats.
void	composeTimeProgressReport(long timeMillis) Composes heartbeat message with progress report to output and updates number of written bytes in stats.
protected void	describeRecord(DataRecord record)
void	endMessage() End current message that was started with beginMessage(MessageType) and copies composed message to output.
protected void	finishComposingMessage(com.devexperts.io.BufferedOutput out) Performs actions necessary to finish composing a message (for example, inserts message size, inserts records descriptions if necessary).
void	flush() Signal that it is safe to flush appended records to disk, e.g.
protected long	getEventTimeSequence(RecordCursor cursor)
protected long	getMessagePayloadSize()
boolean	hasCapacity() Returns true when this message's capacity had not exceeded threshold yet.
protected boolean	inMessage() Returns true if we are within a message and false if not.
void	resetSession() Resets session state for composer with describe records mode.
void	setOptSet(ProtocolOption.Set optSet)
void	setOutput(com.devexperts.io.BufferedOutput output) Changes output for the composed messages.
void	setStats(QDStats stats) Changes stats to gather composer statistics.
void	setWriteEventTimeSequence(boolean writeEventTimeSequence)
void	setWriteHeartbeat(boolean writeHeartbeat)
protected void	undoWriteMessageHeaderStateChange()
boolean	visitData(DataProvider provider, MessageType type) This method consumes available data for data message from the given data provider.
void	visitDescribeProtocol(ProtocolDescriptor descriptor)
void	visitHeartbeat(HeartbeatPayload heartbeatPayload)
void	visitIntField(DataIntField field, int value) Visits next Int-field within current record.
void	visitObjField(DataObjField field, Object value) Visits next Obj-field within current record.
boolean	visitOtherMessage(int messageType, byte[] messageBytes, int offset, int length) This method consumes other message type.
void	visitRecord(DataRecord record, int cipher, String symbol) Deprecated. Use append(RecordCursor)
void	visitRecord(DataRecord record, int cipher, String symbol, long time) Deprecated. Use append(RecordCursor)
boolean	visitSubscription(SubscriptionProvider provider, MessageType type) This method consumes available subscription for subscription message from the given subscription provider.
protected void	writeDescribeProtocolMessage(com.devexperts.io.BufferedOutput out, ProtocolDescriptor descriptor)
protected void	writeEmptyHeartbeatMessage(com.devexperts.io.BufferedOutput out)
protected void	writeEventTimeSequence(long eventTimeSequence)
protected abstract void	writeField(DataField field, RecordCursor cursor)
protected void	writeHeartbeatMessage(com.devexperts.io.BufferedOutput out, HeartbeatPayload heartbeatPayload)
protected abstract void	writeHistorySubscriptionTime(DataRecord record, long time)
protected abstract void	writeIntField(DataIntField field, int value)
protected abstract void	writeMessageHeader(MessageType messageType) Composes header for a message of a specific type.
protected abstract void	writeObjField(DataObjField field, Object value)
protected void	writeOtherMessageBody(byte[] messageBytes, int offset, int length)
protected abstract int	writeRecordHeader(DataRecord record, int cipher, String symbol, int eventFlags)
protected void	writeRecordPayload(RecordCursor cursor, int eventFlags)

Methods inherited from class com.devexperts.qd.qtp.AbstractMessageVisitor

visitHistoryAddSubscription, visitHistoryData, visitHistoryRemoveSubscription, visitStreamAddSubscription, visitStreamData, visitStreamRemoveSubscription, visitTickerAddSubscription, visitTickerData, visitTickerRemoveSubscription

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

scheme

protected final DataScheme scheme

stats

protected QDStats stats

currentMessageType

protected MessageType currentMessageType

writeEventTimeSequence

protected boolean writeEventTimeSequence

writeHeartbeat

protected boolean writeHeartbeat

optSet

protected ProtocolOption.Set optSet

msg

protected final com.devexperts.io.ChunkedOutput msg

This is temporary buffer where message is composed into. It always uses default chunk pool that is aligned in chunk size with QTPConstants.COMPOSER_THRESHOLD.

Constructor Detail

AbstractQTPComposer

protected AbstractQTPComposer(DataScheme scheme,
                              boolean describeRecords)

Constructs composer. You must setOutput before using this composer.

Parameters:

scheme - the data scheme.

describeRecords - if true, then describe messages are composed right before records are used for the first time and this instance keeps its state and shall not be reused for different communication sessions. See resetSession().

Method Detail

setOutput

public void setOutput(com.devexperts.io.BufferedOutput output)

Changes output for the composed messages.

Parameters:

output - output for the composed messages.

setStats

public void setStats(QDStats stats)

Changes stats to gather composer statistics.

Parameters:

stats - stats to gather composer statistics.

setWriteEventTimeSequence

public void setWriteEventTimeSequence(boolean writeEventTimeSequence)

setWriteHeartbeat

public void setWriteHeartbeat(boolean writeHeartbeat)

setOptSet

public void setOptSet(ProtocolOption.Set optSet)

resetSession

public void resetSession()

Resets session state for composer with describe records mode. Resets the state of all described records so far, so that describe messages are started to being sent again, as if this composer was just created.

Throws:

UnsupportedOperationException - when composer was constructed without describe records mode.

See Also:

AbstractQTPComposer(DataScheme, boolean)

compose

public final boolean compose(MessageProvider provider)

Composes message from the corresponding message provider to output and updates number of written bytes in stats.

This is the outer method from which all the other visitXXX and writeXXX should be written in order to properly record written bytes statistics if needed. However, there are also other composeXXX methods that must be invoked directly (not from inside of this compose invocation).

Inside of this method endMessage() and beginMessage(MessageType) can be used to to arbitrarily delimit messages.

Returns:

false if it had composed all available messages and no messages left to compose, true if more message remain to be composed

composeDescribeProtocol

public final void composeDescribeProtocol(ProtocolDescriptor descriptor)

Composes describe protocol message to output and updates number of written bytes in stats.

composeEmptyHeartbeat

public final void composeEmptyHeartbeat()

Composes empty heartbeat message to output and updates number of written bytes in stats.

composeHeartbeatMessage

public final void composeHeartbeatMessage(HeartbeatPayload heartbeatPayload)

Composes heartbeat message to output and updates number of written bytes in stats.

composeTimeProgressReport

public final void composeTimeProgressReport(long timeMillis)

Composes heartbeat message with progress report to output and updates number of written bytes in stats.

hasCapacity

public boolean hasCapacity()

Returns true when this message's capacity had not exceeded threshold yet. This implementation always returns true when invoked not in message, e.g. the overall number of bytes composed is not checked by this implementation (only the message size for data and subscription messages is limited). Each call to beginMessage(MessageType) resets capacity threshold for a new message.

Specified by:

hasCapacity in interface DataVisitor

Specified by:

hasCapacity in interface RecordSink

Specified by:

hasCapacity in interface SubscriptionVisitor

getMessagePayloadSize

protected long getMessagePayloadSize()

visitDescribeProtocol

public void visitDescribeProtocol(ProtocolDescriptor descriptor)

Specified by:

visitDescribeProtocol in interface MessageVisitor

Overrides:

visitDescribeProtocol in class AbstractMessageVisitor

visitHeartbeat

public void visitHeartbeat(HeartbeatPayload heartbeatPayload)

Specified by:

visitHeartbeat in interface MessageVisitor

Overrides:

visitHeartbeat in class AbstractMessageVisitor

visitData

public boolean visitData(DataProvider provider,
                         MessageType type)

This method consumes available data for data message from the given data provider.

Specified by:

visitData in interface MessageVisitor

Specified by:

visitData in class AbstractMessageVisitor

Returns:

false if it had retrieved all available data and no data left, true if more data remains to be consumed.

visitSubscription

public boolean visitSubscription(SubscriptionProvider provider,
                                 MessageType type)

This method consumes available subscription for subscription message from the given subscription provider.

Specified by:

visitSubscription in interface MessageVisitor

Specified by:

visitSubscription in class AbstractMessageVisitor

Returns:

false if it had retrieved all available subscription and nothing left, true if more subscription remains to be consumed.

visitOtherMessage

public boolean visitOtherMessage(int messageType,
                                 byte[] messageBytes,
                                 int offset,
                                 int length)

This method consumes other message type.

Specified by:

visitOtherMessage in interface MessageVisitor

Overrides:

visitOtherMessage in class AbstractMessageVisitor

Parameters:

messageType - integer number representing a type of the message.

messageBytes - array containing message data.

offset - position of the first byte of message data in messageBytes array.

length - number of bytes starting from offset in messageBytes related to this message.

Returns:

true if the whole message was not processed because the visitor is full and false if the message was successfully processed.

append

public void append(RecordCursor cursor)

Description copied from interface: RecordSink

Adds a record to this sink from the specified RecordCursor.

Specified by:

append in interface RecordSink

flush

public void flush()

Signal that it is safe to flush appended records to disk, e.g. no locks are currently held.

This implementation does nothing.

Specified by:

flush in interface RecordSink

visitRecord

public void visitRecord(DataRecord record,
                        int cipher,
                        String symbol)

Deprecated. Use append(RecordCursor)

Description copied from interface: DataVisitor

Visits next record.

Specified by:

visitRecord in interface DataVisitor

Specified by:

visitRecord in interface RecordSink

visitRecord

public void visitRecord(DataRecord record,
                        int cipher,
                        String symbol,
                        long time)

Deprecated. Use append(RecordCursor)

Description copied from interface: SubscriptionVisitor

Visits next record.

Specified by:

visitRecord in interface RecordSink

Specified by:

visitRecord in interface SubscriptionVisitor

visitIntField

public void visitIntField(DataIntField field,
                          int value)

Description copied from interface: DataVisitor

Visits next Int-field within current record.

Specified by:

visitIntField in interface DataVisitor

Specified by:

visitIntField in interface RecordSink

visitObjField

public void visitObjField(DataObjField field,
                          Object value)

Description copied from interface: DataVisitor

Visits next Obj-field within current record.

Specified by:

visitObjField in interface DataVisitor

Specified by:

visitObjField in interface RecordSink

inMessage

protected final boolean inMessage()

Returns true if we are within a message and false if not.

beginMessage

public final void beginMessage(MessageType messageType)

Begins new message, must be followed by endMessage().

endMessage

public final void endMessage()

End current message that was started with beginMessage(MessageType) and copies composed message to output.

This method can be invoked while composing message with visitData(DataProvider, MessageType) or visitSubscription(SubscriptionProvider, MessageType). It can be followed by resetSession(). If invoked while composing, the message should be resumed by beginMessage(MessageType) invocation before returning control to the composing method.

abortMessageAndRethrow

protected void abortMessageAndRethrow(Throwable t)

writeDescribeProtocolMessage

protected void writeDescribeProtocolMessage(com.devexperts.io.BufferedOutput out,
                                            ProtocolDescriptor descriptor)
                                     throws IOException

Throws:

IOException

writeEmptyHeartbeatMessage

protected void writeEmptyHeartbeatMessage(com.devexperts.io.BufferedOutput out)
                                   throws IOException

Throws:

IOException

writeHeartbeatMessage

protected void writeHeartbeatMessage(com.devexperts.io.BufferedOutput out,
                                     HeartbeatPayload heartbeatPayload)
                              throws IOException

Throws:

IOException

writeMessageHeader

protected abstract void writeMessageHeader(MessageType messageType)
                                    throws IOException

Composes header for a message of a specific type.

Parameters:

messageType - the message type.

Throws:

IOException - never. If it has been thrown then it means an internal error.

undoWriteMessageHeaderStateChange

protected void undoWriteMessageHeaderStateChange()

writeRecordHeader

protected abstract int writeRecordHeader(DataRecord record,
                                         int cipher,
                                         String symbol,
                                         int eventFlags)
                                  throws IOException

Throws:

IOException

writeRecordPayload

protected void writeRecordPayload(RecordCursor cursor,
                                  int eventFlags)
                           throws IOException

Throws:

IOException

writeEventTimeSequence

protected void writeEventTimeSequence(long eventTimeSequence)
                               throws IOException

Throws:

IOException

writeHistorySubscriptionTime

protected abstract void writeHistorySubscriptionTime(DataRecord record,
                                                     long time)
                                              throws IOException

Throws:

IOException

writeIntField

protected abstract void writeIntField(DataIntField field,
                                      int value)
                               throws IOException

Throws:

IOException

writeObjField

protected abstract void writeObjField(DataObjField field,
                                      Object value)
                               throws IOException

Throws:

IOException

writeField

protected abstract void writeField(DataField field,
                                   RecordCursor cursor)
                            throws IOException

Throws:

IOException

writeOtherMessageBody

protected void writeOtherMessageBody(byte[] messageBytes,
                                     int offset,
                                     int length)
                              throws IOException

Throws:

IOException

finishComposingMessage

protected void finishComposingMessage(com.devexperts.io.BufferedOutput out)
                               throws IOException

Performs actions necessary to finish composing a message (for example, inserts message size, inserts records descriptions if necessary). Used by BinaryQTPComposer. This implementation just copies message from msg to out.

Throws:

IOException

getEventTimeSequence

protected long getEventTimeSequence(RecordCursor cursor)

describeRecord

protected void describeRecord(DataRecord record)
                       throws IOException

Throws:

IOException

beginRecord

protected final void beginRecord(DataRecord record)