epics/pvAccess
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

IntrospectionRegistry changes, does not compile

Browse Source
This commit is contained in:
Matej Sekoranja
2012-07-24 23:00:20 +02:00
parent 1b6587336e
commit 3ea38d79dd
18 changed files with 174 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pvAccessApp/Makefile
View File

@@ -44,6 +44,7 @@ INC += blockingTCP.h
INC += channelSearchManager.h
INC += simpleChannelSearchManagerImpl.h
INC += transportRegistry.h
INC += serializationHelper.h
LIBSRCS += blockingUDPTransport.cpp
LIBSRCS += blockingUDPConnector.cpp
LIBSRCS += beaconHandler.cpp
@@ -55,6 +56,7 @@ LIBSRCS += simpleChannelSearchManagerImpl.cpp
LIBSRCS += abstractResponseHandler.cpp
LIBSRCS += blockingTCPAcceptor.cpp
LIBSRCS += transportRegistry.cpp
#LIBSRCS += serializationHelper.cpp
SRC_DIRS += $(PVACCESS)/remoteClient
INC += clientContextImpl.h

3
pvAccessApp/remote/blockingClientTCPTransport.cpp
View File

@@ -30,7 +30,6 @@ namespace epics {
TransportClient::shared_pointer client, int8 remoteTransportRevision,
float beaconInterval, int16 priority) :
BlockingTCPTransport(context, channel, responseHandler, receiveBufferSize, priority),
_introspectionRegistry(false),
_connectionTimeout(beaconInterval*1000),
_unresponsiveTransport(false),
_verifyOrEcho(true)
@@ -193,7 +192,7 @@ namespace epics {
}
void BlockingClientTCPTransport::changedTransport() {
_introspectionRegistry.reset();
outgoingIR.reset();
Lock lock(_mutex);
TransportClientMap_t::iterator it = _owners.begin();

1
pvAccessApp/remote/blockingServerTCPTransport.cpp
View File

@@ -24,7 +24,6 @@ namespace pvAccess {
Context::shared_pointer const & context, SOCKET channel,
auto_ptr<ResponseHandler>& responseHandler, int receiveBufferSize) :
BlockingTCPTransport(context, channel, responseHandler, receiveBufferSize, CA_DEFAULT_PRIORITY),
_introspectionRegistry(true),
_lastChannelSID(0)
{
// for performance testing

60
pvAccessApp/remote/blockingTCP.h
View File

@@ -169,6 +169,17 @@ namespace epics {
virtual void flushSendQueue();
virtual void cachedSerialize(
const std::tr1::shared_ptr<const epics::pvData::Field>& field, epics::pvData::ByteBuffer* buffer)
{
outgoingIR.serialize(field, buffer, this);
}
virtual std::tr1::shared_ptr<const epics::pvData::Field>
cachedDeserialize(epics::pvData::ByteBuffer* buffer)
{
return incomingIR.deserialize(buffer, this);
}
protected:
@@ -330,6 +341,10 @@ namespace epics {
epics::pvData::int8 _byteOrderFlag;
/**
* Outgoing (codes generated by this party) introspection registry.
*/
IntrospectionRegistry outgoingIR;
@@ -360,7 +375,10 @@ namespace epics {
epics::pvData::int64 _totalBytesReceived;
/**
* Incoming (codes generated by other party) introspection registry.
*/
IntrospectionRegistry incomingIR;
@@ -478,22 +496,6 @@ namespace epics {
virtual void callback();
virtual IntrospectionRegistry* getIntrospectionRegistry() {
return &_introspectionRegistry;
}
virtual void cachedSerialize(
const std::tr1::shared_ptr<const epics::pvData::Field>& field, epics::pvData::ByteBuffer* buffer)
{
getIntrospectionRegistry()->serialize(field, buffer, this);
}
virtual std::tr1::shared_ptr<const epics::pvData::Field>
cachedDeserialize(epics::pvData::ByteBuffer* buffer)
{
return getIntrospectionRegistry()->deserialize(buffer, this);
}
/**
* Acquires transport.
* @param client client (channel) acquiring the transport
@@ -541,10 +543,6 @@ namespace epics {
TransportSendControl* control);
protected:
/**
* Introspection registry.
*/
IntrospectionRegistry _introspectionRegistry;
virtual void internalClose(bool force);
virtual void internalPostClose(bool force);
@@ -664,22 +662,6 @@ namespace epics {
return thisPointer;
}
virtual IntrospectionRegistry* getIntrospectionRegistry() {
return &_introspectionRegistry;
}
virtual void cachedSerialize(
const std::tr1::shared_ptr<const epics::pvData::Field>& field, epics::pvData::ByteBuffer* buffer)
{
getIntrospectionRegistry()->serialize(field, buffer, this);
}
virtual std::tr1::shared_ptr<const epics::pvData::Field>
cachedDeserialize(epics::pvData::ByteBuffer* buffer)
{
return getIntrospectionRegistry()->deserialize(buffer, this);
}
/**
* Preallocate new channel SID.
* @return new channel server id (SID).
@@ -775,10 +757,6 @@ namespace epics {
virtual ~BlockingServerTCPTransport();
protected:
/**
* Introspection registry.
*/
IntrospectionRegistry _introspectionRegistry;
virtual void internalClose(bool force);
virtual void internalPostClose(bool force);

4
pvAccessApp/remote/blockingUDP.h
View File

@@ -217,10 +217,6 @@ namespace epics {
}
}
virtual IntrospectionRegistry* getIntrospectionRegistry() {
return 0;
}
protected:
AtomicBoolean _closed;

7
pvAccessApp/remote/remote.h
View File

@@ -9,7 +9,6 @@
#include <pv/caConstants.h>
#include <pv/configuration.h>
#include <pv/introspectionRegistry.h>
#include <pv/serialize.h>
#include <pv/pvType.h>
@@ -222,12 +221,6 @@ namespace epics {
*/
virtual void changedTransport() = 0;
/**
* Get introspection registry for transport.
* @return <code>IntrospectionRegistry</code> instance, can be <code>null</code>.
*/
virtual IntrospectionRegistry* getIntrospectionRegistry() = 0;
/**
* Close transport.
* @param force flag indicating force-full (e.g. remote disconnect) close.

1
pvAccessApp/remote/simpleChannelSearchManagerImpl.cpp
View File

@@ -7,6 +7,7 @@
#include <pv/simpleChannelSearchManagerImpl.h>
#include <pv/caConstants.h>
#include <pv/blockingUDP.h>
#include <pv/serializeHelper.h>
#include <stdlib.h>
#include <time.h>

56
pvAccessApp/remoteClient/clientContextImpl.cpp
View File

@@ -28,6 +28,7 @@
#include <pv/beaconHandler.h>
#include <pv/logger.h>
#include <pv/bitSetUtil.h>
#include <pv/serializationHelper.h>
#include <tr1/unordered_map>
@@ -180,7 +181,7 @@ namespace epics {
int8 qos = payloadBuffer->getByte();
Status m_status;
transport->getIntrospectionRegistry()->deserializeStatus(m_status, payloadBuffer, transport.get());
m_status.deserialize(payloadBuffer, transport.get());
try
{
@@ -372,7 +373,7 @@ namespace epics {
if (pendingRequest & QOS_INIT)
{
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
stopRequest();
@@ -527,7 +528,7 @@ namespace epics {
if (pendingRequest & QOS_INIT)
{
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
stopRequest();
@@ -553,7 +554,7 @@ namespace epics {
// create data and its bitSet
{
Lock lock(m_structureMutex);
m_structure = transport->getIntrospectionRegistry()->deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_structure = SerializationHelper::deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_bitSet.reset(new BitSet(m_structure->getNumberFields()));
}
@@ -731,7 +732,7 @@ namespace epics {
if (pendingRequest & QOS_INIT)
{
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
else if (!(pendingRequest & QOS_GET))
{
@@ -766,7 +767,7 @@ namespace epics {
// create data and its bitSet
{
Lock lock(m_structureMutex);
m_structure = transport->getIntrospectionRegistry()->deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_structure = SerializationHelper::deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_bitSet.reset(new BitSet(m_structure->getNumberFields()));
}
@@ -961,7 +962,7 @@ namespace epics {
buffer->putByte((int8)QOS_INIT);
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
else if (pendingRequest & (QOS_GET | QOS_GET_PUT)) {
// noop
@@ -993,12 +994,10 @@ namespace epics {
return true;
}
IntrospectionRegistry* registry = transport->getIntrospectionRegistry();
{
Lock lock(m_structureMutex);
m_putData = registry->deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_getData = registry->deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_putData = SerializationHelper::deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
m_getData = SerializationHelper::deserializeStructureAndCreatePVStructure(payloadBuffer, transport.get());
}
// notify
@@ -1249,14 +1248,14 @@ namespace epics {
buffer->putByte((int8)QOS_INIT);
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
else
{
{
// no need to lock here, since it is already locked via TransportSender IF
//Lock lock(m_structureMutex);
m_channel->getTransport()->getIntrospectionRegistry()->serializeStructure(buffer, control, m_structure);
SerializationHelper::serializeStructureFull(buffer, control, m_structure);
// release arguments structure
m_structure.reset();
}
@@ -1294,7 +1293,7 @@ namespace epics {
}
PVStructure::shared_pointer response(transport->getIntrospectionRegistry()->deserializeStructure(payloadBuffer, transport.get()));
PVStructure::shared_pointer response(SerializationHelper::deserializeStructureFull(payloadBuffer, transport.get()));
EXCEPTION_GUARD(m_channelRPCRequester->requestDone(status, response));
return true;
}
@@ -1444,7 +1443,7 @@ namespace epics {
if (pendingRequest & QOS_INIT)
{
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
else if (pendingRequest & QOS_GET)
{
@@ -1487,7 +1486,7 @@ namespace epics {
}
// create data and its bitSet
FieldConstPtr field = transport->getIntrospectionRegistry()->deserialize(payloadBuffer, transport.get());
FieldConstPtr field = transport->cachedDeserialize(payloadBuffer);
{
Lock lock(m_structureMutex);
m_structure = dynamic_pointer_cast<PVArray>(getPVDataCreate()->createPVField(field));
@@ -1774,11 +1773,11 @@ namespace epics {
virtual void response(Transport::shared_pointer const & transport, int8 version, ByteBuffer* payloadBuffer) {
Status status;
transport->getIntrospectionRegistry()->deserializeStatus(status, payloadBuffer, transport.get());
status.deserialize(payloadBuffer, transport.get());
if (status.isSuccess())
{
// deserialize Field...
FieldConstPtr field = transport->getIntrospectionRegistry()->deserialize(payloadBuffer, transport.get());
FieldConstPtr field = transport->cachedDeserialize(payloadBuffer);
EXCEPTION_GUARD(m_callback->getDone(status, field));
}
else
@@ -1854,7 +1853,7 @@ namespace epics {
return m_monitorElement;
}
virtual void release(MonitorElement::shared_pointer & monitorElement) {
virtual void release(MonitorElement::shared_pointer const & monitorElement) {
Lock guard(m_mutex);
m_gotMonitor = false;
}
@@ -1925,7 +1924,7 @@ namespace epics {
return m_monitorElement;
}
virtual void release(MonitorElement::shared_pointer & monitorElement) {
virtual void release(MonitorElement::shared_pointer const & monitorElement) {
Lock guard(m_mutex);
m_gotMonitor = false;
}
@@ -2037,7 +2036,7 @@ namespace epics {
return m_monitorElement;
}
virtual void release(MonitorElement::shared_pointer & monitorElement) {
virtual void release(MonitorElement::shared_pointer const & monitorElement) {
Lock guard(m_mutex);
m_gotMonitor = false;
}
@@ -2171,7 +2170,7 @@ namespace epics {
if (pendingRequest & QOS_INIT)
{
// pvRequest
m_channel->getTransport()->getIntrospectionRegistry()->serializePVRequest(buffer, control, m_pvRequest);
SerializationHelper::serializePVRequest(buffer, control, m_pvRequest);
}
stopRequest();
@@ -2194,8 +2193,7 @@ namespace epics {
StructureConstPtr structure =
dynamic_pointer_cast<const Structure>(
transport->getIntrospectionRegistry()->
deserialize(payloadBuffer, transport.get())
transport->cachedDeserialize(payloadBuffer)
);
m_monitorStrategy->init(structure);
@@ -2235,7 +2233,7 @@ namespace epics {
if (qos & QOS_INIT)
{
Status status;
transport->getIntrospectionRegistry()->deserializeStatus(status, payloadBuffer, transport.get());
status.deserialize(payloadBuffer, transport.get());
if (status.isSuccess())
{
m_mutex.lock();
@@ -2247,7 +2245,7 @@ namespace epics {
else if (qos & QOS_DESTROY)
{
Status status;
transport->getIntrospectionRegistry()->deserializeStatus(status, payloadBuffer, transport.get());
status.deserialize(payloadBuffer, transport.get());
m_mutex.lock();
m_initialized = false;
@@ -2328,7 +2326,7 @@ namespace epics {
return m_monitorStrategy->poll();
}
virtual void release(MonitorElement::shared_pointer & monitorElement)
virtual void release(MonitorElement::shared_pointer const & monitorElement)
{
m_monitorStrategy->release(monitorElement);
}
@@ -2548,7 +2546,7 @@ namespace epics {
// TODO smart pointers
// extra data
PVFieldPtr data;
const FieldConstPtr field = IntrospectionRegistry::deserializeFull(payloadBuffer, transport.get());
const FieldConstPtr field = getFieldCreate()->deserialize(payloadBuffer, transport.get());
if (field != 0)
{
data = getPVDataCreate()->createPVField(field);
@@ -2648,7 +2646,7 @@ namespace epics {
pvAccessID sid = payloadBuffer->getInt();
Status status;
transport->getIntrospectionRegistry()->deserializeStatus(status, payloadBuffer, transport.get());
status.deserialize(payloadBuffer, transport.get());
// TODO optimize
ChannelImpl::shared_pointer channel = static_pointer_cast<ChannelImpl>(_context.lock()->getChannel(cid));

2
pvAccessApp/server/baseChannelRequester.cpp
View File

@@ -122,7 +122,7 @@ void BaseChannelRequesterFailureMessageTransportSender::send(ByteBuffer* buffer,
control->startMessage(_command, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->put(_qos);
_transport->getIntrospectionRegistry()->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
void BaseChannelRequesterFailureMessageTransportSender::lock()

7
pvAccessApp/server/beaconEmitter.cpp
View File

@@ -9,8 +9,7 @@
#endif
#include <pv/beaconEmitter.h>
#include <pv/introspectionRegistry.h>
#include <pv/serializationHelper.h>
#include <pv/logger.h>
#include <algorithm>
@@ -101,12 +100,12 @@ void BeaconEmitter::send(ByteBuffer* buffer, TransportSendControl* control)
if (serverStatus)
{
// introspection interface + data
IntrospectionRegistry::serializeFull(serverStatus->getField(), buffer, control);
serverStatus->getField()->serialize(buffer, control);
serverStatus->serialize(buffer, control);
}
else
{
IntrospectionRegistry::serializeFull(FieldConstPtr(), buffer, control);
SerializationHelper::serializeNullField(buffer, control);
}
control->flush(true);

63
pvAccessApp/server/responseHandlers.cpp
View File

@@ -7,6 +7,7 @@
#include <pv/responseHandlers.h>
#include <pv/remote.h>
#include <pv/hexDump.h>
#include <pv/serializationHelper.h>
#include <pv/byteBuffer.h>
@@ -459,7 +460,7 @@ void ServerChannelRequesterImpl::send(ByteBuffer* buffer, TransportSendControl*
control->startMessage((int8)CMD_CREATE_CHANNEL, 2*sizeof(int32)/sizeof(int8));
buffer->putInt(serverChannelImpl->getCID());
buffer->putInt(serverChannelImpl->getSID());
transport->getIntrospectionRegistry()->serializeStatus(buffer, control, status);
status.serialize(buffer, control);
}
}
@@ -471,7 +472,7 @@ void ServerChannelRequesterImpl::createChannelFailedResponse(ByteBuffer* buffer,
control->startMessage((int8)CMD_CREATE_CHANNEL, 2*sizeof(int32)/sizeof(int8));
buffer->putInt(_cid);
buffer->putInt(-1);
transport->getIntrospectionRegistry()->serializeStatus(buffer, control, status);
status.serialize(buffer, control);
}
}
@@ -546,7 +547,7 @@ void ServerGetHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelGetRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -675,10 +676,9 @@ void ServerChannelGetRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
control->startMessage((int8)CMD_GET, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->put((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
if (_status.isSuccess())
@@ -686,7 +686,7 @@ void ServerChannelGetRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
if (request & QOS_INIT)
{
Lock guard(_mutex);
introspectionRegistry->serialize(_pvStructure != NULL ? _pvStructure->getField() : FieldConstPtr(), buffer, control);
control->cachedSerialize(_pvStructure != NULL ? _pvStructure->getField() : FieldConstPtr(), buffer);
}
else
@@ -736,7 +736,7 @@ void ServerPutHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelPutRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -891,10 +891,9 @@ void ServerChannelPutRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
control->startMessage((int32)CMD_PUT, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
if (_status.isSuccess())
@@ -902,7 +901,7 @@ void ServerChannelPutRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
if ((QOS_INIT & request) != 0)
{
Lock guard(_mutex);
introspectionRegistry->serialize(_pvStructure != NULL ? _pvStructure->getField() : FieldConstPtr(), buffer, control);
control->cachedSerialize(_pvStructure != NULL ? _pvStructure->getField() : FieldConstPtr(), buffer);
}
else if ((QOS_GET & request) != 0)
{
@@ -947,7 +946,7 @@ void ServerPutGetHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelPutGetRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -1109,10 +1108,9 @@ void ServerChannelPutGetRequesterImpl::send(ByteBuffer* buffer, TransportSendCon
control->startMessage((int32)12, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
if (_status.isSuccess())
@@ -1120,8 +1118,8 @@ void ServerChannelPutGetRequesterImpl::send(ByteBuffer* buffer, TransportSendCon
if ((QOS_INIT & request) != 0)
{
Lock guard(_mutex);
introspectionRegistry->serialize(_pvPutStructure != NULL ? _pvPutStructure->getField() : FieldConstPtr(), buffer, control);
introspectionRegistry->serialize(_pvGetStructure != NULL ? _pvGetStructure->getField() : FieldConstPtr(), buffer, control);
control->cachedSerialize(_pvPutStructure != NULL ? _pvPutStructure->getField() : FieldConstPtr(), buffer);
control->cachedSerialize(_pvGetStructure != NULL ? _pvGetStructure->getField() : FieldConstPtr(), buffer);
}
else if ((QOS_GET & request) != 0)
{
@@ -1177,7 +1175,7 @@ void ServerMonitorHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerMonitorRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -1246,7 +1244,7 @@ void ServerMonitorRequesterImpl::activate(PVStructure::shared_pointer const & pv
INIT_EXCEPTION_GUARD(CMD_MONITOR, _channelMonitor = _channel->getChannel()->createMonitor(thisPointer, pvRequest));
}
void ServerMonitorRequesterImpl::monitorConnect(const Status& status, Monitor::shared_pointer & monitor, epics::pvData::StructureConstPtr const & structure)
void ServerMonitorRequesterImpl::monitorConnect(const Status& status, Monitor::shared_pointer const & monitor, epics::pvData::StructureConstPtr const & structure)
{
{
Lock guard(_mutex);
@@ -1327,16 +1325,15 @@ void ServerMonitorRequesterImpl::send(ByteBuffer* buffer, TransportSendControl*
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
if (_status.isSuccess())
{
// valid due to _mutex lock above
introspectionRegistry->serialize(_structure, buffer, control);
control->cachedSerialize(_structure, buffer);
}
stopRequest();
startRequest(QOS_DEFAULT);
@@ -1395,7 +1392,7 @@ void ServerArrayHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelArrayRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -1564,10 +1561,9 @@ void ServerChannelArrayRequesterImpl::send(ByteBuffer* buffer, TransportSendCont
control->startMessage((int32)CMD_ARRAY, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
if (_status.isSuccess())
@@ -1581,7 +1577,7 @@ void ServerChannelArrayRequesterImpl::send(ByteBuffer* buffer, TransportSendCont
else if ((QOS_INIT & request) != 0)
{
Lock guard(_mutex);
introspectionRegistry->serialize(_pvArray != NULL ? _pvArray->getField() : FieldConstPtr(), buffer, control);
control->cachedSerialize(_pvArray != NULL ? _pvArray->getField() : FieldConstPtr(), buffer);
}
}
@@ -1660,7 +1656,7 @@ void ServerProcessHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelProcessRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -1775,10 +1771,9 @@ void ServerChannelProcessRequesterImpl::send(ByteBuffer* buffer, TransportSendCo
control->startMessage((int32)CMD_PROCESS, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
stopRequest();
@@ -1862,11 +1857,10 @@ void ServerGetFieldRequesterImpl::send(ByteBuffer* buffer, TransportSendControl*
{
control->startMessage((int8)CMD_GET_FIELD, sizeof(int32)/sizeof(int8));
buffer->putInt(_ioid);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
introspectionRegistry->serialize(_field, buffer, control);
_status.serialize(buffer, control);
control->cachedSerialize(_field, buffer);
}
}
@@ -1898,7 +1892,7 @@ void ServerRPCHandler::handleResponse(osiSockAddr* responseFrom,
if (init)
{
// pvRequest
PVStructure::shared_pointer pvRequest(transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvRequest(SerializationHelper::deserializePVRequest(payloadBuffer, transport.get()));
// create...
ServerChannelRPCRequesterImpl::create(_context, channel, ioid, transport, pvRequest);
@@ -1923,7 +1917,7 @@ void ServerRPCHandler::handleResponse(osiSockAddr* responseFrom,
// deserialize put data
ChannelRPC::shared_pointer channelRPC = request->getChannelRPC();
// pvArgument
PVStructure::shared_pointer pvArgument(transport->getIntrospectionRegistry()->deserializeStructure(payloadBuffer, transport.get()));
PVStructure::shared_pointer pvArgument(SerializationHelper::deserializeStructureFull(payloadBuffer, transport.get()));
channelRPC->request(pvArgument, lastRequest);
}
}
@@ -2020,11 +2014,10 @@ void ServerChannelRPCRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
control->startMessage((int32)CMD_RPC, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
if (_status.isSuccess())
{
@@ -2034,7 +2027,7 @@ void ServerChannelRPCRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
}
else
{
introspectionRegistry->serializeStructure(buffer, control, _pvResponse);
SerializationHelper::serializeStructureFull(buffer, control, _pvResponse);
}
}
}

4
pvAccessApp/server/responseHandlers.h
View File

@@ -482,7 +482,7 @@ namespace pvAccess {
ServerChannelImpl::shared_pointer const & channel, const pvAccessID ioid,
Transport::shared_pointer const & transport,epics::pvData::PVStructure::shared_pointer const & pvRequest);
void monitorConnect(const epics::pvData::Status& status, epics::pvData::Monitor::shared_pointer & monitor, epics::pvData::StructureConstPtr const & structure);
void monitorConnect(const epics::pvData::Status& status, epics::pvData::Monitor::shared_pointer const & monitor, epics::pvData::StructureConstPtr const & structure);
void unlisten(epics::pvData::Monitor::shared_pointer const & monitor);
void monitorEvent(epics::pvData::Monitor::shared_pointer const & monitor);
void lock();
@@ -674,7 +674,7 @@ namespace pvAccess {
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) {
control->startMessage((epics::pvData::int8)CMD_GET_FIELD, sizeof(epics::pvData::int32)/sizeof(epics::pvData::int8));
buffer->putInt(_ioid);
_transport->getIntrospectionRegistry()->serializeStatus(buffer, control, _status);
_status.serialize(buffer, control);
}
void lock() {

1
pvAccessApp/utils/configuration.h
View File

@@ -21,6 +21,7 @@
#include <fstream>
#include <map>
// TODO implement using smart pointers
namespace epics {
namespace pvAccess {

2
pvAccessApp/utils/inetAddressUtil.h
View File

@@ -13,6 +13,8 @@
#include <osiSock.h>
#include <vector>
// TODO implement using smart pointers
namespace epics {
namespace pvAccess {

190
pvAccessApp/utils/introspectionRegistry.cpp
View File

@@ -17,12 +17,10 @@ namespace pvAccess {
const int8 IntrospectionRegistry::NULL_TYPE_CODE = (int8)-1;
const int8 IntrospectionRegistry::ONLY_ID_TYPE_CODE = (int8)-2;
const int8 IntrospectionRegistry::FULL_WITH_ID_TYPE_CODE = (int8)-3;
PVDataCreatePtr IntrospectionRegistry::_pvDataCreate (getPVDataCreate());
FieldCreatePtr IntrospectionRegistry::_fieldCreate(getFieldCreate());
IntrospectionRegistry::IntrospectionRegistry(bool serverSide)
IntrospectionRegistry::IntrospectionRegistry()
{
_direction = serverSide ? 1 : -1;
reset();
}
@@ -33,37 +31,29 @@ IntrospectionRegistry::~IntrospectionRegistry()
void IntrospectionRegistry::reset()
{
Lock guard(_mutex);
_outgoingIdPointer = _direction;
_pointer = 1;
_registry.clear();
}
FieldConstPtr IntrospectionRegistry::getIntrospectionInterface(const short id)
FieldConstPtr IntrospectionRegistry::getIntrospectionInterface(const int16 id)
{
Lock guard(_mutex);
_registryIter = _registry.find(id);
if(_registryIter == _registry.end())
registryMap_t::iterator registryIter = _registry.find(id);
if(registryIter == _registry.end())
{
return FieldConstPtr();
}
return _registryIter->second;
return registryIter->second;
}
void IntrospectionRegistry::registerIntrospectionInterface(const short id,FieldConstPtr field)
void IntrospectionRegistry::registerIntrospectionInterface(const int16 id, FieldConstPtr const & field)
{
Lock guard(_mutex);
_registryIter = _registry.find(id);
_registry[id] = field;
}
short IntrospectionRegistry::registerIntrospectionInterface(FieldConstPtr field, bool& existing)
int16 IntrospectionRegistry::registerIntrospectionInterface(FieldConstPtr const & field, bool& existing)
{
Lock guard(_mutex);
short key;
int16 key;
// TODO this is slow
if(registryContainsValue(field, key))
{
existing = true;
@@ -71,19 +61,8 @@ short IntrospectionRegistry::registerIntrospectionInterface(FieldConstPtr field,
else
{
existing = false;
key = _outgoingIdPointer;
_outgoingIdPointer += _direction;
// wrap check
if(_outgoingIdPointer * _direction < 0)
{
_outgoingIdPointer = _direction;
}
//first decrement reference on old value
_registryIter = _registry.find(key);
key = _pointer++;
_registry[key] = field;
}
return key;
}
@@ -92,43 +71,97 @@ void IntrospectionRegistry::printKeysAndValues(string name)
{
string buffer;
cout << "############## print of all key/values of " << name.c_str() << " registry : ###################" << endl;
for(_registryIter = _registry.begin(); _registryIter != _registry.end(); _registryIter++)
for(registryMap_t::iterator registryIter = _registry.begin(); registryIter != _registry.end(); registryIter++)
{
buffer.clear();
cout << "\t" << "Key: "<< _registryIter->first << endl;
cout << "\t" << "Value: " << _registryIter->second << endl;
cout << "\t" << "Key: "<< registryIter->first << endl;
cout << "\t" << "Value: " << registryIter->second << endl;
cout << "\t" << "References: " << buffer.c_str() << endl;
buffer.clear();
_registryIter->second->toString(&buffer);
registryIter->second->toString(&buffer);
cout << "\t" << "Value toString: " << buffer.c_str() << endl;
}
}
// TODO !!!!
bool IntrospectionRegistry::registryContainsValue(FieldConstPtr field, short& key)
// TODO slow !!!!
bool IntrospectionRegistry::registryContainsValue(FieldConstPtr const & field, int16& key)
{
for(_registryRIter = _registry.rbegin(); _registryRIter != _registry.rend(); _registryRIter++)
for(registryMap_t::reverse_iterator registryRIter = _registry.rbegin(); registryRIter != _registry.rend(); registryRIter++)
{
if((*field) == (*_registryRIter->second))
if(*(field.get()) == *(registryRIter->second))
{
key = _registryRIter->first;
key = registryRIter->first;
return true;
}
}
return false;
}
void IntrospectionRegistry::serialize(FieldConstPtr field, ByteBuffer* buffer, SerializableControl* control)
void IntrospectionRegistry::serialize(FieldConstPtr const & field, ByteBuffer* buffer, SerializableControl* control)
{
serialize(field, StructureConstPtr(), buffer, control, this);
if (field.get() == NULL)
{
// TODO
//SerializationHelper::serializeNullField(buffer, control);
control->ensureBuffer(1);
buffer->putByte(IntrospectionRegistry::NULL_TYPE_CODE);
}
else
{
// only structures registry check
if (field->getType() == structure)
{
bool existing;
const int16 key = registerIntrospectionInterface(field, existing);
if (existing) {
control->ensureBuffer(3);
buffer->putByte(ONLY_ID_TYPE_CODE);
buffer->putShort(key);
return;
}
else {
control->ensureBuffer(3);
buffer->putByte(FULL_WITH_ID_TYPE_CODE); // could also be a mask
buffer->putShort(key);
}
}
field->serialize(buffer, control);
}
}
FieldConstPtr IntrospectionRegistry::deserialize(ByteBuffer* buffer, DeserializableControl* control)
{
return deserialize(buffer, control, this);
control->ensureData(1);
size_t pos = buffer->getPosition();
const int8 typeCode = buffer->getByte();
if (typeCode == NULL_TYPE_CODE)
{
return FieldConstPtr();
}
else if (typeCode == ONLY_ID_TYPE_CODE)
{
control->ensureData(sizeof(int16)/sizeof(int8));
return getIntrospectionInterface(buffer->getShort());
}
// could also be a mask
if(typeCode == IntrospectionRegistry::FULL_WITH_ID_TYPE_CODE)
{
control->ensureData(sizeof(int16)/sizeof(int8));
const short key = buffer->getShort();
FieldConstPtr field = _fieldCreate->deserialize(buffer, control);
registerIntrospectionInterface(key, field);
return field;
}
// return typeCode back
buffer->setPosition(pos);
return _fieldCreate->deserialize(buffer, control);
}
/*
void IntrospectionRegistry::serializeFull(FieldConstPtr field, ByteBuffer* buffer, SerializableControl* control)
{
serialize(field, StructureConstPtr(), buffer, control, NULL);
@@ -139,73 +172,6 @@ FieldConstPtr IntrospectionRegistry::deserializeFull(ByteBuffer* buffer, Deseria
return deserialize(buffer, control, NULL);
}
void IntrospectionRegistry::serialize(FieldConstPtr field, StructureConstPtr parent, ByteBuffer* buffer,
SerializableControl* control, IntrospectionRegistry* registry)
{
if (field == NULL)
{
control->ensureBuffer(1);
buffer->putByte(IntrospectionRegistry::NULL_TYPE_CODE);
}
else
{
// use registry check
// only top IFs and structures
if (registry != NULL && (parent == NULL || field->getType() == epics::pvData::structure || field->getType() == epics::pvData::structureArray))
{
bool existing;
const short key = registry->registerIntrospectionInterface(field, existing);
if(existing)
{
control->ensureBuffer(1+sizeof(int16)/sizeof(int8));
buffer->putByte(ONLY_ID_TYPE_CODE);
buffer->putShort(key);
return;
}
else
{
control->ensureBuffer(1+sizeof(int16)/sizeof(int8));
buffer->putByte(FULL_WITH_ID_TYPE_CODE); // could also be a mask
buffer->putShort(key);
}
}
field->serialize(buffer, control);
}
}
FieldConstPtr IntrospectionRegistry::deserialize(ByteBuffer* buffer, DeserializableControl* control, IntrospectionRegistry* registry)
{
control->ensureData(1);
size_t pos = buffer->getPosition();
const int8 typeCode = buffer->getByte();
if(typeCode == IntrospectionRegistry::NULL_TYPE_CODE)
{
return FieldConstPtr();
}
else if(typeCode == IntrospectionRegistry::ONLY_ID_TYPE_CODE)
{
control->ensureData(sizeof(int16)/sizeof(int8));
FieldConstPtr field = registry->getIntrospectionInterface(buffer->getShort());
return field;
}
// could also be a mask
if(typeCode == IntrospectionRegistry::FULL_WITH_ID_TYPE_CODE)
{
control->ensureData(sizeof(int16)/sizeof(int8));
const short key = buffer->getShort();
FieldConstPtr field = deserialize(buffer, control, registry);
registry->registerIntrospectionInterface(key, field);
return field;
}
buffer->setPosition(pos);
// TODO
return getFieldCreate()->deserialize(buffer, control);
}
void IntrospectionRegistry::serializeStructure(ByteBuffer* buffer, SerializableControl* control, PVStructurePtr pvStructure)
{
if (pvStructure == NULL)
@@ -261,7 +227,7 @@ void IntrospectionRegistry::deserializeStatus(Status &status, ByteBuffer* buffer
{
status.deserialize(buffer, control);
}
*/
}}

116
pvAccessApp/utils/introspectionRegistry.h
View File

@@ -34,7 +34,7 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
*/
class IntrospectionRegistry : public epics::pvData::NoDefaultMethods {
public:
IntrospectionRegistry(bool serverSide);
IntrospectionRegistry();
virtual ~IntrospectionRegistry();
void printKeysAndValues(std::string name);
@@ -49,7 +49,7 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
*
* @return <code>Field</code> instance for given ID.
*/
epics::pvData::FieldConstPtr getIntrospectionInterface(const short id);
epics::pvData::FieldConstPtr getIntrospectionInterface(const epics::pvData::int16 id);
/**
* Registers introspection interface with given ID. Always INCOMING.
@@ -57,7 +57,7 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
* @param id id of the introspection interface to register
* @param field introspection interface to register
*/
void registerIntrospectionInterface(const short id, epics::pvData::FieldConstPtr field);
void registerIntrospectionInterface(const epics::pvData::int16 id, epics::pvData::FieldConstPtr const & field);
/**
* Registers introspection interface and get it's ID. Always OUTGOING.
@@ -69,7 +69,7 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
*
* @return id of given introspection interface
*/
short registerIntrospectionInterface(epics::pvData::FieldConstPtr field, bool& existing);
epics::pvData::int16 registerIntrospectionInterface(epics::pvData::FieldConstPtr const & field, bool& existing);
/**
* Serializes introspection interface
@@ -78,7 +78,7 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
* @param buffer
* @param control
*/
void serialize(epics::pvData::FieldConstPtr field, epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control);
void serialize(epics::pvData::FieldConstPtr const & field, epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control);
/**
* Deserializes introspection interface
@@ -92,30 +92,6 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
*/
epics::pvData::FieldConstPtr deserialize(epics::pvData::ByteBuffer* buffer, epics::pvData::DeserializableControl* control);
/**
* Serializes introspection interface. But this time really fully not like
* the serialize which only says it serializes but in fact does not. :)
*
* TODO
*
* @param field
* @param buffer
* @param control
*/
static void serializeFull(epics::pvData::FieldConstPtr field, epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control);
/**
* Deserializes introspection interface
*
* TODO
*
* @param buffer
* @param control
*
* @return <code>Field</code> deserialized from the buffer.
*/
static epics::pvData::FieldConstPtr deserializeFull(epics::pvData::ByteBuffer* buffer, epics::pvData::DeserializableControl* control);
/**
* Null type.
*/
@@ -131,94 +107,16 @@ typedef std::map<const short,epics::pvData::FieldConstPtr> registryMap_t;
*/
const static epics::pvData::int8 FULL_WITH_ID_TYPE_CODE;
static void serialize(epics::pvData::FieldConstPtr field, epics::pvData::StructureConstPtr parent, epics::pvData::ByteBuffer* buffer,
epics::pvData::SerializableControl* control, IntrospectionRegistry* registry);
/**
* @param buffer
* @param control
* @param registry
* @param structure
*/
static epics::pvData::FieldConstPtr deserialize(epics::pvData::ByteBuffer* buffer, epics::pvData::DeserializableControl* control, IntrospectionRegistry* registry);
/**
* Serialize optional PVStructrue.
* @param buffer data buffer.
*/
void serializeStructure(epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control, epics::pvData::PVStructurePtr pvStructure);
/**
* Deserialize optional PVStructrue.
* @param payloadBuffer data buffer.
* @return deserialized PVStructure, can be <code>null</code>.
*/
epics::pvData::PVStructurePtr deserializeStructure(epics::pvData::ByteBuffer* payloadBuffer, epics::pvData::DeserializableControl* control);
/**
* Serialize PVRequest.
* @param buffer data buffer.
*/
void serializePVRequest(epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control, epics::pvData::PVStructurePtr pvRequest);
/**
* Deserialize PVRequest.
* @param payloadBuffer data buffer.
* @param control serialization control.
*
* @return deserialized PVRequest, can be <code>null</code>.
*/
epics::pvData::PVStructurePtr deserializePVRequest(epics::pvData::ByteBuffer* payloadBuffer, epics::pvData::DeserializableControl* control);
/**
* Deserialize Structure and create PVStructure instance.
*
* @param payloadBuffer data buffer.
* @param control serialization control.
*
* @return PVStructure instance, can be <code>null</code>.
*/
epics::pvData::PVStructurePtr deserializeStructureAndCreatePVStructure(epics::pvData::ByteBuffer* payloadBuffer, epics::pvData::DeserializableControl* control);
/**
* Serialize status.
*
* @param buffer data buffer.
* @param control serialization control.
* @param status status to serialize.
*/
void serializeStatus(epics::pvData::ByteBuffer* buffer, epics::pvData::SerializableControl* control, const epics::pvData::Status &status);
/**
* Deserialize status.
*
* @param buffer data buffer.
* @param control serialization control.
*/
void deserializeStatus(epics::pvData::Status &status, epics::pvData::ByteBuffer* buffer, epics::pvData::DeserializableControl* control);
private:
registryMap_t _registry;
registryMap_t::iterator _registryIter;
registryMap_t::reverse_iterator _registryRIter;
short _outgoingIdPointer;
short _direction;
epics::pvData::Mutex _mutex;
/**
* PVField factory.
*/
static epics::pvData::PVDataCreatePtr _pvDataCreate;
epics::pvData::int16 _pointer;
/**
* Field factory.
*/
static epics::pvData::FieldCreatePtr _fieldCreate;
bool registryContainsValue(epics::pvData::FieldConstPtr field, short& key);
bool registryContainsValue(epics::pvData::FieldConstPtr const & field, epics::pvData::int16& key);
};
}}

2
testApp/remote/testRemoteClientImpl.cpp
View File

@@ -366,7 +366,7 @@ class MonitorRequesterImpl : public MonitorRequester
std::cout << "[" << getRequesterName() << "] message(" << message << ", " << getMessageTypeName(messageType) << ")" << std::endl;
}
virtual void monitorConnect(const epics::pvData::Status& status, Monitor::shared_pointer & monitor, StructureConstPtr const & structure)
virtual void monitorConnect(const epics::pvData::Status& status, Monitor::shared_pointer const & monitor, StructureConstPtr const & structure)
{
std::cout << "monitorConnect(" << status.toString() << ")" << std::endl;
if (status.isSuccess() && structure)

2
testApp/remote/testServer.cpp
View File

@@ -643,7 +643,7 @@ class MockMonitor : public Monitor, public MonitorElement, public std::tr1::enab
}
}
virtual void release(MonitorElement::shared_pointer & monitorElement)
virtual void release(MonitorElement::shared_pointer const & monitorElement)
{
Lock xx(m_lock);
if (m_count)