#include "codec.h"

Inheritance diagram for epics::pvAccess::detail::BlockingServerTCPTransportCodec:

Collaboration diagram for epics::pvAccess::detail::BlockingServerTCPTransportCodec:

Public Member Functions
	POINTER_DEFINITIONS (BlockingServerTCPTransportCodec)

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

virtual void	release (pvAccessID) OVERRIDE FINAL

pvAccessID	preallocateChannelSID ()

void	depreallocateChannelSID (pvAccessID)

void	registerChannel (pvAccessID sid, std::tr1::shared_ptr< ServerChannel > const &channel)

void	unregisterChannel (pvAccessID sid)

std::tr1::shared_ptr< ServerChannel >	getChannel (pvAccessID sid)

void	getChannels (std::vector< std::tr1::shared_ptr< ServerChannel > > &channels) const

size_t	getChannelCount () const

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

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

void	authNZInitialize (const std::string &securityPluginName, const epics::pvData::PVStructure::shared_pointer &data)

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

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

virtual	~BlockingServerTCPTransportCodec () 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 void	start ()

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

Static Public Member Functions
static shared_pointer	create (Context::shared_pointer const &context, SOCKET channel, ResponseHandler::shared_pointer const &responseHandler, int sendBufferSize, int receiveBufferSize)

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
	BlockingServerTCPTransportCodec (Context::shared_pointer const &context, SOCKET channel, ResponseHandler::shared_pointer const &responseHandler, int32_t sendBufferSize, int32_t receiveBufferSize)

void	destroyAllChannels ()

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 441 of file codec.h.

Constructor & Destructor Documentation

epics::pvAccess::detail::BlockingServerTCPTransportCodec::BlockingServerTCPTransportCodec	(	Context::shared_pointer const &	context,
		SOCKET	channel,
		ResponseHandler::shared_pointer const &	responseHandler,
		int32_t	sendBufferSize,
		int32_t	receiveBufferSize
	)

protected

Definition at line 1434 of file codec.cpp.

     :BlockingTCPTransportCodec(true, context, channel, responseHandler,
                                sendBufferSize, receiveBufferSize, PVA_DEFAULT_PRIORITY)
     ,_lastChannelSID(0x12003400)
     ,_verificationStatus(pvData::Status::fatal("Uninitialized error"))
     ,_verifyOrVerified(false)
 {
     // NOTE: priority not yet known, default priority is used to
     //register/unregister
     // TODO implement priorities in Reactor... not that user will
     // change it.. still getPriority() must return "registered" priority!
 }

epics::pvAccess::detail::BlockingServerTCPTransportCodec::~BlockingServerTCPTransportCodec ( )

virtual

Definition at line 1453 of file codec.cpp.

1453 {

1454 }

Member Function Documentation

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

inlinevirtual

Acquires transport.

Parameters

client client (channel) acquiring the transport

Returns: true if transport was granted, false otherwise.

Implements epics::pvAccess::Transport.

Definition at line 475 of file codec.h.

     {
         return false;
     }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::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 1629 of file codec.cpp.

 {
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "Authentication completed with status '%s' for PVA client: %s.", Status::StatusTypeName[status.getType()], _socketName.c_str());
     }
 
     if(peer)
         AuthorizationRegistry::plugins().run(peer);
 
     bool isVerified;
     {
         Guard G(_mutex);
         isVerified = _verified;
         if(status.isSuccess())
             _peerInfo = peer;
         else
             _peerInfo.reset();
     }
 
     if (!isVerified)
         verified(status);
     else if (!status.isSuccess())
     {
         string errorMessage = "Re-authentication failed: " + status.getMessage();
         if (!status.getStackDump().empty())
             errorMessage += "\n" + status.getStackDump();
         LOG(logLevelInfo, "%s", errorMessage.c_str());
 
         close();
     }
 }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::authNZInitialize	(	const std::string &	securityPluginName,
		const epics::pvData::PVStructure::shared_pointer &	data
	)

Definition at line 1663 of file codec.cpp.

 {
     AuthenticationPlugin::shared_pointer plugin(AuthenticationRegistry::servers().lookup(securityPluginName));
     // attempting the force use of an un-advertised/non-existant plugin is treated as a protocol error.
     // We cheat here by assuming the the registry doesn't often change after server start,
     // and don't test if securityPluginName is in advertisedAuthPlugins
     if(!plugin)
         throw std::runtime_error(_socketName+" failing attempt to select non-existant auth. plugin "+securityPluginName);
 
     PeerInfo::shared_pointer info(new PeerInfo);
     info->peer = _socketName;
     info->transport = "pva";
     info->transportVersion = getRevision();
     info->authority = securityPluginName;
 
     if (!plugin->isValidFor(*info))
         verified(pvData::Status::error("invalid security plug-in name"));
 
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(logLevelDebug, "Accepted security plug-in '%s' for PVA client: %s.", securityPluginName.c_str(), _socketName.c_str());
     }
 
     AuthenticationSession::shared_pointer sess(plugin->createSession(info, shared_from_this(), data));
 
     Guard G(_mutex);
     _authSessionName = securityPluginName;
     _authSession.swap(sess);
 }

static shared_pointer epics::pvAccess::detail::BlockingServerTCPTransportCodec::create	(	Context::shared_pointer const &	context,
		SOCKET	channel,
		ResponseHandler::shared_pointer const &	responseHandler,
		int	sendBufferSize,
		int	receiveBufferSize
	)

inlinestatic

Definition at line 457 of file codec.h.

     {
         shared_pointer thisPointer(
             new BlockingServerTCPTransportCodec(
                 context, channel, responseHandler,
                 sendBufferSize, receiveBufferSize)
         );
         thisPointer->activate();
         return thisPointer;
     }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::depreallocateChannelSID ( pvAccessID )

inline

Definition at line 484 of file codec.h.

484 {}

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::destroyAllChannels ( )

protected

Definition at line 1604 of file codec.cpp.

                                                          {
     Lock lock(_channelsMutex);
     if(_channels.size()==0) return;
 
     if (IS_LOGGABLE(logLevelDebug))
     {
         LOG(
             logLevelDebug,
             "Transport to %s still has %zu channel(s) active and closing...",
             _socketName.c_str(), _channels.size());
     }
 
     _channels_t temp;
     temp.swap(_channels);
 
     for(_channels_t::iterator it(temp.begin()), end(temp.end()); it!=end; ++it)
         it->second->destroy();
 }

ServerChannel::shared_pointer epics::pvAccess::detail::BlockingServerTCPTransportCodec::getChannel ( pvAccessID sid )

Definition at line 1486 of file codec.cpp.

                                                           {
 
     Lock lock(_channelsMutex);
 
     std::map<pvAccessID, ServerChannel::shared_pointer>::iterator it =
         _channels.find(sid);
 
     if(it!=_channels.end()) return it->second;
 
     return ServerChannel::shared_pointer();
 }

size_t epics::pvAccess::detail::BlockingServerTCPTransportCodec::getChannelCount ( ) const

Definition at line 1499 of file codec.cpp.

                                                               {
 
     Lock lock(_channelsMutex);
     return _channels.size();
 }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::getChannels ( std::vector< std::tr1::shared_ptr< ServerChannel > > & channels ) const

Definition at line 1505 of file codec.cpp.

 {
     Lock lock(_channelsMutex);
     for(_channels_t::const_iterator it(_channels.begin()), end(_channels.end());
         it!=end; ++it)
     {
         channels.push_back(it->second);
     }
 }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::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 1623 of file codec.cpp.

                                                     {
     Transport::shared_pointer thisSharedPtr = shared_from_this();
     BlockingTCPTransportCodec::internalClose();
     destroyAllChannels();
 }

epics::pvAccess::detail::BlockingServerTCPTransportCodec::POINTER_DEFINITIONS ( BlockingServerTCPTransportCodec )

pvAccessID epics::pvAccess::detail::BlockingServerTCPTransportCodec::preallocateChannelSID ( )

Definition at line 1457 of file codec.cpp.

                                                                   {
 
     Lock lock(_channelsMutex);
     // search first free (theoretically possible loop of death)
     pvAccessID sid = ++_lastChannelSID;
     while(_channels.find(sid)!=_channels.end())
         sid = ++_lastChannelSID;
     return sid;
 }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::registerChannel	(	pvAccessID	sid,
		std::tr1::shared_ptr< ServerChannel > const &	channel
	)

Definition at line 1468 of file codec.cpp.

                                                  {
 
     Lock lock(_channelsMutex);
     _channels[sid] = channel;
 
 }

virtual void epics::pvAccess::detail::BlockingServerTCPTransportCodec::release ( pvAccessID clientId )

inlinevirtual

Releases transport.

Parameters

client client (channel) releasing the transport

Implements epics::pvAccess::Transport.

Definition at line 480 of file codec.h.

480 {}

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::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 1515 of file codec.cpp.

                                        {
 
     if (!_verifyOrVerified)
     {
         _verifyOrVerified = true;
 
         //
         // set byte order control message
         //
 
         ensureBuffer(PVA_MESSAGE_HEADER_SIZE);
         buffer->putByte(PVA_MAGIC);
         buffer->putByte(PVA_SERVER_PROTOCOL_REVISION);
         buffer->putByte(
             0x01 | 0x40 | ((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG)
                            ? 0x80 : 0x00));     // control + server + endian
         buffer->putByte(CMD_SET_ENDIANESS);     // set byte order
         buffer->putInt(0);
 
 
         //
         // send verification message
         //
         control->startMessage(CMD_CONNECTION_VALIDATION, 4+2);
 
         // receive buffer size
         buffer->putInt(static_cast<int32>(getReceiveBufferSize()));
 
         // server introspection registy max size
         // TODO
         buffer->putShort(0x7FFF);
 
         // list of authNZ plugin names advertised to this client
 
         AuthenticationRegistry::list_t plugins;
         AuthenticationRegistry::servers().snapshot(plugins); // copy
         std::vector<std::string> validSPNames;
         validSPNames.reserve(plugins.size()); // assume all will be valid
 
         PeerInfo info;
         info.transport = "pva";
         info.peer = _socketName;
         info.transportVersion = this->getRevision();
 
         // filter plugins which may be used by this peer
         for(AuthenticationRegistry::list_t::iterator it(plugins.begin()), end(plugins.end());
             it!=end; ++it)
         {
             info.authority = it->first;
             if(it->second->isValidFor(info))
                 validSPNames.push_back(it->first);
         }
 
         SerializeHelper::writeSize(validSPNames.size(), buffer, this);
         for (vector<string>::const_iterator iter(validSPNames.begin()), end(validSPNames.end());
              iter != end; iter++)
         {
             SerializeHelper::serializeString(*iter, buffer, this);
         }
 
         {
             Guard G(_mutex);
             advertisedAuthPlugins.swap(validSPNames);
         }
 
         // send immediately
         control->flush(true);
     }
     else
     {
         //
         // send verified message
         //
         control->startMessage(CMD_CONNECTION_VALIDATED, 0);
 
         pvData::Status sts;
         {
             Lock lock(_mutex);
             sts = _verificationStatus;
         }
         sts.serialize(buffer, control);
 
         // send immediately
         control->flush(true);
 
     }
 }

void epics::pvAccess::detail::BlockingServerTCPTransportCodec::unregisterChannel ( pvAccessID sid )

Definition at line 1478 of file codec.cpp.

                                                                       {
 
     Lock lock(_channelsMutex);
     _channels.erase(sid);
 }

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

inlinevirtual

Notify transport that it is has been verified.

Parameters

status vefification status;

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

Definition at line 511 of file codec.h.

                                                                            {
         {
             epicsGuard<epicsMutex> G(_mutex);
             _verificationStatus = status;
         }
         BlockingTCPTransportCodec::verified(status);
     }

virtual bool epics::pvAccess::detail::BlockingServerTCPTransportCodec::verify ( epics::pvData::int32 timeoutMs )

inlinevirtual

Waits (if needed) until transport is verified, i.e. verified() method is being called.

Parameters

timeoutMs timeout to wait for verification, infinite if 0.

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

Definition at line 498 of file codec.h.

                                                                    {
 
         TransportSender::shared_pointer transportSender =
             std::tr1::dynamic_pointer_cast<TransportSender>(shared_from_this());
         enqueueSendRequest(transportSender);
 
         bool verifiedStatus = BlockingTCPTransportCodec::verify(timeoutMs);
 
         enqueueSendRequest(transportSender);
 
         return verifiedStatus;
     }

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