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

remote.h interfaces sync with Java - cleanup

Browse Source
This commit is contained in:
Matej Sekoranja
2012-07-25 12:46:04 +02:00
parent 48f8b3b77e
commit 4f15626281
18 changed files with 202 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
pvAccessApp/remote/beaconHandler.cpp
View File

@@ -14,7 +14,7 @@ using namespace epics::pvAccess;
namespace epics {
namespace pvAccess {
BeaconHandler::BeaconHandler(Context::shared_pointer context,
BeaconHandler::BeaconHandler(Context::shared_pointer const & context,
const osiSockAddr* responseFrom) :
_context(Context::weak_pointer(context)),
_responseFrom(*responseFrom),
@@ -42,10 +42,8 @@ void BeaconHandler::beaconNotify(osiSockAddr* from, int8 remoteTransportRevision
PVFieldPtr data)
{
bool networkChanged = updateBeacon(remoteTransportRevision, timestamp, startupTime, sequentalID);
if(networkChanged)
{
if (networkChanged)
changedTransport();
}
}
bool BeaconHandler::updateBeacon(int8 remoteTransportRevision, TimeStamp* timestamp,
@@ -58,10 +56,7 @@ bool BeaconHandler::updateBeacon(int8 remoteTransportRevision, TimeStamp* timest
_serverStartupTime = *startupTime;
// new server up..
_context.lock()->beaconAnomalyNotify();
// notify corresponding transport(s)
beaconArrivalNotify();
_context.lock()->newServerDetected();
return false;
}
@@ -69,30 +64,15 @@ bool BeaconHandler::updateBeacon(int8 remoteTransportRevision, TimeStamp* timest
bool networkChange = !(_serverStartupTime == *startupTime);
if (networkChange)
{
_context.lock()->beaconAnomalyNotify();
}
else
{
beaconArrivalNotify();
// update startup time
_serverStartupTime = *startupTime;
_context.lock()->newServerDetected();
return true;
}
return networkChange;
}
void BeaconHandler::beaconArrivalNotify()
{
auto_ptr<TransportRegistry::transportVector_t> transports =
_context.lock()->getTransportRegistry()->get("TCP", &_responseFrom);
if (!transports.get())
return;
// notify all
for (TransportRegistry::transportVector_t::iterator iter = transports->begin();
iter != transports->end();
iter++)
{
(*iter)->aliveNotification();
}
return false;
}
void BeaconHandler::changedTransport()

9
pvAccessApp/remote/beaconHandler.h
View File

@@ -23,15 +23,14 @@ namespace pvAccess {
class BeaconHandler
{
public:
typedef std::tr1::shared_ptr<BeaconHandler> shared_pointer;
typedef std::tr1::shared_ptr<const BeaconHandler> const_shared_pointer;
POINTER_DEFINITIONS(BeaconHandler);
/**
* Constructor.
* @param transport transport to be used to send beacons.
* @param context CA context.
*/
BeaconHandler(Context::shared_pointer context, const osiSockAddr* responseFrom);
BeaconHandler(Context::shared_pointer const & context, const osiSockAddr* responseFrom);
/**
* Test Constructor (for testing)
* @param transport transport to be used to send beacons.
@@ -81,10 +80,6 @@ namespace pvAccess {
epics::pvData::TimeStamp* timestamp,
epics::pvData::TimeStamp* startupTime,
epics::pvData::int16 sequentalID);
/**
* Notify transport about beacon arrival.
*/
void beaconArrivalNotify();
/**
* Changed transport (server restarted) notify.
*/

2
pvAccessApp/remote/blockingClientTCPTransport.cpp
View File

@@ -165,7 +165,7 @@ namespace epics {
// not used anymore, close it
// TODO consider delayed destruction (can improve performance!!!)
if(_owners.size()==0) close(false);
if(_owners.size()==0) close(); // TODO close(false)
}
void BlockingClientTCPTransport::aliveNotification() {

54
pvAccessApp/remote/blockingTCP.h
View File

@@ -57,19 +57,23 @@ namespace epics {
return _closed.get();
}
virtual void setRemoteMinorRevision(epics::pvData::int8 minorRevision) {
_remoteTransportRevision = minorRevision;
virtual epics::pvData::int8 getRevision() const {
return CA_PROTOCOL_REVISION;
}
virtual void setRemoteTransportReceiveBufferSize(int remoteTransportReceiveBufferSize) {
virtual void setRemoteRevision(epics::pvData::int8 revision) {
_remoteTransportRevision = revision;
}
virtual void setRemoteTransportReceiveBufferSize(std::size_t remoteTransportReceiveBufferSize) {
_remoteTransportReceiveBufferSize = remoteTransportReceiveBufferSize;
}
virtual void setRemoteTransportSocketReceiveBufferSize(int socketReceiveBufferSize) {
virtual void setRemoteTransportSocketReceiveBufferSize(std::size_t socketReceiveBufferSize) {
_remoteTransportSocketReceiveBufferSize = socketReceiveBufferSize;
}
virtual const epics::pvData::String getType() const {
virtual epics::pvData::String getType() const {
return epics::pvData::String("TCP");
}
@@ -89,7 +93,7 @@ namespace epics {
return _priority;
}
virtual int getReceiveBufferSize() const {
virtual std::size_t getReceiveBufferSize() const {
return _socketBuffer->getSize();
}
@@ -97,15 +101,16 @@ namespace epics {
* Get remote transport receive buffer size (in bytes).
* @return remote transport receive buffer size
*/
int getRemoteTransportReceiveBufferSize() {
virtual std::size_t getRemoteTransportReceiveBufferSize() const {
return _remoteTransportReceiveBufferSize;
}
virtual int getSocketReceiveBufferSize() const;
virtual std::size_t getSocketReceiveBufferSize() const;
virtual bool isVerified() {
virtual bool verify(epics::pvData::int32 timeoutMs) {
epics::pvData::Lock lock(_verifiedMutex);
return _verified;
// TODO !!!
}
virtual void verified() {
@@ -139,7 +144,12 @@ namespace epics {
virtual void alignData(std::size_t alignment);
virtual void close(bool force);
virtual void close();
virtual void setByteOrder(int byteOrder)
{
// TODO !!!
}
SendQueueFlushStrategy getSendQueueFlushStrategy() {
return _flushStrategy;
@@ -459,12 +469,10 @@ namespace epics {
class BlockingClientTCPTransport : public BlockingTCPTransport,
public TransportSender,
public epics::pvData::TimerCallback,
public ReferenceCountingTransport {
public epics::pvData::TimerCallback {
public:
typedef std::tr1::shared_ptr<BlockingClientTCPTransport> shared_pointer;
typedef std::tr1::shared_ptr<const BlockingClientTCPTransport> const_shared_pointer;
POINTER_DEFINITIONS(BlockingClientTCPTransport);
private:
BlockingClientTCPTransport(Context::shared_pointer const & context, SOCKET channel,
@@ -596,6 +604,8 @@ namespace epics {
*/
class BlockingTCPConnector : public Connector {
public:
POINTER_DEFINITIONS(BlockingTCPConnector);
BlockingTCPConnector(Context::shared_pointer const & context, int receiveBufferSize,
float beaconInterval);
@@ -645,8 +655,7 @@ namespace epics {
public ChannelHostingTransport,
public TransportSender {
public:
typedef std::tr1::shared_ptr<BlockingServerTCPTransport> shared_pointer;
typedef std::tr1::shared_ptr<const BlockingServerTCPTransport> const_shared_pointer;
POINTER_DEFINITIONS(BlockingServerTCPTransport);
private:
BlockingServerTCPTransport(Context::shared_pointer const & context, SOCKET channel,
@@ -662,6 +671,13 @@ namespace epics {
return thisPointer;
}
virtual bool acquire(std::tr1::shared_ptr<TransportClient> const & client)
{
return false;
}
virtual void release(pvAccessID clientId) {}
/**
* Preallocate new channel SID.
* @return new channel server id (SID).
@@ -783,8 +799,7 @@ namespace epics {
class ResponseHandlerFactory
{
public:
typedef std::tr1::shared_ptr<ResponseHandlerFactory> shared_pointer;
typedef std::tr1::shared_ptr<const ResponseHandlerFactory> const_shared_pointer;
POINTER_DEFINITIONS(ResponseHandlerFactory);
virtual ~ResponseHandlerFactory() {};
@@ -798,8 +813,7 @@ namespace epics {
*/
class BlockingTCPAcceptor {
public:
typedef std::tr1::shared_ptr<BlockingTCPAcceptor> shared_pointer;
typedef std::tr1::shared_ptr<const BlockingTCPAcceptor> const_shared_pointer;
POINTER_DEFINITIONS(BlockingTCPAcceptor);
/**
* @param context

2
pvAccessApp/remote/blockingTCPAcceptor.cpp
View File

@@ -195,7 +195,7 @@ namespace pvAccess {
// validate connection
if(!validateConnection(transport, ipAddrStr)) {
transport->close(true);
transport->close();
LOG(
logLevelDebug,
"Connection to CA client %s failed to be validated, closing it.",

4
pvAccessApp/remote/blockingTCPConnector.cpp
View File

@@ -163,13 +163,13 @@ namespace epics {
return transport;
} catch(std::exception& ex) {
if(transport.get())
transport->close(true);
transport->close();
else if(socket!=INVALID_SOCKET) epicsSocketDestroy(socket);
_namedLocker.releaseSynchronizationObject(&address);
throw;
} catch(...) {
if(transport.get())
transport->close(true);
transport->close();
else if(socket!=INVALID_SOCKET) epicsSocketDestroy(socket);
_namedLocker.releaseSynchronizationObject(&address);
throw;

21
pvAccessApp/remote/blockingTCPTransport.cpp
View File

@@ -182,7 +182,7 @@ namespace pvAccess {
BlockingTCPTransport::~BlockingTCPTransport() {
PVACCESS_REFCOUNT_MONITOR_DESTRUCT(blockingTCPTransport);
close(true);
close();
// TODO use auto_ptr class members
@@ -248,7 +248,7 @@ namespace pvAccess {
_sendBuffer->putInt(0);
}
void BlockingTCPTransport::close(bool force) {
void BlockingTCPTransport::close() {
Lock lock(_mutex);
// already closed check
@@ -259,6 +259,9 @@ namespace pvAccess {
Transport::shared_pointer thisSharedPtr = shared_from_this();
_context->getTransportRegistry()->remove(thisSharedPtr).get();
// TODO !!!
bool force = true;
// clean resources
internalClose(force);
@@ -281,7 +284,7 @@ namespace pvAccess {
void BlockingTCPTransport::internalPostClose(bool force) {
}
int BlockingTCPTransport::getSocketReceiveBufferSize() const {
size_t BlockingTCPTransport::getSocketReceiveBufferSize() const {
// Get value of the SO_RCVBUF option for this DatagramSocket,
// that is the buffer size used by the platform for input on
// this DatagramSocket.
@@ -299,7 +302,7 @@ namespace pvAccess {
errStr);
}
return sockBufSize;
return (size_t)sockBufSize;
}
bool BlockingTCPTransport::waitUntilVerified(double timeout) {
@@ -542,7 +545,7 @@ namespace pvAccess {
}
// error (disconnect, end-of-stream) detected
close(true);
close();
if(nestedCall)
THROW_BASE_EXCEPTION("bytesRead < 0");
@@ -586,7 +589,7 @@ namespace pvAccess {
logLevelError,
"Invalid header received from client %s, disconnecting...",
inetAddressToString(_socketAddress).c_str());
close(true);
close();
return;
}
@@ -658,7 +661,7 @@ namespace pvAccess {
"Unknown packet type %d, received from client %s, disconnecting...",
type,
inetAddressToString(_socketAddress).c_str());
close(true);
close();
return;
}
}
@@ -707,7 +710,7 @@ namespace pvAccess {
}
} catch(...) {
// close connection
close(true);
close();
if(nestedCall) throw;
}
@@ -860,7 +863,7 @@ namespace pvAccess {
// buffer->getPosition(), limit);
} // while
} catch(...) {
close(true);
close();
throw;
}

40
pvAccessApp/remote/blockingUDP.h
View File

@@ -31,8 +31,7 @@ namespace epics {
public std::tr1::enable_shared_from_this<BlockingUDPTransport>
{
public:
typedef std::tr1::shared_ptr<BlockingUDPTransport> shared_pointer;
typedef std::tr1::shared_ptr<const BlockingUDPTransport> const_shared_pointer;
POINTER_DEFINITIONS(BlockingUDPTransport);
private:
BlockingUDPTransport(std::auto_ptr<ResponseHandler>& responseHandler,
@@ -60,31 +59,35 @@ namespace epics {
return &_bindAddress;
}
virtual const epics::pvData::String getType() const {
virtual epics::pvData::String getType() const {
return epics::pvData::String("UDP");
}
virtual int getReceiveBufferSize() const {
virtual std::size_t getReceiveBufferSize() const {
return _receiveBuffer->getSize();
}
virtual int getSocketReceiveBufferSize() const;
virtual std::size_t getSocketReceiveBufferSize() const;
virtual epics::pvData::int16 getPriority() const {
return CA_DEFAULT_PRIORITY;
}
virtual void setRemoteMinorRevision(epics::pvData::int8 minor) {
virtual epics::pvData::int8 getRevision() const {
return CA_PROTOCOL_REVISION;
}
virtual void setRemoteRevision(epics::pvData::int8 revision) {
// noop
}
virtual void setRemoteTransportReceiveBufferSize(
int receiveBufferSize) {
std::size_t receiveBufferSize) {
// noop for UDP (limited by 64k; MAX_UDP_SEND for CA)
}
virtual void setRemoteTransportSocketReceiveBufferSize(
int socketReceiveBufferSize) {
std::size_t socketReceiveBufferSize) {
// noop for UDP (limited by 64k; MAX_UDP_SEND for CA)
}
@@ -96,19 +99,24 @@ namespace epics {
// noop
}
virtual bool isVerified() {
return false;
virtual bool verify(epics::pvData::int32 timeoutMs) {
// noop
return true;
}
virtual void verified() {
// noop
}
virtual void setByteOrder(int byteOrder) {
// TODO
}
virtual void enqueueSendRequest(TransportSender::shared_pointer const & sender);
void start();
virtual void close(bool forced);
virtual void close();
virtual void ensureData(std::size_t size) {
// noop
@@ -157,6 +165,13 @@ namespace epics {
return epics::pvData::getFieldCreate()->deserialize(buffer, this);
}
virtual bool acquire(std::tr1::shared_ptr<TransportClient> const & client)
{
return false;
}
virtual void release(pvAccessID clientId) {}
/**
* Set ignore list.
* @param addresses list of ignored addresses.
@@ -232,7 +247,7 @@ namespace epics {
bool processBuffer(Transport::shared_pointer const & transport, osiSockAddr& fromAddress, epics::pvData::ByteBuffer* receiveBuffer);
void close(bool forced, bool waitForThreadToComplete);
void close(bool waitForThreadToComplete);
// Context only used for logging in this class
@@ -295,6 +310,7 @@ namespace epics {
public Connector,
private epics::pvData::NoDefaultMethods {
public:
POINTER_DEFINITIONS(BlockingUDPConnector);
BlockingUDPConnector(
bool reuseSocket,

14
pvAccessApp/remote/blockingUDPTransport.cpp
View File

@@ -89,11 +89,11 @@ namespace epics {
BlockingUDPTransport::threadRunner, this);
}
void BlockingUDPTransport::close(bool forced) {
close(forced, true);
void BlockingUDPTransport::close() {
close(true);
}
void BlockingUDPTransport::close(bool forced, bool waitForThreadToComplete) {
void BlockingUDPTransport::close(bool waitForThreadToComplete) {
{
Lock guard(_mutex);
if(_closed.get()) return;
@@ -226,14 +226,14 @@ namespace epics {
LOG(logLevelError, "Socket recvfrom error: %s", errStr);
}
close(true, false);
close(false);
break;
}
}
} catch(...) {
// TODO: catch all exceptions, and act accordingly
close(true, false);
close(false);
}
String threadName = "UDP-receive "+inetAddressToString(_bindAddress);
@@ -333,7 +333,7 @@ namespace epics {
return allOK;
}
int BlockingUDPTransport::getSocketReceiveBufferSize() const {
size_t BlockingUDPTransport::getSocketReceiveBufferSize() const {
// Get value of the SO_RCVBUF option for this DatagramSocket,
// that is the buffer size used by the platform for input on
// this DatagramSocket.
@@ -349,7 +349,7 @@ namespace epics {
LOG(logLevelError, "Socket getsockopt SO_RCVBUF error: %s", errStr);
}
return sockBufSize;
return (size_t)sockBufSize;
}
void BlockingUDPTransport::threadRunner(void* param) {

196
pvAccessApp/remote/remote.h
View File

@@ -109,8 +109,7 @@ namespace epics {
*/
class TransportSendControl : public epics::pvData::SerializableControl {
public:
typedef std::tr1::shared_ptr<TransportSendControl> shared_pointer;
typedef std::tr1::shared_ptr<const TransportSendControl> const_shared_pointer;
POINTER_DEFINITIONS(TransportSendControl);
virtual ~TransportSendControl() {}
@@ -119,7 +118,7 @@ namespace epics {
virtual void flush(bool lastMessageCompleted) = 0;
virtual void setRecipient(const osiSockAddr& sendTo) = 0;
virtual void setRecipient(osiSockAddr const & sendTo) = 0;
};
/**
@@ -127,8 +126,7 @@ namespace epics {
*/
class TransportSender : public Lockable {
public:
typedef std::tr1::shared_ptr<TransportSender> shared_pointer;
typedef std::tr1::shared_ptr<const TransportSender> const_shared_pointer;
POINTER_DEFINITIONS(TransportSender);
virtual ~TransportSender() {}
@@ -138,54 +136,68 @@ namespace epics {
* Calls on <code>TransportSendControl</code> instance must be made from
* calling thread. Moreover, ownership is valid only for the time of call
* of this method.
* NOTE: these limitations allows efficient implementation.
* NOTE: these limitations allow efficient implementation.
*/
virtual void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) = 0;
};
class TransportClient;
/**
* Interface defining transport (connection).
*/
class Transport : public epics::pvData::DeserializableControl {
public:
typedef std::tr1::shared_ptr<Transport> shared_pointer;
typedef std::tr1::shared_ptr<const Transport> const_shared_pointer;
typedef std::tr1::weak_ptr<Transport> weak_pointer;
typedef std::tr1::weak_ptr<const Transport> const_weak_pointer;
POINTER_DEFINITIONS(Transport);
virtual ~Transport() {}
/**
* Acquires transport.
* @param client client (channel) acquiring the transport
* @return <code>true</code> if transport was granted, <code>false</code> otherwise.
*/
//virtual bool acquire(TransportClient::shared_pointer const & client) = 0;
virtual bool acquire(std::tr1::shared_ptr<TransportClient> const & client) = 0;
/**
* Releases transport.
* @param client client (channel) releasing the transport
*/
virtual void release(pvAccessID clientId) = 0;
//virtual void release(TransportClient::shared_pointer const & client) = 0;
/**
* Get protocol type (tcp, udp, ssl, etc.).
* @return protocol type.
*/
virtual epics::pvData::String getType() const = 0;
/**
* Get remote address.
* @return remote address, can be null.
*/
virtual const osiSockAddr* getRemoteAddress() const = 0;
/**
* Get protocol type (tcp, udp, ssl, etc.).
* @return protocol type.
*/
virtual const epics::pvData::String getType() const = 0;
// TODO getContext?
/**
* Transport protocol minor revision.
* @return protocol minor revision.
*/
virtual epics::pvData::int8 getRevision() const {
return CA_PROTOCOL_REVISION;
}
virtual epics::pvData::int8 getRevision() const = 0;
/**
* Get receive buffer size.
* @return receive buffer size.
*/
virtual int getReceiveBufferSize() const = 0;
virtual std::size_t getReceiveBufferSize() const = 0;
/**
* Get socket receive buffer size.
* @return socket receive buffer size.
*/
virtual int getSocketReceiveBufferSize() const = 0;
virtual std::size_t getSocketReceiveBufferSize() const = 0;
/**
* Transport priority.
@@ -194,27 +206,29 @@ namespace epics {
virtual epics::pvData::int16 getPriority() const = 0;
/**
* Set remote transport protocol minor revision.
* @param minor protocol minor revision.
* Set remote transport protocol revision.
* @param revision protocol revision.
*/
virtual void setRemoteMinorRevision(epics::pvData::int8 minor) = 0;
virtual void setRemoteRevision(epics::pvData::int8 revision) = 0;
/**
* Set remote transport receive buffer size.
* @param receiveBufferSize receive buffer size.
*/
virtual void setRemoteTransportReceiveBufferSize(int receiveBufferSize) = 0;
virtual void setRemoteTransportReceiveBufferSize(std::size_t receiveBufferSize) = 0;
/**
* Set remote transport socket receive buffer size.
* @param socketReceiveBufferSize remote socket receive buffer size.
*/
virtual void setRemoteTransportSocketReceiveBufferSize(int socketReceiveBufferSize) = 0;
virtual void setRemoteTransportSocketReceiveBufferSize(std::size_t socketReceiveBufferSize) = 0;
/**
* Notification transport that is still alive.
*/
virtual void aliveNotification() = 0;
/**
* Set byte order.
* @param byteOrder byte order to set.
*/
// TODO enum
virtual void setByteOrder(int byteOrder) = 0;
/**
* Notification that transport has changed.
@@ -222,38 +236,37 @@ namespace epics {
virtual void changedTransport() = 0;
/**
* Close transport.
* @param force flag indicating force-full (e.g. remote disconnect) close.
* Enqueue send request.
* @param sender
*/
virtual void close(bool force) = 0;
/**
* Check connection status.
* @return <code>true</code> if connected.
*/
virtual bool isClosed() = 0;
/**
* Get transport verification status.
* @return verification flag.
*/
virtual bool isVerified() = 0;
virtual void enqueueSendRequest(TransportSender::shared_pointer const & sender) = 0;
/**
* Notify transport that it is has been verified.
*/
virtual void verified() = 0;
/**
* Waits (if needed) until transport is verified, i.e. verified() method is being called.
* @param timeoutMs timeout to wait for verification, infinite if 0.
*/
virtual bool verify(epics::pvData::int32 timeoutMs) = 0;
/**
* Enqueue send request.
* @param sender
* Notification transport that is still alive.
*/
virtual void enqueueSendRequest(TransportSender::shared_pointer const & sender) = 0;
virtual void aliveNotification() = 0;
virtual void enqueueOnlySendRequest(TransportSender::shared_pointer const & sender) {};
virtual void flushSendQueue() {};
/**
* Close transport.
*/
virtual void close() = 0;
/**
* Check connection status.
* @return <code>true</code> if connected.
*/
virtual bool isClosed() = 0;
};
class Channel;
@@ -263,27 +276,30 @@ namespace epics {
*/
class Context {
public:
typedef std::tr1::shared_ptr<Context> shared_pointer;
typedef std::tr1::shared_ptr<const Context> const_shared_pointer;
typedef std::tr1::weak_ptr<Context> weak_pointer;
typedef std::tr1::weak_ptr<const Context> const_weak_pointer;
POINTER_DEFINITIONS(Context);
virtual ~Context() {}
virtual std::tr1::shared_ptr<Channel> getChannel(pvAccessID id) = 0;
virtual Transport::shared_pointer getSearchTransport() = 0;
virtual epics::pvData::Timer::shared_pointer getTimer() = 0;
//virtual TransportRegistry::shared_pointer getTransportRegistry() = 0;
virtual std::tr1::shared_ptr<TransportRegistry> getTransportRegistry() = 0;
virtual Configuration::shared_pointer getConfiguration() = 0;
virtual void beaconAnomalyNotify() = 0;
///
/// due to ClientContextImpl
///
virtual void newServerDetected() = 0;
virtual std::tr1::shared_ptr<Channel> getChannel(pvAccessID id) = 0;
virtual Transport::shared_pointer getSearchTransport() = 0;
};
/**
@@ -291,8 +307,7 @@ namespace epics {
*/
class ResponseHandler {
public:
typedef std::tr1::shared_ptr<ResponseHandler> shared_pointer;
typedef std::tr1::shared_ptr<const ResponseHandler> const_shared_pointer;
POINTER_DEFINITIONS(ResponseHandler);
virtual ~ResponseHandler() {}
@@ -348,10 +363,7 @@ namespace epics {
*/
class TransportClient {
public:
typedef std::tr1::shared_ptr<TransportClient> shared_pointer;
typedef std::tr1::shared_ptr<const TransportClient> const_shared_pointer;
typedef std::tr1::weak_ptr<TransportClient> weak_pointer;
typedef std::tr1::weak_ptr<const TransportClient> const_weak_pointer;
POINTER_DEFINITIONS(TransportClient);
virtual ~TransportClient() {
}
@@ -405,35 +417,9 @@ namespace epics {
};
/**
* Interface defining reference counting transport IF.
*/
class ReferenceCountingTransport {
public:
virtual ~ReferenceCountingTransport() {
}
/**
* Acquires transport.
* @param client client (channel) acquiring the transport
* @return <code>true</code> if transport was granted, <code>false</code> otherwise.
*/
virtual bool acquire(TransportClient::shared_pointer const & client) = 0;
/**
* Releases transport.
* @param client client (channel) releasing the transport
*/
virtual void release(pvAccessID clientId) = 0;
//virtual void release(TransportClient::shared_pointer const & client) = 0;
};
class ServerChannel {
public:
typedef std::tr1::shared_ptr<ServerChannel> shared_pointer;
typedef std::tr1::shared_ptr<const ServerChannel> const_shared_pointer;
typedef std::tr1::weak_ptr<ServerChannel> weak_pointer;
typedef std::tr1::weak_ptr<const ServerChannel> const_weak_pointer;
POINTER_DEFINITIONS(ServerChannel);
virtual ~ServerChannel() {}
/**
@@ -454,8 +440,7 @@ namespace epics {
*/
class ChannelHostingTransport {
public:
typedef std::tr1::shared_ptr<ChannelHostingTransport> shared_pointer;
typedef std::tr1::shared_ptr<const ChannelHostingTransport> const_shared_pointer;
POINTER_DEFINITIONS(ChannelHostingTransport);
virtual ~ChannelHostingTransport() {}
@@ -510,10 +495,7 @@ namespace epics {
*/
class ResponseRequest {
public:
typedef std::tr1::shared_ptr<ResponseRequest> shared_pointer;
typedef std::tr1::shared_ptr<const ResponseRequest> const_shared_pointer;
typedef std::tr1::weak_ptr<ResponseRequest> weak_pointer;
typedef std::tr1::weak_ptr<const ResponseRequest> const_weak_pointer;
POINTER_DEFINITIONS(ResponseRequest);
virtual ~ResponseRequest() {}
@@ -537,7 +519,7 @@ namespace epics {
* Report status to clients (e.g. disconnected).
* @param status to report.
*/
virtual void reportStatus(const epics::pvData::Status& status) = 0;
virtual void reportStatus(epics::pvData::Status const & status) = 0;
/**
* Get request requester.
@@ -547,13 +529,10 @@ namespace epics {
};
/**
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: DataResponse.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class DataResponse : public ResponseRequest {
public:
typedef std::tr1::shared_ptr<DataResponse> shared_pointer;
typedef std::tr1::shared_ptr<const DataResponse> const_shared_pointer;
POINTER_DEFINITIONS(DataResponse);
virtual ~DataResponse() {}
@@ -571,13 +550,10 @@ namespace epics {
* A request that expects an response multiple responses.
* Responses identified by its I/O ID.
* This interface needs to be extended (to provide method called on response).
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
* @version $Id: SubscriptionRequest.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
*/
class SubscriptionRequest /*: public ResponseRequest*/ {
public:
typedef std::tr1::shared_ptr<SubscriptionRequest> shared_pointer;
typedef std::tr1::shared_ptr<const SubscriptionRequest> const_shared_pointer;
POINTER_DEFINITIONS(SubscriptionRequest);
virtual ~SubscriptionRequest() {}

27
pvAccessApp/remoteClient/clientContextImpl.cpp
View File

@@ -2578,7 +2578,7 @@ namespace epics {
transport->setRemoteTransportReceiveBufferSize(payloadBuffer->getInt());
transport->setRemoteTransportSocketReceiveBufferSize(payloadBuffer->getInt());
transport->setRemoteMinorRevision(version);
transport->setRemoteRevision(version);
TransportSender::shared_pointer sender = dynamic_pointer_cast<TransportSender>(transport);
if (sender.get()) {
transport->enqueueSendRequest(sender);
@@ -3167,12 +3167,7 @@ namespace epics {
{
disconnectPendingIO(false);
ReferenceCountingTransport* rct = dynamic_cast<ReferenceCountingTransport*>(m_transport.get());
if (rct)
{
//TransportClient::shared_pointer thisPointer = shared_from_this();
rct->release(getID());
}
m_transport->release(getID());
}
else if (m_transport == transport)
{
@@ -3295,11 +3290,7 @@ namespace epics {
else if (m_transport)
{
// unresponsive state, do not forget to release transport
ReferenceCountingTransport* rct = dynamic_cast<ReferenceCountingTransport*>(m_transport.get());
if (rct) {
//TransportClient::shared_pointer thisPointer = shared_from_this();
rct->release(getID());
}
m_transport->release(getID());
m_transport.reset();
}
@@ -3345,11 +3336,7 @@ namespace epics {
m_transport->enqueueSendRequest(thisSender);
}
ReferenceCountingTransport* rct = dynamic_cast<ReferenceCountingTransport*>(m_transport.get());
if (rct) {
//TransportClient::shared_pointer thisPointer = shared_from_this();
rct->release(getID());
}
m_transport->release(getID());
m_transport.reset();
}
@@ -3979,8 +3966,8 @@ TODO
destroyAllChannels();
// stop UDPs
m_searchTransport->close(true);
m_broadcastTransport->close(true);
m_searchTransport->close();
m_broadcastTransport->close();
}
void destroyAllChannels() {
@@ -4154,7 +4141,7 @@ TODO
/**
* Called each time beacon anomaly is detected.
*/
void beaconAnomalyNotify()
virtual void newServerDetected()
{
if (m_channelSearchManager)
m_channelSearchManager->beaconAnomalyNotify();

6
pvAccessApp/remoteClient/clientContextImpl.h
View File

@@ -87,7 +87,7 @@ namespace epics {
virtual void dispose() = 0;
virtual ChannelSearchManager::shared_pointer getChannelSearchManager() = 0;
virtual ChannelSearchManager::shared_pointer getChannelSearchManager() = 0;
virtual void checkChannelName(epics::pvData::String& name) = 0;
virtual void registerChannel(ChannelImpl::shared_pointer const & channel) = 0;
@@ -103,12 +103,12 @@ namespace epics {
virtual Transport::shared_pointer getTransport(TransportClient::shared_pointer const & client, osiSockAddr* serverAddress, epics::pvData::int8 minorRevision, epics::pvData::int16 priority) = 0;
virtual void beaconAnomalyNotify() = 0;
virtual void newServerDetected() = 0;
virtual std::tr1::shared_ptr<BeaconHandler> getBeaconHandler(osiSockAddr* responseFrom) = 0;
};
extern ClientContextImpl::shared_pointer createClientContextImpl();
}

4
pvAccessApp/server/responseHandlers.cpp
View File

@@ -110,7 +110,7 @@ void ServerConnectionValidationHandler::handleResponse(
payloadBuffer->getInt());
transport->setRemoteTransportSocketReceiveBufferSize(
payloadBuffer->getInt());
transport->setRemoteMinorRevision(version);
transport->setRemoteRevision(version);
// TODO support priority !!!
//transport.setPriority(payloadBuffer.getShort());
}
@@ -319,7 +319,7 @@ void ServerCreateChannelHandler::handleResponse(osiSockAddr* responseFrom,
void ServerCreateChannelHandler::disconnect(Transport::shared_pointer const & transport)
{
transport->close(true);
transport->close();
}
ServerChannelRequesterImpl::ServerChannelRequesterImpl(Transport::shared_pointer const & transport,

6
pvAccessApp/server/serverContext.cpp
View File

@@ -336,7 +336,7 @@ void ServerContextImpl::internalDestroy()
// stop responding to search requests
if (_broadcastTransport != NULL)
{
_broadcastTransport->close(true);
_broadcastTransport->close();
_broadcastTransport.reset();
}
@@ -382,7 +382,7 @@ void ServerContextImpl::destroyAllTransports()
Transport::shared_pointer transport = (*transports)[i];
try
{
transport->close(true);
transport->close();
}
catch (std::exception &e)
{
@@ -553,7 +553,7 @@ Transport::shared_pointer ServerContextImpl::getSearchTransport()
return Transport::shared_pointer();
}
void ServerContextImpl::beaconAnomalyNotify()
void ServerContextImpl::newServerDetected()
{
// not used
}

2
pvAccessApp/server/serverContext.h
View File

@@ -130,7 +130,7 @@ public:
TransportRegistry::shared_pointer getTransportRegistry();
std::auto_ptr<ResponseHandler> createResponseHandler();
void beaconAnomalyNotify();
virtual void newServerDetected();
/**
* Version.

2
testApp/remote/testBlockingTCPClnt.cpp
View File

@@ -58,7 +58,7 @@ public:
}
virtual void acquire() {}
virtual void release() {}
virtual void beaconAnomalyNotify() {}
virtual void newServerDetected() {}
private:
std::tr1::shared_ptr<TransportRegistry> _tr;

2
testApp/remote/testBlockingUDPClnt.cpp
View File

@@ -50,7 +50,7 @@ public:
}
virtual void acquire() {}
virtual void release() {}
virtual void beaconAnomalyNotify() {}
virtual void newServerDetected() {}
};
class DummyResponseHandler : public ResponseHandler {

2
testApp/remote/testBlockingUDPSrv.cpp
View File

@@ -44,7 +44,7 @@ public:
}
virtual void acquire() {}
virtual void release() {}
virtual void beaconAnomalyNotify() {}
virtual void newServerDetected() {}
};
class DummyResponseHandler : public ResponseHandler {