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

merged

Browse Source
This commit is contained in:
Matej Sekoranja
2011-02-21 15:11:57 +01:00
parent 71d65f4d46 1aecd168ab
commit 49ebe7787c
8 changed files with 476 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pvAccessApp/Makefile
View File

@@ -43,10 +43,11 @@ SRC_DIRS += $(PVACCESS)/server
INC += serverContext.h
INC += responseHandlers.h
INC += serverChannelImpl.h
INC += baseChannelRequester.h
LIBSRCS += responseHandlers.cpp
LIBSRCS += serverContext.cpp
LIBSRCS += serverChannelImpl.cpp
LIBSRCS += baseChannelRequester.cpp
SRC_DIRS += $(PVACCESS)/factory
LIBSRCS += ChannelAccessFactory.cpp

151
pvAccessApp/server/baseChannelRequester.cpp Normal file
View File

@@ -0,0 +1,151 @@
/*
* baseChannelRequester.cpp
*/
#include "baseChannelRequester.h"
using namespace epics::pvData;
namespace epics { namespace pvAccess {
const Status BaseChannelRequester::okStatus = Status();
const Status BaseChannelRequester::badCIDStatus = Status(Status::STATUSTYPE_ERROR, "bad channel id");
const Status BaseChannelRequester::badIOIDStatus = Status(Status::STATUSTYPE_ERROR, "bad request id");
const Status BaseChannelRequester::noReadACLStatus = Status(Status::STATUSTYPE_ERROR, "no read access");
const Status BaseChannelRequester::noWriteACLStatus = Status(Status::STATUSTYPE_ERROR, "no write access");
const Status BaseChannelRequester::noProcessACLStatus = Status(Status::STATUSTYPE_ERROR, "no process access");
const Status BaseChannelRequester::otherRequestPendingStatus = Status(Status::STATUSTYPE_ERROR, "other request pending");
const int32 BaseChannelRequester::NULL_REQUEST = -1;
BaseChannelRequester::BaseChannelRequester(ServerContextImpl* context, ServerChannelImpl* channel,const pvAccessID ioid, Transport* transport) :
_ioid(ioid),
_transport(transport),
_channel(channel),
_context(context),
_pendingRequest(BaseChannelRequester::NULL_REQUEST)
{
}
boolean BaseChannelRequester::startRequest(int32 qos)
{
Lock guard(_mutex);
if (_pendingRequest != NULL_REQUEST)
{
return false;
}
_pendingRequest = qos;
return true;
}
void BaseChannelRequester::stopRequest()
{
Lock guard(_mutex);
_pendingRequest = NULL_REQUEST;
}
int32 BaseChannelRequester::getPendingRequest()
{
Lock guard(_mutex);
return _pendingRequest;
}
String BaseChannelRequester::getRequesterName()
{
stringstream name;
name << typeid(*_transport).name() << "/" << _ioid;
return name.str();
}
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)
{
transport->enqueueSendRequest( new BaseChannelRequesterMessageTransportSender(ioid, message, messageType));
}
void BaseChannelRequester::sendFailureMessage(const int8 command, Transport* transport, const pvAccessID ioid, const int8 qos, const Status status)
{
transport->enqueueSendRequest( new BaseChannelRequesterFailureMessageTransportSender(command, transport, ioid, qos, status));
}
BaseChannelRequesterMessageTransportSender::BaseChannelRequesterMessageTransportSender(const pvAccessID ioid, const String message,const epics::pvData::MessageType messageType):
_ioid(ioid),
_message(message),
_messageType(messageType)
{
}
void BaseChannelRequesterMessageTransportSender::send(ByteBuffer* buffer, TransportSendControl* control)
{
control->startMessage((int8)18, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->putByte((int8)_messageType);
epics::pvData::SerializeHelper::serializeString(_message, buffer, control);
}
void BaseChannelRequesterMessageTransportSender::lock()
{
// noop
}
void BaseChannelRequesterMessageTransportSender::unlock()
{
// noop
}
void BaseChannelRequesterMessageTransportSender::release()
{
delete this;
}
void BaseChannelRequesterMessageTransportSender::acquire()
{
// noop
}
BaseChannelRequesterFailureMessageTransportSender::BaseChannelRequesterFailureMessageTransportSender(const int8 command,
Transport* transport, const pvAccessID ioid, const int8 qos, const Status status) :
_command(command),
_ioid(ioid),
_qos(qos),
_status(status),
_transport(transport)
{
}
void BaseChannelRequesterFailureMessageTransportSender::send(ByteBuffer* buffer, TransportSendControl* control)
{
control->startMessage(_command, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->put(_qos);
_transport->getIntrospectionRegistry()->serializeStatus(buffer, control, _status);
}
void BaseChannelRequesterFailureMessageTransportSender::lock()
{
// noop
}
void BaseChannelRequesterFailureMessageTransportSender::unlock()
{
// noop
}
void BaseChannelRequesterFailureMessageTransportSender::release()
{
delete this;
}
void BaseChannelRequesterFailureMessageTransportSender::acquire()
{
// noop
}
}
}

86
pvAccessApp/server/baseChannelRequester.h Normal file
View File

@@ -0,0 +1,86 @@
/*
* baseChannelRequester.h
*/
#ifndef BASECHANNELREQUESTER_H_
#define BASECHANNELREQUESTER_H_
#include "serverContext.h"
#include "serverChannelImpl.h"
#include <requester.h>
#include <destroyable.h>
namespace epics {
namespace pvAccess {
class BaseChannelRequester : public epics::pvData::Requester, public epics::pvData::Destroyable
{
public:
BaseChannelRequester(ServerContextImpl* context, ServerChannelImpl* channel,const pvAccessID ioid, Transport* transport);
~BaseChannelRequester() {};
boolean startRequest(int32 qos);
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);
static void sendFailureMessage(const int8 command, Transport* transport, const pvAccessID ioid, const int8 qos, const Status status);
static const Status okStatus;
static const Status badCIDStatus;
static const Status badIOIDStatus;
static const Status noReadACLStatus;
static const Status noWriteACLStatus;
static const Status noProcessACLStatus;
static const Status otherRequestPendingStatus;
protected:
const pvAccessID _ioid;
Transport* _transport;
ServerChannelImpl* _channel;
private:
ServerContextImpl* _context;
static const int32 NULL_REQUEST;
int32 _pendingRequest;
epics::pvData::Mutex _mutex;
};
class BaseChannelRequesterMessageTransportSender : public TransportSender
{
public:
BaseChannelRequesterMessageTransportSender(const pvAccessID _ioid, const String message,const epics::pvData::MessageType messageType);
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
void lock();
void unlock();
void release();
void acquire();
private:
const pvAccessID _ioid;
const String _message;
const epics::pvData::MessageType _messageType;
};
class BaseChannelRequesterFailureMessageTransportSender : public TransportSender
{
public:
BaseChannelRequesterFailureMessageTransportSender(const int8 command, Transport* transport, const pvAccessID ioid, const int8 qos, const Status status);
void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control);
void lock();
void unlock();
void release();
void acquire();
private:
const int8 _command;
const pvAccessID _ioid;
const int8 _qos;
const Status _status;
Transport* _transport;
};
}
}
#endif /* BASECHANNELREQUESTER_H_ */

206
pvAccessApp/server/responseHandlers.cpp
View File

@@ -41,7 +41,7 @@ namespace epics {
ServerResponseHandler::ServerResponseHandler(ServerContextImpl* context) {
BadResponse* badResponse = new BadResponse(context);
_badResponse = new BadResponse(context);
_handlerTable = new ResponseHandler*[HANDLER_TABLE_LENGTH];
// TODO add real handlers, as they are developed
@@ -49,33 +49,34 @@ namespace epics {
_handlerTable[1] = new ConnectionValidationHandler(context);
_handlerTable[2] = new EchoHandler(context);
_handlerTable[3] = new SearchHandler(context);
_handlerTable[4] = badResponse;
_handlerTable[4] = _badResponse;
_handlerTable[5] = new IntrospectionSearchHandler(context);
_handlerTable[6] = badResponse;
_handlerTable[6] = _badResponse;
_handlerTable[7] = new CreateChannelHandler(context);
_handlerTable[8] = new DestroyChannelHandler(context);
_handlerTable[9] = badResponse;
_handlerTable[10] = badResponse;
_handlerTable[11] = badResponse;
_handlerTable[12] = badResponse;
_handlerTable[13] = badResponse;
_handlerTable[14] = badResponse;
_handlerTable[15] = badResponse;
_handlerTable[16] = badResponse;
_handlerTable[17] = badResponse;
_handlerTable[18] = badResponse;
_handlerTable[19] = badResponse;
_handlerTable[20] = badResponse;
_handlerTable[21] = badResponse;
_handlerTable[22] = badResponse;
_handlerTable[23] = badResponse;
_handlerTable[24] = badResponse;
_handlerTable[25] = badResponse;
_handlerTable[26] = badResponse;
_handlerTable[27] = badResponse;
_handlerTable[9] = _badResponse;
//_handlerTable[10] = new GetHandler(context);
_handlerTable[11] = _badResponse;
_handlerTable[12] = _badResponse;
_handlerTable[13] = _badResponse;
_handlerTable[14] = _badResponse;
_handlerTable[15] = _badResponse;
_handlerTable[16] = _badResponse;
_handlerTable[17] = _badResponse;
_handlerTable[18] = _badResponse;
_handlerTable[19] = _badResponse;
_handlerTable[20] = _badResponse;
_handlerTable[21] = _badResponse;
_handlerTable[22] = _badResponse;
_handlerTable[23] = _badResponse;
_handlerTable[24] = _badResponse;
_handlerTable[25] = _badResponse;
_handlerTable[26] = _badResponse;
_handlerTable[27] = _badResponse;
}
ServerResponseHandler::~ServerResponseHandler() {
delete _badResponse;
delete _handlerTable[0];
delete _handlerTable[1];
delete _handlerTable[2];
@@ -83,6 +84,7 @@ namespace epics {
delete _handlerTable[5];
delete _handlerTable[7];
delete _handlerTable[8];
delete _handlerTable[10];
delete _handlerTable[27];
delete[] _handlerTable;
}
@@ -329,7 +331,7 @@ namespace epics {
}
void ChannelRequesterImpl::channelCreated(const Status& status, Channel* const channel)
void ChannelRequesterImpl::channelCreated(const Status& status, Channel* channel)
{
Lock guard(_mutex);
_status = status;
@@ -337,7 +339,7 @@ namespace epics {
_transport->enqueueSendRequest(this);
}
void ChannelRequesterImpl::channelStateChange(Channel* constc, const Channel::ConnectionState isConnected)
void ChannelRequesterImpl::channelStateChange(Channel* c, const Channel::ConnectionState isConnected)
{
//noop
}
@@ -481,7 +483,161 @@ namespace epics {
casTransport->unregisterChannel(sid);
// send response back
transport->enqueueSendRequest( new DestroyChannelHandlerTransportSender(cid, sid));
transport->enqueueSendRequest(new DestroyChannelHandlerTransportSender(cid, sid));
}
/****************************************************************************************/
void GetHandler::handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer) {
AbstractServerResponseHandler::handleResponse(responseFrom,
transport, version, command, payloadSize, payloadBuffer);
// NOTE: we do not explicitly check if transport is OK
ChannelHostingTransport* casTransport = dynamic_cast<ChannelHostingTransport*>(transport);
transport->ensureData(2*sizeof(int32)/sizeof(int8)+1);
const pvAccessID sid = payloadBuffer->getInt();
const pvAccessID ioid = payloadBuffer->getInt();
// mode
const int8 qosCode = payloadBuffer->getByte();
ServerChannelImpl* channel = static_cast<ServerChannelImpl*>(casTransport->getChannel(sid));
if (channel == NULL)
{
BaseChannelRequester::sendFailureMessage((int8)10, transport, ioid, qosCode, BaseChannelRequester::badCIDStatus);
return;
}
const boolean init = (QOS_INIT & qosCode) != 0;
if (init)
{
// pvRequest
//PVStructurePtr pvRequest = transport->getIntrospectionRegistry()->deserializePVRequest(payloadBuffer, transport);
// create...
// new ChannelGetRequesterImpl(_context, channel, ioid, transport, pvRequest);
}
/* else
{
final boolean lastRequest = QoS.DESTROY.isSet(qosCode);
ChannelGetRequesterImpl request = (ChannelGetRequesterImpl)channel.getRequest(ioid);
if (request == null) {
BaseChannelRequester.sendFailureMessage((byte)10, transport, ioid, qosCode, BaseChannelRequester.badIOIDStatus);
return;
}
if (!request.startRequest(qosCode)) {
BaseChannelRequester.sendFailureMessage((byte)10, transport, ioid, qosCode, BaseChannelRequester.otherRequestPendingStatus);
return;
}
request.getChannelGet().get(lastRequest);
}
*/
}
ChannelGetRequesterImpl::ChannelGetRequesterImpl(ServerContextImpl* context, ServerChannelImpl* channel, const pvAccessID ioid, Transport* transport,
epics::pvData::PVStructurePtr pvRequest) :
BaseChannelRequester(context, channel, ioid, transport)
{
startRequest(QOS_INIT);
channel->registerRequest(ioid, this);
_channelGet = channel->getChannel()->createChannelGet(this, pvRequest);
// TODO what if last call fails... registration is still present
}
void ChannelGetRequesterImpl::channelGetConnect(const epics::pvData::Status& status, ChannelGet* channelGet, epics::pvData::PVStructurePtr pvStructure,
epics::pvData::BitSet* bitSet)
{
{
Lock guard(_mutex);
_bitSet = bitSet;
_pvStructure = pvStructure;
_status = status;
_channelGet = channelGet;
}
_transport->enqueueSendRequest(this);
// self-destruction
if (!status.isSuccess())
{
destroy();
}
}
void ChannelGetRequesterImpl::getDone(const epics::pvData::Status& status)
{
{
Lock guard(_mutex);
_status = status;
}
_transport->enqueueSendRequest(this);
}
void ChannelGetRequesterImpl::destroy()
{
_channel->unregisterRequest(_ioid);
if (_channelGet != NULL)
{
_channelGet->destroy();
}
}
ChannelGet* ChannelGetRequesterImpl::getChannelGet()
{
return _channelGet;
}
void ChannelGetRequesterImpl::lock()
{
//TODO
}
void ChannelGetRequesterImpl::unlock()
{
//TODO
}
void ChannelGetRequesterImpl::send(ByteBuffer* buffer, TransportSendControl* control)
{
const int32 request = getPendingRequest();
control->startMessage((int8)10, sizeof(int32)/sizeof(int8) + 1);
buffer->putInt(_ioid);
buffer->put((int8)request);
IntrospectionRegistry* introspectionRegistry = _transport->getIntrospectionRegistry();
{
Lock guard(_mutex);
introspectionRegistry->serializeStatus(buffer, control, _status);
}
if (_status.isSuccess())
{
if (request & QOS_INIT)
{
Lock guard(_mutex);
introspectionRegistry->serialize(_pvStructure != NULL ? _pvStructure->getField() : NULL, buffer, control);
}
else
{
_bitSet->serialize(buffer, control);
_pvStructure->serialize(buffer, control, _bitSet);
}
}
stopRequest();
// lastRequest
if (request & QOS_DESTROY)
{
destroy();
}
}
}
}

54
pvAccessApp/server/responseHandlers.h
View File

@@ -11,6 +11,7 @@
#include "serverContext.h"
#include "remote.h"
#include "serverChannelImpl.h"
#include "baseChannelRequester.h"
namespace epics {
namespace pvAccess {
@@ -73,6 +74,10 @@ namespace epics {
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
private:
static const int HANDLER_TABLE_LENGTH = 28;
/**
* Bad response handlers.
*/
BadResponse *_badResponse;
/**
* Table of response handlers for each command ID.
*/
@@ -226,6 +231,7 @@ namespace epics {
AbstractServerResponseHandler(context, "Create channel request") {
}
//TODO where is implementation???
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
@@ -237,9 +243,9 @@ namespace epics {
class ChannelRequesterImpl : public ChannelRequester, public TransportSender
{
public:
ChannelRequesterImpl(Transport* transport, const String channelName, const int32 cid);
void channelCreated(const Status& status, Channel* const channel);
void channelStateChange(Channel* const c, const Channel::ConnectionState isConnected);
ChannelRequesterImpl(Transport* transport, const String channelName, const pvAccessID cid);
void channelCreated(const 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();
@@ -310,6 +316,48 @@ namespace epics {
pvAccessID _sid;
};
/****************************************************************************************/
/**
* Get request handler.
*/
class GetHandler : public AbstractServerResponseHandler
{
public:
/**
* @param context
*/
GetHandler(ServerContextImpl* context) :
AbstractServerResponseHandler(context, "Get request") {
}
virtual void handleResponse(osiSockAddr* responseFrom,
Transport* transport, int8 version, int8 command,
int payloadSize, epics::pvData::ByteBuffer* payloadBuffer);
};
class ChannelGetRequesterImpl : private BaseChannelRequester, public ChannelGetRequester, public TransportSender
{
public:
ChannelGetRequesterImpl(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);
void getDone(const epics::pvData::Status& status);
void destroy();
/**
* @return the channelGet
*/
ChannelGet* getChannelGet();
void lock();
void unlock();
void send(ByteBuffer* buffer, TransportSendControl* control);
private:
ChannelGet* _channelGet;
epics::pvData::BitSet* _bitSet;
epics::pvData::PVStructurePtr _pvStructure;
epics::pvData::Status _status;
epics::pvData::Mutex _mutex;
};
}
}

2
pvAccessApp/server/serverChannelImpl.cpp
View File

@@ -40,7 +40,7 @@ int16 ServerChannelImpl::getAccessRights()
return 0;
}
void ServerChannelImpl::registerRequest(const pvAccessID id, Destroyable* const request)
void ServerChannelImpl::registerRequest(const pvAccessID id, Destroyable* request)
{
if (request == NULL)
{

2
pvAccessApp/server/serverChannelImpl.h
View File

@@ -59,7 +59,7 @@ public:
* @param id request ID.
* @param request request to be registered.
*/
void registerRequest(pvAccessID id, Destroyable* const request);
void registerRequest(pvAccessID id, Destroyable* request);
/**
* Unregister request.

6
pvAccessApp/server/serverContext.h
View File

@@ -327,7 +327,7 @@ private:
/**
* Timer.
*/
Timer* _timer;
epics::pvData::Timer* _timer;
/**
* Broadcast transport needed for channel searches.
@@ -373,12 +373,12 @@ private:
/**
* Run mutex.
*/
Mutex _mutex;
epics::pvData::Mutex _mutex;
/**
* Run event.
*/
Event _runEvent;
epics::pvData::Event _runEvent;
/**
* Beacon server status provider interface (optional).