epics::pvAccess::detail::AbstractCodec Class Reference

#include "codec.h"

Inheritance diagram for epics::pvAccess::detail::AbstractCodec:

Collaboration diagram for epics::pvAccess::detail::AbstractCodec:

Public Member Functions
	AbstractCodec (bool serverFlag, size_t sendBufferSize, size_t receiveBufferSize, int32_t socketSendBufferSize, bool blockingProcessQueue)
virtual void	processControlMessage ()=0
virtual void	processApplicationMessage ()=0
virtual const osiSockAddr *	getLastReadBufferSocketAddress ()=0
virtual void	invalidDataStreamHandler ()=0
virtual void	readPollOne ()=0
virtual void	writePollOne ()=0
virtual void	scheduleSend ()=0
virtual void	sendCompleted ()=0
virtual bool	terminated ()=0
virtual int	write (epics::pvData::ByteBuffer *src)=0
virtual int	read (epics::pvData::ByteBuffer *dst)=0
virtual bool	isOpen ()=0
virtual	~AbstractCodec ()
virtual void	alignBuffer (std::size_t alignment) OVERRIDE FINAL
virtual void	ensureData (std::size_t size) OVERRIDE FINAL
virtual void	alignData (std::size_t alignment) OVERRIDE FINAL
virtual void	startMessage (epics::pvData::int8 command, std::size_t ensureCapacity=0, epics::pvData::int32 payloadSize=0) OVERRIDE FINAL
void	putControlMessage (epics::pvData::int8 command, epics::pvData::int32 data)
virtual void	endMessage () OVERRIDE FINAL
virtual void	ensureBuffer (std::size_t size) OVERRIDE FINAL
virtual void	flushSerializeBuffer () OVERRIDE FINAL
virtual void	flush (bool lastMessageCompleted) OVERRIDE FINAL
void	processWrite ()
void	processRead ()
void	processSendQueue ()
virtual void	enqueueSendRequest (TransportSender::shared_pointer const &sender) OVERRIDE FINAL
void	enqueueSendRequest (TransportSender::shared_pointer const &sender, std::size_t requiredBufferSize)
void	setSenderThread ()
virtual void	setRecipient (osiSockAddr const &sendTo) OVERRIDE FINAL
virtual void	setByteOrder (int byteOrder) OVERRIDE FINAL
virtual bool	directSerialize (epics::pvData::ByteBuffer *, const char *, std::size_t, std::size_t) OVERRIDE
virtual bool	directDeserialize (epics::pvData::ByteBuffer *, char *, std::size_t, std::size_t) OVERRIDE
bool	sendQueueEmpty () const
epics::pvData::int8	getRevision () const
Public Member Functions inherited from epics::pvAccess::TransportSendControl
	POINTER_DEFINITIONS (TransportSendControl)
virtual	~TransportSendControl ()
Public Member Functions inherited from epics::pvData::SerializableControl
virtual	~SerializableControl ()
virtual void	cachedSerialize (std::tr1::shared_ptr< const Field > const &field, ByteBuffer *buffer)=0
Public Member Functions inherited from epics::pvAccess::Transport
	POINTER_DEFINITIONS (Transport)
	Transport ()
virtual	~Transport ()
virtual bool	acquire (std::tr1::shared_ptr< ClientChannelImpl > const &client)=0
virtual void	release (pvAccessID clientId)=0
virtual std::string	getType () const =0
virtual const osiSockAddr &	getRemoteAddress () const =0
virtual const std::string &	getRemoteName () const =0
virtual std::size_t	getReceiveBufferSize () const =0
virtual epics::pvData::int16	getPriority () const =0
virtual void	setRemoteTransportReceiveBufferSize (std::size_t receiveBufferSize)=0
virtual void	setRemoteTransportSocketReceiveBufferSize (std::size_t socketReceiveBufferSize)=0
virtual void	flushSendQueue ()=0
virtual void	verified (epics::pvData::Status const &status)=0
virtual bool	verify (epics::pvData::int32 timeoutMs)=0
virtual void	close ()=0
virtual void	waitJoin ()
	Call after close() to wait for any worker threads to exit. More...
virtual bool	isClosed ()=0
virtual void	authNZMessage (epics::pvData::PVStructure::shared_pointer const &data)=0
Public Member Functions inherited from epics::pvData::DeserializableControl
virtual	~DeserializableControl ()
virtual std::tr1::shared_ptr< const Field >	cachedDeserialize (ByteBuffer *buffer)=0

Static Public Member Functions
static std::size_t	alignedValue (std::size_t value, std::size_t alignment)

Public Attributes
epics::pvData::Mutex	_mutex
Public Attributes inherited from epics::pvAccess::Transport
size_t	_totalBytesSent
size_t	_totalBytesRecv

Static Public Attributes
static const std::size_t	MAX_MESSAGE_PROCESS = 100
static const std::size_t	MAX_MESSAGE_SEND = 100
static const std::size_t	MAX_ENSURE_SIZE = 1024
static const std::size_t	MAX_ENSURE_DATA_SIZE = MAX_ENSURE_SIZE/2
static const std::size_t	MAX_ENSURE_BUFFER_SIZE = MAX_ENSURE_SIZE
static const std::size_t	MAX_ENSURE_DATA_BUFFER_SIZE = 1024
Static Public Attributes inherited from epics::pvAccess::Transport
static size_t	num_instances

Protected Member Functions
virtual void	sendBufferFull (int tries)=0
void	send (epics::pvData::ByteBuffer *buffer)
void	flushSendBuffer ()
virtual void	setRxTimeout (bool ena)

Protected Attributes
ReadMode	_readMode
int8_t	_version
int8_t	_flags
int8_t	_command
int32_t	_payloadSize
epics::pvData::int32	_remoteTransportSocketReceiveBufferSize
osiSockAddr	_sendTo
epicsThreadId	_senderThread
WriteMode	_writeMode
bool	_writeOpReady
epics::pvData::ByteBuffer	_socketBuffer
epics::pvData::ByteBuffer	_sendBuffer
fair_queue< TransportSender >	_sendQueue
const epics::pvData::int8	_clientServerFlag

Detailed Description

Definition at line 127 of file codec.h.

Constructor & Destructor Documentation

epics::pvAccess::detail::AbstractCodec::AbstractCodec	(	bool	serverFlag,
		size_t	sendBufferSize,
		size_t	receiveBufferSize,
		int32_t	socketSendBufferSize,
		bool	blockingProcessQueue
	)

Definition at line 90 of file codec.cpp.

                              :
    //PROTECTED
    _readMode(NORMAL), _version(0), _flags(0), _command(0), _payloadSize(0),
    _remoteTransportSocketReceiveBufferSize(MAX_TCP_RECV),
    _senderThread(0),
    _writeMode(PROCESS_SEND_QUEUE),
    _writeOpReady(false),
    _socketBuffer(bufSizeSelect(receiveBufferSize)),
    _sendBuffer(bufSizeSelect(sendBufferSize)),
    //PRIVATE
    _storedPayloadSize(0), _storedPosition(0), _startPosition(0),
    _maxSendPayloadSize(_sendBuffer.getSize() - 2*PVA_MESSAGE_HEADER_SIZE),    // start msg + control
    _lastMessageStartPosition(std::numeric_limits<size_t>::max()),_lastSegmentedMessageType(0),
    _lastSegmentedMessageCommand(0), _nextMessagePayloadOffset(0),
    _byteOrderFlag(EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG ? 0x80 : 0x00),
    _clientServerFlag(serverFlag ? 0x40 : 0x00)
{
    if (_socketBuffer.getSize() < 2*MAX_ENSURE_SIZE)
        throw std::invalid_argument(
            "receiveBuffer.capacity() < 2*MAX_ENSURE_SIZE");

    if (_sendBuffer.getSize() < 2*MAX_ENSURE_SIZE)
        throw std::invalid_argument("sendBuffer() < 2*MAX_ENSURE_SIZE");

    // initialize to be empty
    _socketBuffer.setPosition(_socketBuffer.getLimit());
    _startPosition = _socketBuffer.getPosition();

    // clear send
    _sendBuffer.clear();
}

virtual epics::pvAccess::detail::AbstractCodec::~AbstractCodec ( )

inlinevirtual

Definition at line 161 of file codec.h.

    {
    }

Member Function Documentation

void epics::pvAccess::detail::AbstractCodec::alignBuffer ( std::size_t  alignment )

virtual

Add pad bytes to buffer.

Parameters

alignment

alignment required.

Implements epics::pvData::SerializableControl.

Definition at line 580 of file codec.cpp.

                                                   {

    std::size_t k = (alignment - 1);
    std::size_t pos = _sendBuffer.getPosition();
    std::size_t newpos = (pos + k) & (~k);
    if (pos == newpos)
        return;

    // for safety reasons we really pad (override previous message data)
    std::size_t padCount = newpos - pos;
    _sendBuffer.put(PADDING_BYTES, 0, padCount);
}

void epics::pvAccess::detail::AbstractCodec::alignData ( std::size_t  alignment )

virtual

Align buffer. Note that this takes care only current buffer alignment. If streaming protocol is used, care must be taken that entire stream is aligned.

Parameters

alignment

size in bytes, must be power of two.

Implements epics::pvData::DeserializableControl.

Definition at line 546 of file codec.cpp.

                                                 {

    std::size_t k = (alignment - 1);
    std::size_t pos = _socketBuffer.getPosition();
    std::size_t newpos = (pos + k) & (~k);
    if (pos == newpos)
        return;

    std::size_t diff = _socketBuffer.getLimit() - newpos;
    if (diff > 0)
    {
        _socketBuffer.setPosition(newpos);
        return;
    }

    ensureData(diff);

    // position has changed, recalculate
    newpos = (_socketBuffer.getPosition() + k) & (~k);
    _socketBuffer.setPosition(newpos);
}

std::size_t epics::pvAccess::detail::AbstractCodec::alignedValue ( std::size_t  value, std::size_t  alignment  )

static

Definition at line 537 of file codec.cpp.

                         {

    std::size_t k = (alignment - 1);
    return (value + k) & (~k);
}

bool epics::pvAccess::detail::AbstractCodec::directDeserialize ( epics::pvData::ByteBuffer *  existingBuffer, char *  deserializeTo, std::size_t  elementCount, std::size_t  elementSize  )

virtual

Method for deserializing array data. Hook for supplying custom deserialization implementation. The deserialization implementation need not be provided. Returns true if method performs deserialization, false otherwise. This should only be used for arrays of primitive types. i.e. boolean, byte,..., double. It cannot be called for string, structure, or union arrays.

Parameters

existingBuffer	the existing buffer from the caller.
deserializeTo	location of data.
elementCount	number of elements.
elementSize	element size.

Returns

true if deserialization performed, else false.

Implements epics::pvData::DeserializableControl.

Definition at line 1003 of file codec.cpp.

{
    return false;
}

bool epics::pvAccess::detail::AbstractCodec::directSerialize ( epics::pvData::ByteBuffer *  existingBuffer, const char *  toSerialize, std::size_t  elementCount, std::size_t  elementSize  )

virtual

Method for serializing primitive array data. Hook for supplying custom serialization implementation. The serialization implementation need not be provided. Returns true if method performs serialization, false otherwise. This should only be used for arrays of primitive types, i. e. boolean, byte,..., double. It cannot be called for string, structure, or union arrays.

Parameters

existingBuffer	the existing buffer from the caller.
toSerialize	location of data to be put into buffer.
elementCount	number of elements.
elementSize	element size.

Returns

true if serialization performed, else false.

Implements epics::pvData::SerializableControl.

Definition at line 961 of file codec.cpp.

{
    // TODO overflow check of "size_t count", overflow int32 field of payloadSize header field
    // TODO max message size in connection validation
    std::size_t count = elementCount * elementSize;

    // TODO find smart limit
    // check if direct mode actually pays off
    if (count < 64*1024)
        return false;

    //
    // first end current message, and write a header of next "directly serialized" message
    //

    // first end current message indicating the we will segment
    endMessage(true);

    // append segmented message header with payloadSize == count
    // TODO size_t to int32
    startMessage(_lastSegmentedMessageCommand, 0, static_cast<int32>(count));

    // flush
    flushSendBuffer();

    // TODO think if alignment is preserved after...

    //
    // send toSerialize buffer
    //
    ByteBuffer wrappedBuffer(const_cast<char*>(toSerialize), count);
    send(&wrappedBuffer);

    //
    // continue where we left before calling directSerialize
    //
    startMessage(_lastSegmentedMessageCommand, 0);

    return true;
}

void epics::pvAccess::detail::AbstractCodec::endMessage ( )

virtual

Implements epics::pvAccess::TransportSendControl.

Definition at line 632 of file codec.cpp.

                               {
    endMessage(false);
}

void epics::pvAccess::detail::AbstractCodec::enqueueSendRequest ( TransportSender::shared_pointer const &  sender )

virtual

Enqueue send request.

Parameters

sender

Implements epics::pvAccess::Transport.

Definition at line 873 of file codec.cpp.

                                                  {
    _sendQueue.push_back(sender);
    scheduleSend();
}

void epics::pvAccess::detail::AbstractCodec::enqueueSendRequest	(	TransportSender::shared_pointer const &	sender,
		std::size_t	requiredBufferSize
	)

Definition at line 928 of file codec.cpp.

                                  {

    if (_senderThread == epicsThreadGetIdSelf() &&
            _sendQueue.empty() &&
            _sendBuffer.getRemaining() >= requiredBufferSize)
    {
        processSender(sender);
        if (_sendBuffer.getPosition() > 0)
        {
            scheduleSend();
        }
    }
    else
        enqueueSendRequest(sender);
}

void epics::pvAccess::detail::AbstractCodec::ensureBuffer ( std::size_t  size )

virtual

Make sure buffer has at least size bytes remaining. If not flush existing buffer and provide a new one.

Parameters

size	The number of bytes.

Implements epics::pvData::SerializableControl.

Definition at line 701 of file codec.cpp.

                                               {

    if (_sendBuffer.getRemaining() >= size)
        return;

    // too large for buffer...
    if (_maxSendPayloadSize < size) {
        std::ostringstream msg;
        msg << "requested for buffer size " <<
            size << ", but only " << _maxSendPayloadSize << " available.";
        std::string s = msg.str();
        LOG(logLevelWarn,
            "%s at %s:%d.,", msg.str().c_str(), __FILE__, __LINE__);
        throw std::invalid_argument(s);
    }

    while (_sendBuffer.getRemaining() < size)
        flush(false);
}

void epics::pvAccess::detail::AbstractCodec::ensureData ( std::size_t  size )

virtual

Helper method. Ensures specified size of bytes, provides it if necessary.

Parameters

size	The number of bytes.

Implements epics::pvData::DeserializableControl.

Definition at line 423 of file codec.cpp.

                                             {

    // enough of data?
    if (_socketBuffer.getRemaining() >= size)
        return;

    // to large for buffer...
    if (size > MAX_ENSURE_DATA_SIZE)    {// half for SPLIT, half for SEGMENTED
        std::ostringstream msg;
        msg << "requested for buffer size " << size
            << ", but maximum " << MAX_ENSURE_DATA_SIZE << " is allowed.";
        LOG(logLevelWarn,
            "%s at %s:%d.,", msg.str().c_str(), __FILE__, __LINE__);
        throw std::invalid_argument(msg.str());
    }

    try
    {

        // subtract what was already processed
        std::size_t pos = _socketBuffer.getPosition();
        _storedPayloadSize -= pos - _storedPosition;

        // SPLIT message case
        // no more data and we have some payload left => read buffer
        // NOTE: (storedPayloadSize >= size) does not work if size
        //spans over multiple messages
        if (_storedPayloadSize >= (_storedLimit-pos))
        {
            // just read up remaining payload
            // this will move current (<size) part of the buffer
            // to the beginning of the buffer
            ReadMode storedMode = _readMode;
            _readMode = SPLIT;
            readToBuffer(size, true);
            _readMode = storedMode;
            _storedPosition = _socketBuffer.getPosition();
            _storedLimit = _socketBuffer.getLimit();
            _socketBuffer.setLimit(
                std::min<std::size_t>(
                    _storedPosition + _storedPayloadSize, _storedLimit));

            // check needed, if not enough data is available or
            // we run into segmented message
            ensureData(size);
        }
        // SEGMENTED message case
        else
        {
            // TODO check flags
            //if (flags && SEGMENTED_FLAGS_MASK == 0)
            //  throw IllegalStateException("segmented message expected,
            //but current message flag does not indicate it");


            // copy remaining bytes of payload to safe area
            //[0 to MAX_ENSURE_DATA_BUFFER_SIZE/2), if any
            // remaining is relative to payload since buffer is
            //bounded from outside
            std::size_t remainingBytes = _socketBuffer.getRemaining();
            for (std::size_t i = 0; i < remainingBytes; i++)
                _socketBuffer.putByte(i, _socketBuffer.getByte());

            // restore limit (there might be some data already present
            //and readToBuffer needs to know real limit)
            _socketBuffer.setLimit(_storedLimit);

            // we expect segmented message, we expect header
            // that (and maybe some control packets) needs to be "removed"
            // so that we get combined payload
            ReadMode storedMode = _readMode;
            _readMode = SEGMENTED;
            processRead();
            _readMode = storedMode;

            // make sure we have all the data (maybe we run into SPLIT)
            readToBuffer(size - remainingBytes, true);

            // SPLIT cannot mess with this, since start of the message,
            //i.e. current position, is always aligned
            _socketBuffer.setPosition(
                _socketBuffer.getPosition());

            // copy before position (i.e. start of the payload)
            for (int32_t i = remainingBytes - 1,
                    j = _socketBuffer.getPosition() - 1; i >= 0; i--, j--)
                _socketBuffer.putByte(j, _socketBuffer.getByte(i));

            _startPosition = _socketBuffer.getPosition() - remainingBytes;
            _socketBuffer.setPosition(_startPosition);

            _storedPayloadSize += remainingBytes;
            _storedPosition = _startPosition;
            _storedLimit = _socketBuffer.getLimit();
            _socketBuffer.setLimit(
                std::min<std::size_t>(
                    _storedPosition + _storedPayloadSize, _storedLimit));

            // sequential small segmented messages in the buffer
            ensureData(size);
        }
    }
    catch (io_exception &) {
        try {
            close();
        } catch (io_exception & ) {
            // noop, best-effort close
        }
        throw connection_closed_exception(
            "Failed to ensure data to read buffer.");
    }
}

void epics::pvAccess::detail::AbstractCodec::flush ( bool  lastMessageCompleted )

virtual

Implements epics::pvAccess::TransportSendControl.

Definition at line 747 of file codec.cpp.

                                                   {

    // automatic end
    endMessage(!lastMessageCompleted);

    // flush send buffer
    flushSendBuffer();

    // start with last header
    if (!lastMessageCompleted && _lastSegmentedMessageType != 0)
        startMessage(_lastSegmentedMessageCommand, 0);
}

void epics::pvAccess::detail::AbstractCodec::flushSendBuffer ( )

protected

Definition at line 726 of file codec.cpp.

                                    {

    _sendBuffer.flip();

    try {
        send(&_sendBuffer);
    } catch (io_exception &) {
        try {
            if (isOpen())
                close();
        } catch (io_exception &) {
            // noop, best-effort close
        }
        throw connection_closed_exception("Failed to send buffer.");
    }

    _sendBuffer.clear();

    _lastMessageStartPosition = std::numeric_limits<size_t>::max();
}

void epics::pvAccess::detail::AbstractCodec::flushSerializeBuffer ( )

virtual

Done with this buffer. Flush it.

Implements epics::pvData::SerializableControl.

Definition at line 722 of file codec.cpp.

                                         {
    flush(false);
}

virtual const osiSockAddr* epics::pvAccess::detail::AbstractCodec::getLastReadBufferSocketAddress ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

epics::pvData::int8 epics::pvAccess::detail::AbstractCodec::getRevision ( ) const

inline

Definition at line 205 of file codec.h.

                                        {
        epicsGuard<epicsMutex> G(_mutex);
        int8_t myver = _clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION;
        return myver < _version ? myver : _version;
    }

virtual void epics::pvAccess::detail::AbstractCodec::invalidDataStreamHandler ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual bool epics::pvAccess::detail::AbstractCodec::isOpen ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::processApplicationMessage ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::processControlMessage ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

void epics::pvAccess::detail::AbstractCodec::processRead

(

)

Definition at line 129 of file codec.cpp.

                                {
    switch (_readMode)
    {
    case NORMAL:
        processReadNormal();
        break;
    case SEGMENTED:
        processReadSegmented();
        break;
    case SPLIT:
        throw std::logic_error("ReadMode == SPLIT not supported");
    }

}

void epics::pvAccess::detail::AbstractCodec::processSendQueue

(

)

Definition at line 833 of file codec.cpp.

{

    {
        std::size_t senderProcessed = 0;
        while (senderProcessed++ < MAX_MESSAGE_SEND)
        {
            TransportSender::shared_pointer sender;
            _sendQueue.pop_front_try(sender);
            if (sender.get() == 0)
            {
                // flush
                if (_sendBuffer.getPosition() > 0)
                    flush(true);

                sendCompleted();    // do not schedule sending

                if (terminated())   // termination
                    break;
                // termination (we want to process even if shutdown)
                _sendQueue.pop_front(sender);
            }

            try {
                processSender(sender);
            } catch(...) {
                if (_sendBuffer.getPosition() > 0)
                    flush(true);
                sendCompleted();
                throw;
            }
        }
    }

    // flush
    if (_sendBuffer.getPosition() > 0)
        flush(true);
}

void epics::pvAccess::detail::AbstractCodec::processWrite

(

)

Definition at line 762 of file codec.cpp.

                                 {

    switch (_writeMode)
    {
    case PROCESS_SEND_QUEUE:
        processSendQueue();
        break;
    case WAIT_FOR_READY_SIGNAL:
        _writeOpReady = true;
        break;
    }
}

void epics::pvAccess::detail::AbstractCodec::putControlMessage	(	epics::pvData::int8	command,
		epics::pvData::int32	data
	)

Definition at line 617 of file codec.cpp.

                           {

    _lastMessageStartPosition =
        std::numeric_limits<size_t>::max();     // TODO revise this
    ensureBuffer(PVA_MESSAGE_HEADER_SIZE);
    _sendBuffer.putByte(PVA_MAGIC);
    _sendBuffer.putByte(_clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION);
    _sendBuffer.putByte((0x01 | _byteOrderFlag | _clientServerFlag));   // control message
    _sendBuffer.putByte(command);   // command
    _sendBuffer.putInt(data);       // data
}

virtual int epics::pvAccess::detail::AbstractCodec::read ( epics::pvData::ByteBuffer *  dst )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::readPollOne ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::scheduleSend ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

void epics::pvAccess::detail::AbstractCodec::send ( epics::pvData::ByteBuffer *  buffer )

protected

Definition at line 776 of file codec.cpp.

{

    // On Windows, limiting the buffer size is important to prevent
    // poor throughput performances when transferring large amount of
    // data over non-blocking socket. See Microsoft KB article KB823764.
    // We do it also for other systems just to be safe.
    std::size_t maxBytesToSend = (size_t)-1;
    //  std::min<int32_t>(
    //  _socketSendBufferSize, _remoteTransportSocketReceiveBufferSize) / 2;

    std::size_t limit = buffer->getLimit();
    std::size_t bytesToSend = limit - buffer->getPosition();

    // limit sending
    if (bytesToSend > maxBytesToSend)
    {
        bytesToSend = maxBytesToSend;
        buffer->setLimit(buffer->getPosition() + bytesToSend);
    }

    int tries = 0;
    while (buffer->getRemaining() > 0)
    {

        //int p = buffer.position();
        int bytesSent = write(buffer);

        if (bytesSent < 0)
        {
            // connection lost
            close();
            throw connection_closed_exception("bytesSent < 0");
        }
        else if (bytesSent == 0)
        {
            sendBufferFull(tries++);
            continue;
        }

        atomic::add(_totalBytesSent, bytesSent);

        // readjust limit
        if (bytesToSend == maxBytesToSend)
        {
            bytesToSend = limit - buffer->getPosition();

            if(bytesToSend > maxBytesToSend)
                bytesToSend = maxBytesToSend;

            buffer->setLimit(buffer->getPosition() + bytesToSend);
        }
        tries = 0;
    }
}

virtual void epics::pvAccess::detail::AbstractCodec::sendBufferFull ( int  tries )

protectedpure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::sendCompleted ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

bool epics::pvAccess::detail::AbstractCodec::sendQueueEmpty ( ) const

inline

Definition at line 201 of file codec.h.

                                {
        return _sendQueue.empty();
    }

void epics::pvAccess::detail::AbstractCodec::setByteOrder ( int  byteOrder )

virtual

Set byte order.

Parameters

byteOrder

byte order to set.

Implements epics::pvAccess::Transport.

Definition at line 952 of file codec.cpp.

{
    _socketBuffer.setEndianess(byteOrder);
    // TODO sync
    _sendBuffer.setEndianess(byteOrder);
    _byteOrderFlag = EPICS_ENDIAN_BIG == byteOrder ? 0x80 : 0x00;
}

void epics::pvAccess::detail::AbstractCodec::setRecipient ( osiSockAddr const &  sendTo )

virtual

Implements epics::pvAccess::TransportSendControl.

Definition at line 947 of file codec.cpp.

                                                           {
    _sendTo = sendTo;
}

virtual void epics::pvAccess::detail::AbstractCodec::setRxTimeout ( bool  ena )

inlineprotectedvirtual

Reimplemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

Definition at line 217 of file codec.h.

{}

void epics::pvAccess::detail::AbstractCodec::setSenderThread

(

)

Definition at line 880 of file codec.cpp.

{
    _senderThread = epicsThreadGetIdSelf();
}

void epics::pvAccess::detail::AbstractCodec::startMessage ( epics::pvData::int8  command, std::size_t  ensureCapacity = 0, epics::pvData::int32  payloadSize = 0  )

virtual

Implements epics::pvAccess::TransportSendControl.

Definition at line 594 of file codec.cpp.

                                  {
    _lastMessageStartPosition =
        std::numeric_limits<size_t>::max();     // TODO revise this
    ensureBuffer(
        PVA_MESSAGE_HEADER_SIZE + ensureCapacity + _nextMessagePayloadOffset);
    _lastMessageStartPosition = _sendBuffer.getPosition();
    _sendBuffer.putByte(PVA_MAGIC);
    _sendBuffer.putByte(_clientServerFlag ? PVA_SERVER_PROTOCOL_REVISION : PVA_CLIENT_PROTOCOL_REVISION);
    _sendBuffer.putByte(
        (_lastSegmentedMessageType | _byteOrderFlag | _clientServerFlag));  // data message
    _sendBuffer.putByte(command);   // command
    _sendBuffer.putInt(payloadSize);

    // apply offset
    if (_nextMessagePayloadOffset > 0)
        _sendBuffer.setPosition(
            _sendBuffer.getPosition() + _nextMessagePayloadOffset);
}

virtual bool epics::pvAccess::detail::AbstractCodec::terminated ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual int epics::pvAccess::detail::AbstractCodec::write ( epics::pvData::ByteBuffer *  src )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

virtual void epics::pvAccess::detail::AbstractCodec::writePollOne ( )

pure virtual

Implemented in epics::pvAccess::detail::BlockingTCPTransportCodec.

Member Data Documentation

const epics::pvData::int8 epics::pvAccess::detail::AbstractCodec::_clientServerFlag

protected

Definition at line 260 of file codec.h.

int8_t epics::pvAccess::detail::AbstractCodec::_command

protected

Definition at line 222 of file codec.h.

int8_t epics::pvAccess::detail::AbstractCodec::_flags

protected

Definition at line 221 of file codec.h.

epics::pvData::Mutex epics::pvAccess::detail::AbstractCodec::_mutex

mutable

Definition at line 264 of file codec.h.

int32_t epics::pvAccess::detail::AbstractCodec::_payloadSize

protected

Definition at line 223 of file codec.h.

ReadMode epics::pvAccess::detail::AbstractCodec::_readMode

protected

Definition at line 219 of file codec.h.

epics::pvData::int32 epics::pvAccess::detail::AbstractCodec::_remoteTransportSocketReceiveBufferSize

protected

Definition at line 224 of file codec.h.

epics::pvData::ByteBuffer epics::pvAccess::detail::AbstractCodec::_sendBuffer

protected

Definition at line 232 of file codec.h.

epicsThreadId epics::pvAccess::detail::AbstractCodec::_senderThread

protected

Definition at line 227 of file codec.h.

fair_queue<TransportSender> epics::pvAccess::detail::AbstractCodec::_sendQueue

protected

Definition at line 234 of file codec.h.

osiSockAddr epics::pvAccess::detail::AbstractCodec::_sendTo

protected

Definition at line 226 of file codec.h.

epics::pvData::ByteBuffer epics::pvAccess::detail::AbstractCodec::_socketBuffer

protected

Definition at line 231 of file codec.h.

int8_t epics::pvAccess::detail::AbstractCodec::_version

protected

Definition at line 220 of file codec.h.

WriteMode epics::pvAccess::detail::AbstractCodec::_writeMode

protected

Definition at line 228 of file codec.h.

bool epics::pvAccess::detail::AbstractCodec::_writeOpReady

protected

Definition at line 229 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_ENSURE_BUFFER_SIZE = MAX_ENSURE_SIZE

static

Definition at line 137 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_ENSURE_DATA_BUFFER_SIZE = 1024

static

Definition at line 138 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_ENSURE_DATA_SIZE = MAX_ENSURE_SIZE/2

static

Definition at line 136 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_ENSURE_SIZE = 1024

static

Definition at line 135 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_MESSAGE_PROCESS = 100

static

Definition at line 133 of file codec.h.

const std::size_t epics::pvAccess::detail::AbstractCodec::MAX_MESSAGE_SEND = 100

static

Definition at line 134 of file codec.h.

---

The documentation for this class was generated from the following files:

• modules/pvAccess/src/remote/pv/codec.h
• modules/pvAccess/src/remote/codec.cpp