#include "codec.h"

Inheritance diagram for epics::pvAccess::detail::BlockingClientTCPTransportCodec:

Collaboration diagram for epics::pvAccess::detail::BlockingClientTCPTransportCodec:

Public Member Functions
	POINTER_DEFINITIONS (BlockingClientTCPTransportCodec)

virtual void	start () OVERRIDE FINAL

virtual	~BlockingClientTCPTransportCodec () OVERRIDE FINAL

virtual void	timerStopped () OVERRIDE FINAL

virtual void	callback () OVERRIDE FINAL

virtual bool	acquire (std::tr1::shared_ptr< ClientChannelImpl > const &client) OVERRIDE FINAL

virtual void	release (pvAccessID clientId) OVERRIDE FINAL

virtual void	send (epics::pvData::ByteBuffer buffer, TransportSendControl control) OVERRIDE FINAL

void	authNZInitialize (const std::vector< std::string > &offeredSecurityPlugins)

virtual void	authenticationCompleted (epics::pvData::Status const &status, const std::tr1::shared_ptr< PeerInfo > &peer) OVERRIDE FINAL

virtual void	verified (epics::pvData::Status const &status) OVERRIDE FINAL

Public Member Functions inherited from epics::pvAccess::detail::BlockingTCPTransportCodec
	POINTER_DEFINITIONS (BlockingTCPTransportCodec)

	BlockingTCPTransportCodec (bool serverFlag, Context::shared_pointer const &context, SOCKET channel, ResponseHandler::shared_pointer const &responseHandler, size_t sendBufferSize, size_t receiveBufferSize, epics::pvData::int16 priority)

virtual	~BlockingTCPTransportCodec ()

virtual void	readPollOne () OVERRIDE FINAL

virtual void	writePollOne () OVERRIDE FINAL

virtual void	scheduleSend () OVERRIDE FINAL

virtual void	sendCompleted () OVERRIDE FINAL

virtual void	close () OVERRIDE FINAL

virtual void	waitJoin () OVERRIDE FINAL
	Call after close() to wait for any worker threads to exit. More...

virtual bool	terminated () OVERRIDE FINAL

virtual bool	isOpen () OVERRIDE FINAL

virtual int	read (epics::pvData::ByteBuffer *dst) OVERRIDE FINAL

virtual int	write (epics::pvData::ByteBuffer *src) OVERRIDE FINAL

virtual const osiSockAddr *	getLastReadBufferSocketAddress () OVERRIDE FINAL

virtual void	invalidDataStreamHandler () OVERRIDE FINAL

virtual std::string	getType () const OVERRIDE FINAL

virtual void	processControlMessage () OVERRIDE FINAL

virtual void	processApplicationMessage () OVERRIDE FINAL

virtual const osiSockAddr &	getRemoteAddress () const OVERRIDE FINAL

virtual const std::string &	getRemoteName () const OVERRIDE FINAL

virtual std::size_t	getReceiveBufferSize () const OVERRIDE FINAL

virtual epics::pvData::int16	getPriority () const OVERRIDE FINAL

virtual void	setRemoteTransportReceiveBufferSize (std::size_t remoteTransportReceiveBufferSize) OVERRIDE FINAL

virtual void	setRemoteTransportSocketReceiveBufferSize (std::size_t socketReceiveBufferSize) OVERRIDE FINAL

virtual std::tr1::shared_ptr< const epics::pvData::Field >	cachedDeserialize (epics::pvData::ByteBuffer *buffer) OVERRIDE FINAL

virtual void	cachedSerialize (const std::tr1::shared_ptr< const epics::pvData::Field > &field, epics::pvData::ByteBuffer *buffer) OVERRIDE FINAL

virtual void	flushSendQueue () OVERRIDE FINAL

virtual bool	isClosed () OVERRIDE FINAL

void	activate ()

virtual bool	verify (epics::pvData::int32 timeoutMs) OVERRIDE

virtual void	authNZMessage (epics::pvData::PVStructure::shared_pointer const &data) OVERRIDE FINAL

virtual void	sendSecurityPluginMessage (epics::pvData::PVStructure::const_shared_pointer const &data) OVERRIDE FINAL

Public Member Functions inherited from epics::pvAccess::detail::AbstractCodec
	AbstractCodec (bool serverFlag, size_t sendBufferSize, size_t receiveBufferSize, int32_t socketSendBufferSize, bool blockingProcessQueue)

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 ()

Public Member Functions inherited from epics::pvData::DeserializableControl
virtual	~DeserializableControl ()

Public Member Functions inherited from epics::pvAccess::AuthenticationPluginControl
	POINTER_DEFINITIONS (AuthenticationPluginControl)

virtual	~AuthenticationPluginControl ()

Public Member Functions inherited from epics::pvAccess::TransportSender
	POINTER_DEFINITIONS (TransportSender)

	TransportSender ()

virtual	~TransportSender ()

Public Member Functions inherited from epics::pvAccess::Lockable
	POINTER_DEFINITIONS (Lockable)

virtual	~Lockable ()

virtual void	lock ()

virtual void	unlock ()

Public Member Functions inherited from epics::pvAccess::fair_queue< T >::entry
	entry ()

	~entry ()

Public Member Functions inherited from epics::pvData::TimerCallback
	POINTER_DEFINITIONS (TimerCallback)

	TimerCallback ()

virtual	~TimerCallback ()

Static Public Member Functions
static shared_pointer	create (Context::shared_pointer const &context, SOCKET channel, ResponseHandler::shared_pointer const &responseHandler, int32_t sendBufferSize, int32_t receiveBufferSize, std::tr1::shared_ptr< ClientChannelImpl > const &client, int8_t remoteTransportRevision, float heartbeatInterval, int16_t priority)

Static Public Member Functions inherited from epics::pvAccess::detail::AbstractCodec
static std::size_t	alignedValue (std::size_t value, std::size_t alignment)

Protected Member Functions
	BlockingClientTCPTransportCodec (Context::shared_pointer const &context, SOCKET channel, ResponseHandler::shared_pointer const &responseHandler, int32_t sendBufferSize, int32_t receiveBufferSize, std::tr1::shared_ptr< ClientChannelImpl > const &client, epics::pvData::int8 remoteTransportRevision, float heartbeatInterval, int16_t priority)

virtual void	internalClose () OVERRIDE FINAL

Protected Member Functions inherited from epics::pvAccess::detail::BlockingTCPTransportCodec
virtual void	setRxTimeout (bool ena) OVERRIDE FINAL

virtual void	sendBufferFull (int tries) OVERRIDE FINAL

Protected Member Functions inherited from epics::pvAccess::detail::AbstractCodec
void	send (epics::pvData::ByteBuffer *buffer)

void	flushSendBuffer ()

Additional Inherited Members
Public Attributes inherited from epics::pvAccess::detail::BlockingTCPTransportCodec
PeerInfo::const_shared_pointer	_peerInfo

Public Attributes inherited from epics::pvAccess::detail::AbstractCodec
epics::pvData::Mutex	_mutex

Public Attributes inherited from epics::pvAccess::Transport
size_t	_totalBytesSent

size_t	_totalBytesRecv

Public Attributes inherited from epics::pvAccess::TransportSender
size_t	bytesTX

size_t	bytesRX

Static Public Attributes inherited from epics::pvAccess::detail::BlockingTCPTransportCodec
static size_t	num_instances

Static Public Attributes inherited from epics::pvAccess::detail::AbstractCodec
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 Attributes inherited from epics::pvAccess::detail::BlockingTCPTransportCodec
osiSockAddr	_socketAddress

std::string	_socketName

Context::shared_pointer	_context

IntrospectionRegistry	_incomingIR

IntrospectionRegistry	_outgoingIR

std::string	_authSessionName

AuthenticationSession::shared_pointer	_authSession

bool	_verified

epics::pvData::Event	_verifiedEvent

Protected Attributes inherited from epics::pvAccess::detail::AbstractCodec
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 558 of file codec.h.

Constructor & Destructor Documentation

epics::pvAccess::detail::BlockingClientTCPTransportCodec::BlockingClientTCPTransportCodec	(	Context::shared_pointer const &	context,
		SOCKET	channel,
		ResponseHandler::shared_pointer const &	responseHandler,
		int32_t	sendBufferSize,
		int32_t	receiveBufferSize,
		std::tr1::shared_ptr< ClientChannelImpl > const &	client,
		epics::pvData::int8	remoteTransportRevision,
		float	heartbeatInterval,
		int16_t	priority
	)

protected

Definition at line 1698 of file codec.cpp.

                        :
     BlockingTCPTransportCodec(false, context, channel, responseHandler,
                               sendBufferSize, receiveBufferSize, priority),
     _connectionTimeout(heartbeatInterval),
     _verifyOrEcho(true),
     sendQueued(true) // don't start sending echo until after auth complete
 {
     // initialize owners list, send queue
     acquire(client);
 
     // use immediate for clients
     //setFlushStrategy(DELAYED);
 }

epics::pvAccess::detail::BlockingClientTCPTransportCodec::~BlockingClientTCPTransportCodec ( )

virtual

Definition at line 1732 of file codec.cpp.

1732 {

1733 }

Member Function Documentation

bool epics::pvAccess::detail::BlockingClientTCPTransportCodec::acquire ( std::tr1::shared_ptr< ClientChannelImpl > const & client )

virtual

Acquires transport.

Parameters

client client (channel) acquiring the transport

Returns: true if transport was granted, false otherwise.

Implements epics::pvAccess::Transport.

Definition at line 1759 of file codec.cpp.

                                                                                             {
     Lock lock(_mutex);
     if(isClosed()) return false;
 
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "Acquiring transport to %s.", _socketName.c_str());
     }
 
     _owners[client->getID()] = ClientChannelImpl::weak_pointer(client);
     //_owners.insert(ClientChannelImpl::weak_pointer(client));
 
     return true;
 }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::authenticationCompleted	(	epics::pvData::Status const &	status,
		const std::tr1::shared_ptr< PeerInfo > &	peer
	)

virtual

Called by server plugin to indicate the the exchange has completed.

Parameters

status	If !status.isSuccess() then the connection will be closed without being used.
peer	Partially initialized PeerInfo. See AuthenticationPlugin::createSession(). PeerInfo::realm and/or PeerInfo::account will now be considered valid. Caller transfers ownership to callee, which may modify.

Implements epics::pvAccess::AuthenticationPluginControl.

Definition at line 1944 of file codec.cpp.

 {
     // noop for client side (server will send ConnectionValidation message)
 }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::authNZInitialize ( const std::vector< std::string > & offeredSecurityPlugins )

Definition at line 1898 of file codec.cpp.

 {
     AuthenticationRegistry& plugins = AuthenticationRegistry::clients();
     std::string selectedName;
     AuthenticationPlugin::shared_pointer plugin;
 
     // because of a missing break; the original SecurityPlugin effectively treated the offered list as being
     // in order of increasing preference (last is preferred).
     // we continue with this because, hey isn't compatibility fun...
 
     for(std::vector<std::string>::const_reverse_iterator it(offeredSecurityPlugins.rbegin()), end(offeredSecurityPlugins.rend());
         it!=end; ++it)
     {
         plugin = plugins.lookup(*it);
         if(plugin) {
             selectedName = *it;
             break;
         }
     }
 
     if(!plugin) {
         // mis-match and legacy.  some early servers (java?) don't advertise any plugins.
         // treat this as anonymous
         selectedName = "anonymous";
         plugin = plugins.lookup(selectedName);
         assert(plugin); // fallback required
     }
 
     {
         PeerInfo::shared_pointer info(new PeerInfo);
         info->peer = _socketName; // this is the server name
         info->transport = "pva";
         info->transportVersion = getRevision();
         info->authority = selectedName;
 
         AuthenticationSession::shared_pointer sess(plugin->createSession(info, shared_from_this(), pvData::PVStructure::shared_pointer()));
 
         Guard G(_mutex);
         _authSessionName = selectedName;
         _authSession = sess;
     }
 
     TransportSender::shared_pointer transportSender = std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
     enqueueSendRequest(transportSender);
 }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::callback ( )

virtual

The method that is called when a timer expires.

Implements epics::pvData::TimerCallback.

Definition at line 1743 of file codec.cpp.

 {
     {
         Guard G(_mutex);
         if(sendQueued) return;
         sendQueued = true;
     }
     // send echo
     TransportSender::shared_pointer transportSender = std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
     enqueueSendRequest(transportSender);
 }

static shared_pointer epics::pvAccess::detail::BlockingClientTCPTransportCodec::create	(	Context::shared_pointer const &	context,
		SOCKET	channel,
		ResponseHandler::shared_pointer const &	responseHandler,
		int32_t	sendBufferSize,
		int32_t	receiveBufferSize,
		std::tr1::shared_ptr< ClientChannelImpl > const &	client,
		int8_t	remoteTransportRevision,
		float	heartbeatInterval,
		int16_t	priority
	)

inlinestatic

Definition at line 579 of file codec.h.

     {
         shared_pointer thisPointer(
             new BlockingClientTCPTransportCodec(
                 context, channel, responseHandler,
                 sendBufferSize, receiveBufferSize,
                 client, remoteTransportRevision,
                 heartbeatInterval, priority)
         );
         thisPointer->activate();
         return thisPointer;
     }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::internalClose ( )

protectedvirtual

Called from close(). after start of shutdown (isOpen()==false) but before worker thread shutdown.

Reimplemented from epics::pvAccess::detail::BlockingTCPTransportCodec.

Definition at line 1775 of file codec.cpp.

                                                     {
     BlockingTCPTransportCodec::internalClose();
 
     TimerCallbackPtr tcb = std::tr1::dynamic_pointer_cast<TimerCallback>(shared_from_this());
     _context->getTimer()->cancel(tcb);
 
     // _owners cannot change when transport is closed
 
     // Notifies clients about disconnect.
 
     // check if still acquired
     size_t refs = _owners.size();
     if(refs>0) {
 
         if (IS_LOGGABLE(logLevelDebug))
         {
             LOG(
                 logLevelDebug,
                 "Transport to %s still has %zu client(s) active and closing...",
                 _socketName.c_str(), refs);
         }
 
         TransportClientMap_t::iterator it = _owners.begin();
         for(; it!=_owners.end(); it++) {
             ClientChannelImpl::shared_pointer client = it->second.lock();
             if (client)
             {
                 EXCEPTION_GUARD(client->transportClosed());
             }
         }
 
     }
 
     _owners.clear();
 }

epics::pvAccess::detail::BlockingClientTCPTransportCodec::POINTER_DEFINITIONS ( BlockingClientTCPTransportCodec )

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::release ( pvAccessID clientId )

virtual

Releases transport.

Parameters

client client (channel) releasing the transport

Implements epics::pvAccess::Transport.

Definition at line 1812 of file codec.cpp.

                                                                  {
     Lock lock(_mutex);
     if(isClosed()) return;
 
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "Releasing TCP transport to %s.", _socketName.c_str());
     }
 
     _owners.erase(clientID);
     //_owners.erase(ClientChannelImpl::weak_pointer(client));
 
     // not used anymore, close it
     // TODO consider delayed destruction (can improve performance!!!)
     if(_owners.size()==0) {
         lock.unlock();
         close();
     }
 }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::send	(	epics::pvData::ByteBuffer *	buffer,
		TransportSendControl *	control
	)

virtual

Called by transport. By this call transport gives callee ownership over the buffer. Calls on TransportSendControl instance must be made from calling thread. Moreover, ownership is valid only for the time of call of this method. NOTE: these limitations allow efficient implementation.

Implements epics::pvAccess::TransportSender.

Definition at line 1832 of file codec.cpp.

 {
     bool voe;
     {
         Guard G(_mutex);
         sendQueued = false;
         voe = _verifyOrEcho;
         _verifyOrEcho = false;
     }
 
     if(voe) {
         /*
          * send verification response message
          */
 
         control->startMessage(CMD_CONNECTION_VALIDATION, 4+2+2);
 
         // receive buffer size
         buffer->putInt(static_cast<int32>(getReceiveBufferSize()));
 
         // max introspection registry size
         // TODO
         buffer->putShort(0x7FFF);
 
         // QoS (aka connection priority)
         buffer->putShort(getPriority());
 
         std::string pluginName;
         AuthenticationSession::shared_pointer session;
         {
             Guard G(_mutex);
             pluginName = _authSessionName;
             session = _authSession;
         }
 
         if (session)
         {
             // selected authNZ plug-in name
             SerializeHelper::serializeString(_authSessionName, buffer, control);
 
             // optional authNZ plug-in initialization data
             SerializationHelper::serializeFull(buffer, control, session->initializationData());
         }
         else
         {
             //TODO: allowed?
             // emptry authNZ plug-in name
             SerializeHelper::serializeString("", buffer, control);
 
             // no authNZ plug-in initialization data
             SerializationHelper::serializeNullField(buffer, control);
         }
 
         // send immediately
         control->flush(true);
     }
     else {
         control->startMessage(CMD_ECHO, 0);
         // send immediately
         control->flush(true);
     }
 
 }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::start ( )

virtual

Reimplemented from epics::pvAccess::detail::BlockingTCPTransportCodec.

Definition at line 1721 of file codec.cpp.

 {
     TimerCallbackPtr tcb = std::tr1::dynamic_pointer_cast<TimerCallback>(shared_from_this());
     // add some randomness to our timer phase
     double R = float(rand())/RAND_MAX; // [0, 1]
     // shape a bit
     R = R*0.5 + 0.5; // [0.5, 1.0]
     _context->getTimer()->schedulePeriodic(tcb, _connectionTimeout/2.0*R, _connectionTimeout/2.0);
     BlockingTCPTransportCodec::start();
 }

virtual void epics::pvAccess::detail::BlockingClientTCPTransportCodec::timerStopped ( )

inlinevirtual

The timer has stopped.

Implements epics::pvData::TimerCallback.

Definition at line 607 of file codec.h.

                                                {
         // noop
     }

void epics::pvAccess::detail::BlockingClientTCPTransportCodec::verified ( epics::pvData::Status const & status )

virtual

Notify transport that it is has been verified.

Parameters

status vefification status;

Reimplemented from epics::pvAccess::detail::BlockingTCPTransportCodec.

Definition at line 1950 of file codec.cpp.

 {
     AuthenticationSession::shared_pointer sess;
     {
         Guard G(_mutex);
         sess = _authSession;
     }
     if(sess)
         sess->authenticationComplete(status);
     this->BlockingTCPTransportCodec::verified(status);
 }

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

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

Public Member Functions

Static Public Member Functions

Protected Member Functions

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation