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

server requests

Browse Source
This commit is contained in:
Gasper Jansa
2011-02-25 17:09:24 +01:00
parent 975e118264
commit d98a386c9e
8 changed files with 2292 additions and 745 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pvAccessApp/Makefile
View File

@@ -44,10 +44,12 @@ INC += serverContext.h
INC += responseHandlers.h
INC += serverChannelImpl.h
INC += baseChannelRequester.h
INC += referencedTransportSender.h
LIBSRCS += responseHandlers.cpp
LIBSRCS += serverContext.cpp
LIBSRCS += serverChannelImpl.cpp
LIBSRCS += baseChannelRequester.cpp
LIBSRCS += referencedTransportSender.h
SRC_DIRS += $(PVACCESS)/factory
LIBSRCS += ChannelAccessFactory.cpp

2
pvAccessApp/remote/remote.h
View File

@@ -113,7 +113,7 @@ namespace epics {
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: TransportSender.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class TransportSender : public ReferenceCountingInstance {
class TransportSender : virtual public ReferenceCountingInstance {
public:
virtual ~TransportSender() {
}

27
pvAccessApp/server/baseChannelRequester.cpp
View File

@@ -23,7 +23,8 @@ BaseChannelRequester::BaseChannelRequester(ServerContextImpl* context, ServerCha
_transport(transport),
_channel(channel),
_context(context),
_pendingRequest(BaseChannelRequester::NULL_REQUEST)
_pendingRequest(BaseChannelRequester::NULL_REQUEST),
_refCount(1)
{
}
@@ -35,7 +36,6 @@ boolean BaseChannelRequester::startRequest(int32 qos)
{
return false;
}
_pendingRequest = qos;
return true;
}
@@ -59,12 +59,12 @@ String BaseChannelRequester::getRequesterName()
return name.str();
}
void BaseChannelRequester::message(const String& message, const epics::pvData::MessageType messageType)
void BaseChannelRequester::message(const String message, const epics::pvData::MessageType messageType)
{
BaseChannelRequester::message(_transport, _ioid, message, messageType);
}
void BaseChannelRequester::message(Transport* transport, const pvAccessID ioid, const String& message, const MessageType messageType)
void BaseChannelRequester::message(Transport* transport, const pvAccessID ioid, const String message, const MessageType messageType)
{
transport->enqueueSendRequest( new BaseChannelRequesterMessageTransportSender(ioid, message, messageType));
}
@@ -73,6 +73,23 @@ void BaseChannelRequester::sendFailureMessage(const int8 command, Transport* tra
{
transport->enqueueSendRequest( new BaseChannelRequesterFailureMessageTransportSender(command, transport, ioid, qos, status));
}
/*
void BaseChannelRequester::release()
{
_mutex.lock();
_refCount--;
_mutex.unlock();
if (_refCount == 0)
{
delete this;
}
}
void BaseChannelRequester::acquire()
{
Lock guard(_mutex);
_refCount++;
}*/
BaseChannelRequesterMessageTransportSender::BaseChannelRequesterMessageTransportSender(const pvAccessID ioid, const String message,const epics::pvData::MessageType messageType):
_ioid(ioid),
@@ -139,7 +156,7 @@ void BaseChannelRequesterFailureMessageTransportSender::unlock()
void BaseChannelRequesterFailureMessageTransportSender::release()
{
delete this;
}
void BaseChannelRequesterFailureMessageTransportSender::acquire()

11
pvAccessApp/server/baseChannelRequester.h
View File

@@ -14,7 +14,7 @@
namespace epics {
namespace pvAccess {
class BaseChannelRequester : public epics::pvData::Requester, public epics::pvData::Destroyable
class BaseChannelRequester : virtual public epics::pvData::Requester, public epics::pvData::Destroyable//, virtual public ReferenceCountingInstance
{
public:
BaseChannelRequester(ServerContextImpl* context, ServerChannelImpl* channel,const pvAccessID ioid, Transport* transport);
@@ -24,9 +24,11 @@ public:
void stopRequest();
int32 getPendingRequest();
String getRequesterName();
void message(const String& message, const epics::pvData::MessageType messageType);
static void message(Transport* transport, const pvAccessID ioid, const String& message, const epics::pvData::MessageType messageType);
void message(const String message, const epics::pvData::MessageType messageType);
static void message(Transport* transport, const pvAccessID ioid, const String message, const epics::pvData::MessageType messageType);
static void sendFailureMessage(const int8 command, Transport* transport, const pvAccessID ioid, const int8 qos, const Status status);
//void release();
//void acquire();
static const Status okStatus;
static const Status badCIDStatus;
@@ -39,11 +41,12 @@ protected:
const pvAccessID _ioid;
Transport* _transport;
ServerChannelImpl* _channel;
epics::pvData::Mutex _mutex;
private:
ServerContextImpl* _context;
static const int32 NULL_REQUEST;
int32 _pendingRequest;
epics::pvData::Mutex _mutex;
int32 _refCount;
};
class BaseChannelRequesterMessageTransportSender : public TransportSender

2462
pvAccessApp/server/responseHandlers.cpp
View File

File diff suppressed because it is too large Load Diff

503
pvAccessApp/server/responseHandlers.h
View File

@@ -12,6 +12,7 @@
#include "remote.h"
#include "serverChannelImpl.h"
#include "baseChannelRequester.h"
#include "referencedTransportSender.h"
namespace epics {
namespace pvAccess {
@@ -41,16 +42,16 @@ namespace epics {
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: BadResponse.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class BadResponse : public AbstractServerResponseHandler {
class ServerBadResponse : public AbstractServerResponseHandler {
public:
/**
* @param context
*/
BadResponse(ServerContextImpl* context) :
ServerBadResponse(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Bad request") {
}
virtual ~BadResponse() {
virtual ~ServerBadResponse() {
}
virtual void handleResponse(osiSockAddr* responseFrom,
@@ -77,7 +78,7 @@ namespace epics {
/**
* Bad response handlers.
*/
BadResponse *_badResponse;
ServerBadResponse *_badResponse;
/**
* Table of response handlers for each command ID.
*/
@@ -90,12 +91,12 @@ namespace epics {
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: ConnectionValidationHandler.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class ConnectionValidationHandler : public AbstractServerResponseHandler {
class ServerConnectionValidationHandler : public AbstractServerResponseHandler {
public:
/**
* @param context
*/
ConnectionValidationHandler(ServerContextImpl* context) :
ServerConnectionValidationHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Connection validation") {
}
@@ -109,13 +110,13 @@ namespace epics {
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: NoopResponse.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class NoopResponse : public AbstractServerResponseHandler {
class ServerNoopResponse : public AbstractServerResponseHandler {
public:
/**
* @param context
* @param description
*/
NoopResponse(ServerContextImpl* context, String description) :
ServerNoopResponse(ServerContextImpl* context, String description) :
AbstractServerResponseHandler(context, description) {
}
};
@@ -125,12 +126,12 @@ namespace epics {
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: EchoHandler.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class EchoHandler : public AbstractServerResponseHandler {
class ServerEchoHandler : public AbstractServerResponseHandler {
public:
/**
* @param context
*/
EchoHandler(ServerContextImpl* context) :
ServerEchoHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Echo request") {
}
@@ -139,17 +140,40 @@ namespace epics {
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class EchoTransportSender : public ReferencedTransportSender {
public:
EchoTransportSender(osiSockAddr* echoFrom) {
memcpy(&_echoFrom, echoFrom, sizeof(osiSockAddr));
}
virtual void send(ByteBuffer* buffer, TransportSendControl* control) {
control->startMessage(CMD_ECHO, 0);
control->setRecipient(_echoFrom);
}
virtual void lock() {
}
virtual void unlock() {
}
private:
osiSockAddr _echoFrom;
virtual ~EchoTransportSender() {
}
};
/**
* Introspection search request handler.
*/
class IntrospectionSearchHandler : public AbstractServerResponseHandler
class ServerIntrospectionSearchHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
IntrospectionSearchHandler(ServerContextImpl* context) :
ServerIntrospectionSearchHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Search request") {
}
@@ -162,15 +186,15 @@ namespace epics {
/**
* Search channel request handler.
*/
class ChannelFindRequesterImplObjectPool;
class SearchHandler : public AbstractServerResponseHandler
class ServerChannelFindRequesterImplObjectPool;
class ServerSearchHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
SearchHandler(ServerContextImpl* context);
~SearchHandler();
ServerSearchHandler(ServerContextImpl* context);
~ServerSearchHandler();
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
@@ -178,57 +202,55 @@ namespace epics {
private:
ChannelProvider* _provider;
ChannelFindRequesterImplObjectPool* _objectPool;
ServerChannelFindRequesterImplObjectPool* _objectPool;
};
class ChannelFindRequesterImpl: public ChannelFindRequester, public TransportSender
class ServerChannelFindRequesterImpl: public ChannelFindRequester, public ReferencedTransportSender
{
public:
ChannelFindRequesterImpl(ServerContextImpl* context, ChannelFindRequesterImplObjectPool* objectPool);
ServerChannelFindRequesterImpl(ServerContextImpl* context, ServerChannelFindRequesterImplObjectPool* objectPool);
void clear();
ChannelFindRequesterImpl* set(int32 searchSequenceId, int32 cid, osiSockAddr* sendTo, boolean responseRequired);
ServerChannelFindRequesterImpl* set(int32 searchSequenceId, int32 cid, osiSockAddr* sendTo, boolean responseRequired);
void channelFindResult(const epics::pvData::Status& status, ChannelFind* channelFind, boolean wasFound);
void lock();
void unlock();
void acquire();
void release();
void send(ByteBuffer* buffer, TransportSendControl* control);
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
int32 _searchSequenceId;
int32 _cid;
osiSockAddr* _sendTo;
boolean _responseRequired;
boolean _wasFound;
epics::pvData::Mutex _mutex;
ServerContextImpl* _context;
ChannelFindRequesterImplObjectPool* _objectPool;
epics::pvData::Mutex _mutex;
ServerChannelFindRequesterImplObjectPool* _objectPool;
};
class ChannelFindRequesterImplObjectPool
class ServerChannelFindRequesterImplObjectPool
{
public:
ChannelFindRequesterImplObjectPool(ServerContextImpl* context);
~ChannelFindRequesterImplObjectPool();
ChannelFindRequesterImpl* get();
void put(ChannelFindRequesterImpl* element);
ServerChannelFindRequesterImplObjectPool(ServerContextImpl* context);
~ServerChannelFindRequesterImplObjectPool();
ServerChannelFindRequesterImpl* get();
void put(ServerChannelFindRequesterImpl* element);
private:
std::vector<ChannelFindRequesterImpl*> _elements;
epics::pvData::Mutex _mutex;
std::vector<ServerChannelFindRequesterImpl*> _elements;
ServerContextImpl* _context;
epics::pvData::Mutex _mutex;
};
/****************************************************************************************/
/**
* Create channel request handler.
*/
class CreateChannelHandler : public AbstractServerResponseHandler
class ServerCreateChannelHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
CreateChannelHandler(ServerContextImpl* context) :
ServerCreateChannelHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Create channel request") {
_provider = context->getChannelProvider();
}
@@ -245,41 +267,38 @@ namespace epics {
ChannelProvider* _provider;
};
class ChannelRequesterImpl : public ChannelRequester, public TransportSender
class ServerChannelRequesterImpl : public ChannelRequester, public ReferencedTransportSender
{
public:
ChannelRequesterImpl(Transport* transport, const String channelName, const pvAccessID cid);
void channelCreated(const Status& status, Channel* channel);
ServerChannelRequesterImpl(Transport* transport, const String channelName, const pvAccessID cid);
void channelCreated(const epics::pvData::Status& status, Channel* channel);
void channelStateChange(Channel* c, const Channel::ConnectionState isConnected);
String getRequesterName();
void message(const String message, const epics::pvData::MessageType messageType);
void lock();
void unlock();
void send(ByteBuffer* buffer, TransportSendControl* control);
void release();
void acquire();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
Transport* _transport;
const String _channelName;
const pvAccessID _cid;
Status _status;
epics::pvData::Status _status;
Channel* _channel;
epics::pvData::Mutex _mutex;
void createChannelFailedResponse(ByteBuffer* buffer, TransportSendControl* control, const Status& status);
void createChannelFailedResponse(epics::pvData::ByteBuffer* buffer, TransportSendControl* control, const epics::pvData::Status& status);
};
/****************************************************************************************/
/**
* Destroy channel request handler.
*/
class DestroyChannelHandler : public AbstractServerResponseHandler
class ServerDestroyChannelHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
DestroyChannelHandler(ServerContextImpl* context) :
ServerDestroyChannelHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Destroy channel request") {
}
@@ -289,14 +308,14 @@ namespace epics {
};
class DestroyChannelHandlerTransportSender : public TransportSender
class ServerDestroyChannelHandlerTransportSender : public ReferencedTransportSender
{
public:
DestroyChannelHandlerTransportSender(pvAccessID cid, pvAccessID sid): _cid(cid), _sid(sid) {
ServerDestroyChannelHandlerTransportSender(pvAccessID cid, pvAccessID sid): _cid(cid), _sid(sid) {
}
void send(ByteBuffer* buffer, TransportSendControl* control) {
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) {
control->startMessage((int8)8, 2*sizeof(int32)/sizeof(int8));
buffer->putInt(_sid);
buffer->putInt(_cid);
@@ -310,14 +329,6 @@ namespace epics {
// noop
}
void release() {
delete this;
}
void acquire() {
// noop
}
private:
pvAccessID _cid;
pvAccessID _sid;
@@ -327,13 +338,13 @@ namespace epics {
/**
* Get request handler.
*/
class GetHandler : public AbstractServerResponseHandler
class ServerGetHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
GetHandler(ServerContextImpl* context) :
ServerGetHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Get request") {
}
@@ -342,10 +353,10 @@ namespace epics {
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ChannelGetRequesterImpl : public BaseChannelRequester, public ChannelGetRequester, public TransportSender
class ServerChannelGetRequesterImpl : public BaseChannelRequester, public ChannelGetRequester, public ReferencedTransportSender
{
public:
ChannelGetRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,
ServerChannelGetRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,
epics::pvData::PVStructurePtr pvRequest);
void channelGetConnect(const epics::pvData::Status& status, ChannelGet* channelGet, epics::pvData::PVStructurePtr pvStructure,
epics::pvData::BitSet* bitSet);
@@ -355,19 +366,379 @@ namespace epics {
* @return the channelGet
*/
ChannelGet* getChannelGet();
String getRequesterName();
void lock();
void unlock();
void release();
void acquire();
void send(ByteBuffer* buffer, TransportSendControl* control);
void message(const String message, const epics::pvData::MessageType messageType);
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelGet* _channelGet;
epics::pvData::BitSet* _bitSet;
epics::pvData::PVStructurePtr _pvStructure;
epics::pvData::Status _status;
epics::pvData::Mutex _mutex;
};
/****************************************************************************************/
/**
* Put request handler.
*/
class ServerPutHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerPutHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Put request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerChannelPutRequesterImpl : public BaseChannelRequester, public ChannelPutRequester, public ReferencedTransportSender
{
public:
ServerChannelPutRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void channelPutConnect(const epics::pvData::Status& status, ChannelPut* channelPut, epics::pvData::PVStructure* pvStructure, epics::pvData::BitSet* bitSet);
void putDone(const epics::pvData::Status& status);
void getDone(const epics::pvData::Status& status);
void lock();
void unlock();
void destroy();
/**
* @return the channelPut
*/
ChannelPut* getChannelPut();
/**
* @return the bitSet
*/
BitSet* getBitSet();
/**
* @return the pvStructure
*/
PVStructure* getPVStructure();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelPut* _channelPut;
epics::pvData::BitSet* _bitSet;
epics::pvData::PVStructure* _pvStructure;
epics::pvData::Status _status;
};
/****************************************************************************************/
/**
* Put request handler.
*/
class ServerPutGetHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerPutGetHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Put-get request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerChannelPutGetRequesterImpl : public BaseChannelRequester, public ChannelPutGetRequester, public ReferencedTransportSender
{
public:
ServerChannelPutGetRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void channelPutGetConnect(const epics::pvData::Status& status, ChannelPutGet* channelPutGet, epics::pvData::PVStructure* pvPutStructure, epics::pvData::PVStructure* pvGetStructure);
void getGetDone(const epics::pvData::Status& status);
void getPutDone(const epics::pvData::Status& status);
void putGetDone(const epics::pvData::Status& status);
void lock();
void unlock();
void destroy();
/**
* @return the channelPutGet
*/
ChannelPutGet* getChannelPutGet();
/**
* @return the pvPutStructure
*/
PVStructure* getPVPutStructure();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelPutGet* _channelPutGet;
epics::pvData::PVStructure* _pvPutStructure;
epics::pvData::PVStructure* _pvGetStructure;
epics::pvData::Status _status;
};
/****************************************************************************************/
/**
* Monitor request handler.
*/
class ServerMonitorHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerMonitorHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Monitor request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerMonitorRequesterImpl : public BaseChannelRequester, public MonitorRequester, public ReferencedTransportSender
{
public:
ServerMonitorRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void monitorConnect(const epics::pvData::Status& status, epics::pvData::Monitor* monitor, epics::pvData::Structure* structure);
void unlisten(epics::pvData::Monitor* monitor);
void monitorEvent(epics::pvData::Monitor* monitor);
void lock();
void unlock();
void destroy();
/**
* @return the channelMonitor
*/
epics::pvData::Monitor* getChannelMonitor();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
epics::pvData::Monitor* _monitor;
epics::pvData::Monitor* _channelMonitor;
epics::pvData::Structure* _structure;
epics::pvData::Status _status;
};
/****************************************************************************************/
/**
* Array request handler.
*/
class ServerArrayHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerArrayHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Array request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerChannelArrayRequesterImpl : public BaseChannelRequester, public ChannelArrayRequester, public ReferencedTransportSender
{
public:
ServerChannelArrayRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void channelArrayConnect(const epics::pvData::Status& status, ChannelArray* channelArray, epics::pvData::PVArray* pvArray);
void getArrayDone(const epics::pvData::Status& status);
void putArrayDone(const epics::pvData::Status& status);
void setLengthDone(const epics::pvData::Status& status);
void lock();
void unlock();
void destroy();
/**
* @return the channelArray
*/
ChannelArray* getChannelArray();
/**
* @return the pvArray
*/
epics::pvData::PVArray* getPVArray();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelArray* _channelArray;
epics::pvData::PVArray* _pvArray;
epics::pvData::Status _status;
};
/****************************************************************************************/
/**
* Cancel request handler.
*/
class ServerCancelRequestHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerCancelRequestHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Cancel request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
private:
/**
* @param transport
* @param ioid
* @param errorStatus
*/
void failureResponse(Transport* transport, pvAccessID ioid, const epics::pvData::Status& errorStatus);
};
/****************************************************************************************/
/**
* Process request handler.
*/
class ServerProcessHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerProcessHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Process request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerChannelProcessRequesterImpl : public BaseChannelRequester, public ChannelProcessRequester, public ReferencedTransportSender
{
public:
ServerChannelProcessRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void channelProcessConnect(const epics::pvData::Status& status, ChannelProcess* channelProcess);
void processDone(const epics::pvData::Status& status);
void lock();
void unlock();
void destroy();
/**
* @return the channelProcess
*/
ChannelProcess* getChannelProcess();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelProcess* _channelProcess;
epics::pvData::Status _status;
int32 _refCount;
};
/****************************************************************************************/
/**
* Get field request handler.
*/
class ServerGetFieldHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerGetFieldHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Get field request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
private:
void getFieldFailureResponse(Transport* transport, const pvAccessID ioid, const epics::pvData::Status& errorStatus);
};
class ServerGetFieldRequesterImpl : public BaseChannelRequester, public GetFieldRequester, public ReferencedTransportSender
{
public:
ServerGetFieldRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport);
void getDone(const epics::pvData::Status& status, epics::pvData::FieldConstPtr field);
void lock();
void unlock();
void destroy();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
epics::pvData::Status _status;
epics::pvData::FieldConstPtr _field;
};
class ServerGetFieldHandlerTransportSender : public ReferencedTransportSender
{
public:
ServerGetFieldHandlerTransportSender(const pvAccessID ioid,const epics::pvData::Status& status, Transport* transport):
_ioid(ioid), _status(status), _transport(transport) {
}
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) {
control->startMessage((int8)17, sizeof(int32)/sizeof(int8));
buffer->putInt(_ioid);
_transport->getIntrospectionRegistry()->serializeStatus(buffer, control, _status);
}
void lock() {
// noop
}
void unlock() {
// noop
}
private:
const pvAccessID _ioid;
const epics::pvData::Status& _status;
Transport* _transport;
};
/****************************************************************************************/
/**
* RPC handler.
*/
class ServerRPCHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
ServerRPCHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "RPC request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ServerChannelRPCRequesterImpl : public BaseChannelRequester, public ChannelRPCRequester, public ReferencedTransportSender
{
public:
ServerChannelRPCRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,epics::pvData::PVStructure* pvRequest);
void channelRPCConnect(const epics::pvData::Status& status, ChannelRPC* channelRPC, epics::pvData::PVStructure* arguments, epics::pvData::BitSet* bitSet);
void requestDone(const epics::pvData::Status& status, epics::pvData::PVStructure* pvResponse);
void lock();
void unlock();
void destroy();
/**
* @return the channelRPC
*/
ChannelRPC* getChannelRPC();
/**
* @return the pvArguments
*/
PVStructure* getPvArguments();
/**
* @return the agrumentsBitSet
*/
BitSet* getAgrumentsBitSet();
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelRPC* _channelRPC;
epics::pvData::PVStructure* _pvArguments;
epics::pvData::PVStructure* _pvResponse;
epics::pvData::BitSet* _argumentsBitSet;
epics::pvData::Status _status;
};
}
}

25
pvAccessApp/server/serverContext.cpp
View File

@@ -30,7 +30,6 @@ ServerContextImpl::ServerContextImpl():
_receiveBufferSize(MAX_TCP_RECV),
_timer(NULL),
_broadcastTransport(NULL),
_broadcastConnector(NULL),
_beaconEmitter(NULL),
_acceptor(NULL),
_transportRegistry(NULL),
@@ -105,8 +104,7 @@ void ServerContextImpl::loadConfiguration()
void ServerContextImpl::initialize(ChannelAccess* channelAccess)
{
//TODO uncomment
/*Lock guard(_mutex);
Lock guard(_mutex);
if (channelAccess == NULL)
{
THROW_BASE_EXCEPTION("non null channelAccess expected");
@@ -128,7 +126,7 @@ void ServerContextImpl::initialize(ChannelAccess* channelAccess)
{
std::string msg = "Channel provider with name '" + _channelProviderName + "' not available.";
THROW_BASE_EXCEPTION(msg.c_str());
}*/
}
internalInitialize();
@@ -137,6 +135,7 @@ void ServerContextImpl::initialize(ChannelAccess* channelAccess)
void ServerContextImpl::internalInitialize()
{
//TODO should be allocated on stack
_timer = new Timer("pvAccess-server timer",lowerPriority);
_transportRegistry = new TransportRegistry();
@@ -144,14 +143,14 @@ void ServerContextImpl::internalInitialize()
initializeBroadcastTransport();
_acceptor = new BlockingTCPAcceptor(this, _serverPort, _receiveBufferSize);
_serverPort = _acceptor->getBindAddress()->ia.sin_port;
//TODO fix this
//_serverPort = _acceptor->getBindAddress()->ia.sin_port;
_beaconEmitter = new BeaconEmitter(_broadcastTransport, this);
}
void ServerContextImpl::initializeBroadcastTransport()
{
// setup UDP transport
try
{
@@ -161,7 +160,7 @@ void ServerContextImpl::initializeBroadcastTransport()
listenLocalAddress.ia.sin_port = htons(_broadcastPort);
listenLocalAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
// where to send address
// where to send addresses
SOCKET socket = epicsSocketCreate(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (socket == INVALID_SOCKET)
{
@@ -502,9 +501,15 @@ Transport* ServerContextImpl::getSearchTransport()
//TODO
return NULL;
}
// TODO
void ServerContextImpl::acquire() {}
void ServerContextImpl::release() {}
void ServerContextImpl::acquire()
{
// TODO
}
void ServerContextImpl::release()
{
// TODO
}
}
}

5
pvAccessApp/server/serverContext.h
View File

@@ -334,11 +334,6 @@ private:
*/
BlockingUDPTransport* _broadcastTransport;
/**
* Broadcast connector
*/
BlockingUDPConnector* _broadcastConnector;
/**
* Beacon emitter.
*/