merge

pvAccessApp/Makefile

@@ -43,11 +43,15 @@ INC += blockingUDP.h
 INC += beaconEmitter.h
 INC += beaconServerStatusProvider.h
 INC += beaconHandler.h
+INC += blockingTCP.h
 LIBSRCS += blockingUDPTransport.cpp
 LIBSRCS += blockingUDPConnector.cpp
 LIBSRCS += beaconEmitter.cpp
 LIBSRCS += beaconServerStatusProvider.cpp
 LIBSRCS += beaconHandler.cpp
+LIBSRCS += blockingTCPTransport.cpp
+LIBSRCS += blockingClientTCPTransport.cpp
+LIBSRCS += blockingTCPConnector.cpp
 
 LIBRARY = pvAccess
 pvAccess_LIBS += Com

pvAccessApp/remote/blockingClientTCPTransport.cpp

@@ -0,0 +1,216 @@
+/*
+ * BlockingClientTCPTransport.cpp
+ *
+ *  Created on: Jan 3, 2011
+ *      Author: Miha Vitorovic
+ */
+
+/* pvAccess */
+#include "blockingTCP.h"
+
+#include "introspectionRegistry.h"
+
+/* pvData */
+#include <lock.h>
+
+/* EPICSv3 */
+#include <errlog.h>
+
+/* standard */
+#include <set>
+#include <epicsTime.h>
+#include <sstream>
+
+using std::set;
+using namespace epics::pvData;
+
+namespace epics {
+    namespace pvAccess {
+
+        BlockingClientTCPTransport::BlockingClientTCPTransport(
+                Context* context, SOCKET channel,
+                ResponseHandler* responseHandler, int receiveBufferSize,
+                TransportClient* client, short remoteTransportRevision,
+                float beaconInterval, int16 priority) :
+            BlockingTCPTransport(context, channel, responseHandler,
+                    receiveBufferSize, priority), _introspectionRegistry(
+                    new IntrospectionRegistry(false)), _owners(new set<
+                    TransportClient*> ()), _connectionTimeout(beaconInterval
+                    *1000), _unresponsiveTransport(false), _timerNode(
+                    new TimerNode(this)), _mutex(new Mutex()), _ownersMutex(
+                    new Mutex()), _verifyOrEcho(true) {
+
+            // initialize owners list, send queue
+            acquire(client);
+
+            // use immediate for clients
+            setSendQueueFlushStrategy(IMMEDIATE);
+
+            // setup connection timeout timer (watchdog)
+            epicsTimeGetCurrent(const_cast<epicsTimeStamp*> (&_aliveTimestamp));
+
+            context->getTimer()->schedulePeriodic(_timerNode, beaconInterval,
+                    beaconInterval);
+
+            start();
+
+        }
+
+        BlockingClientTCPTransport::~BlockingClientTCPTransport() {
+            delete _introspectionRegistry;
+            delete _owners;
+            delete _timerNode;
+            delete _mutex;
+            delete _ownersMutex;
+        }
+
+        void BlockingClientTCPTransport::callback() {
+            epicsTimeStamp currentTime;
+            epicsTimeGetCurrent(&currentTime);
+
+            double diff = epicsTimeDiffInSeconds(&currentTime,
+                    const_cast<epicsTimeStamp*> (&_aliveTimestamp));
+            if(diff>2*_connectionTimeout) {
+                unresponsiveTransport();
+            }
+            else if(diff>_connectionTimeout) {
+                // send echo
+                enqueueSendRequest(this);
+            }
+        }
+
+        void BlockingClientTCPTransport::unresponsiveTransport() {
+            if(!_unresponsiveTransport) {
+                _unresponsiveTransport = true;
+
+                Lock lock(_ownersMutex);
+                set<TransportClient*>::iterator it = _owners->begin();
+                for(; it!=_owners->end(); it++)
+                    (*it)->transportUnresponsive();
+            }
+        }
+
+        bool BlockingClientTCPTransport::acquire(TransportClient* client) {
+            Lock lock(_mutex);
+
+            if(_closed) return false;
+
+            char ipAddrStr[48];
+            ipAddrToA(&_socketAddress->ia, ipAddrStr, sizeof(ipAddrStr));
+            errlogSevPrintf(errlogInfo, "Acquiring transport to %s.", ipAddrStr);
+
+            _ownersMutex->lock();
+            if(_closed) return false;
+
+            _owners->insert(client);
+            _ownersMutex->unlock();
+
+            return true;
+        }
+
+        void BlockingClientTCPTransport::internalClose(bool forced) {
+            BlockingTCPTransport::internalClose(forced);
+
+            _timerNode->cancel();
+
+            closedNotifyClients();
+        }
+
+        /**
+         * Notifies clients about disconnect.
+         */
+        void BlockingClientTCPTransport::closedNotifyClients() {
+            Lock lock(_ownersMutex);
+
+            // check if still acquired
+            int refs = _owners->size();
+            if(refs>0) {
+                ostringstream temp;
+                char ipAddrStr[48];
+                ipAddrToA(&_socketAddress->ia, ipAddrStr, sizeof(ipAddrStr));
+                temp<<"Transport to "<<ipAddrStr<<" still has "<<refs;
+                temp<<" client(s) active and closing...";
+                errlogSevPrintf(errlogInfo, temp.str().c_str());
+
+                set<TransportClient*>::iterator it = _owners->begin();
+                for(; it!=_owners->end(); it++)
+                    (*it)->transportClosed();
+            }
+
+            _owners->clear();
+        }
+
+        void BlockingClientTCPTransport::release(TransportClient* client) {
+            if(_closed) return;
+
+            char ipAddrStr[48];
+            ipAddrToA(&_socketAddress->ia, ipAddrStr, sizeof(ipAddrStr));
+
+            errlogSevPrintf(errlogInfo, "Releasing transport to %s.", ipAddrStr);
+
+            Lock lock(_ownersMutex);
+            _owners->erase(client);
+
+            // not used anymore
+            // TODO consider delayed destruction (can improve performance!!!)
+            if(_owners->size()==0) close(false);
+        }
+
+        void BlockingClientTCPTransport::aliveNotification() {
+            epicsTimeGetCurrent(const_cast<epicsTimeStamp*> (&_aliveTimestamp));
+            if(_unresponsiveTransport) responsiveTransport();
+        }
+
+        void BlockingClientTCPTransport::responsiveTransport() {
+            if(_unresponsiveTransport) {
+                _unresponsiveTransport = false;
+                Lock lock(_ownersMutex);
+
+                set<TransportClient*>::iterator it = _owners->begin();
+                for(; it!=_owners->end(); it++)
+                    (*it)->transportResponsive(this);
+            }
+        }
+
+        void BlockingClientTCPTransport::changedTransport() {
+            _introspectionRegistry->reset();
+            Lock lock(_ownersMutex);
+
+            set<TransportClient*>::iterator it = _owners->begin();
+            for(; it!=_owners->end(); it++)
+                (*it)->transportChanged();
+        }
+
+        void BlockingClientTCPTransport::send(ByteBuffer* buffer,
+                TransportSendControl* control) {
+            if(_verifyOrEcho) {
+                /*
+                 * send verification response message
+                 */
+
+                control->startMessage(1, 2*sizeof(int32)+sizeof(int16));
+
+                // receive buffer size
+                buffer->putInt(getReceiveBufferSize());
+
+                // socket receive buffer size
+                buffer->putInt(getSocketReceiveBufferSize());
+
+                // connection priority
+                buffer->putShort(getPriority());
+
+                // send immediately
+                control->flush(true);
+
+                _verifyOrEcho = false;
+            }
+            else {
+                control->startMessage(2, 0);
+                // send immediately
+                control->flush(true);
+            }
+
+        }
+
+    }
+}

pvAccessApp/remote/blockingTCP.h

@@ -13,16 +13,22 @@
 #include "remote.h"
 #include "growingCircularBuffer.h"
 #include "transportRegistry.h"
+#include "introspectionRegistry.h"
 
 /* pvData */
 #include <byteBuffer.h>
 #include <pvType.h>
 #include <lock.h>
 #include <epicsThread.h>
+#include <timer.h>
 
 /* EPICSv3 */
 #include <osdSock.h>
 #include <osiSock.h>
+#include <epicsTime.h>
+
+/* standard */
+#include <set>
 
 namespace epics {
     namespace pvAccess {
@@ -40,27 +46,27 @@ namespace epics {
         class BlockingTCPTransport : public Transport,
                 public TransportSendControl {
         public:
-            BlockingTCPTransport(SOCKET channel,
+            BlockingTCPTransport(Context* context, SOCKET channel,
                     ResponseHandler* responseHandler, int receiveBufferSize,
-                    short priority, TransportRegistry* transportRegistry);
+                    int16 priority);
 
-            ~BlockingTCPTransport();
+            virtual ~BlockingTCPTransport();
 
-            bool isClosed() const {
+            virtual bool isClosed() const {
                 return _closed;
             }
 
-            void setRemoteMinorRevision(int minorRevision) {
+            virtual void setRemoteMinorRevision(int8 minorRevision) {
                 _remoteTransportRevision = minorRevision;
             }
 
-            void setRemoteTransportReceiveBufferSize(
+            virtual void setRemoteTransportReceiveBufferSize(
                     int remoteTransportReceiveBufferSize) {
                 _remoteTransportReceiveBufferSize
                         = remoteTransportReceiveBufferSize;
             }
 
-            void setRemoteTransportSocketReceiveBufferSize(
+            virtual void setRemoteTransportSocketReceiveBufferSize(
                     int socketReceiveBufferSize) {
                 _remoteTransportSocketReceiveBufferSize
                         = socketReceiveBufferSize;
@@ -195,9 +201,10 @@ namespace epics {
 
             /**
              * Priority.
-             * NOTE: Priority cannot just be changed, since it is registered in transport registry with given priority.
+             * NOTE: Priority cannot just be changed, since it is registered
+             * in transport registry with given priority.
              */
-            short _priority;
+            int16 _priority;
             // TODO to be implemeneted
 
             /**
@@ -326,7 +333,7 @@ namespace epics {
 
             MonitorSender* _monitorSender;
 
-            TransportRegistry* _transportRegistry;
+            Context* _context;
 
             /**
              * Internal method that clears and releases buffer.
@@ -352,6 +359,171 @@ namespace epics {
             TransportSender* extractFromSendQueue();
         };
 
+        class BlockingClientTCPTransport : public BlockingTCPTransport,
+                public TransportSender,
+                public epics::pvData::TimerCallback,
+                public ReferenceCountingTransport {
+
+        public:
+            BlockingClientTCPTransport(Context* context, SOCKET channel,
+                    ResponseHandler* responseHandler, int receiveBufferSize,
+                    TransportClient* client, short remoteTransportRevision,
+                    float beaconInterval, int16 priority);
+
+            virtual ~BlockingClientTCPTransport();
+
+            virtual void timerStopped() {
+                // noop
+            }
+
+            virtual void callback();
+
+            /**
+             * 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* client);
+
+            virtual IntrospectionRegistry* getIntrospectionRegistry() {
+                return _introspectionRegistry;
+            }
+
+            /**
+             * Releases transport.
+             * @param client client (channel) releasing the transport
+             */
+            virtual void release(TransportClient* client);
+
+            /**
+             * Alive notification.
+             * This method needs to be called (by newly received data or beacon)
+             * at least once in this period, if not echo will be issued
+             * and if there is not response to it, transport will be considered as unresponsive.
+             */
+            virtual void aliveNotification();
+
+            /**
+             * Changed transport (server restared) notify.
+             */
+            virtual void changedTransport();
+
+            virtual void lock() {
+                // noop
+            }
+
+            virtual void unlock() {
+                // noop
+            }
+
+            virtual void send(epics::pvData::ByteBuffer* buffer,
+                    TransportSendControl* control);
+
+        protected:
+            /**
+             * Introspection registry.
+             */
+            IntrospectionRegistry* _introspectionRegistry;
+
+            virtual void internalClose(bool force);
+
+        private:
+
+            /**
+             * Owners (users) of the transport.
+             */
+            std::set<TransportClient*>* _owners;
+
+            /**
+             * Connection timeout (no-traffic) flag.
+             */
+            double _connectionTimeout;
+
+            /**
+             * Unresponsive transport flag.
+             */
+            volatile bool _unresponsiveTransport;
+
+            /**
+             * Timer task node.
+             */
+            TimerNode* _timerNode;
+
+            /**
+             * Timestamp of last "live" event on this transport.
+             */
+            volatile epicsTimeStamp _aliveTimestamp;
+
+            epics::pvData::Mutex* _mutex;
+            epics::pvData::Mutex* _ownersMutex;
+
+            bool _verifyOrEcho;
+
+            void unresponsiveTransport();
+
+            /**
+             * Notifies clients about disconnect.
+             */
+            void closedNotifyClients();
+
+            /**
+             * Responsive transport notify.
+             */
+            void responsiveTransport();
+        };
+
+        /**
+         * Channel Access TCP connector.
+         * @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
+         * @version $Id: BlockingTCPConnector.java,v 1.1 2010/05/03 14:45:47 mrkraimer Exp $
+         */
+        class BlockingTCPConnector : public Connector {
+        public:
+            BlockingTCPConnector(Context* context, int receiveBufferSize,
+                    float beaconInterval);
+
+            virtual ~BlockingTCPConnector();
+
+            virtual Transport* connect(TransportClient* client,
+                        ResponseHandler* responseHandler, osiSockAddr* address,
+                        short transportRevision, int16 priority);
+        private:
+            /**
+             * Lock timeout
+             */
+            static const int LOCK_TIMEOUT = 20*1000; // 20s
+
+            /**
+             * Context instance.
+             */
+            Context* _context;
+
+            /**
+             * Context instance.
+             */
+            //NamedLockPattern* _namedLocker;
+
+            /**
+             * Receive buffer size.
+             */
+            int _receiveBufferSize;
+
+            /**
+             * Beacon interval.
+             */
+            float _beaconInterval;
+
+            /**
+             * Tries to connect to the given address.
+             * @param[in] address
+             * @param[in] tries
+             * @return the SOCKET
+             * @throws IOException
+             */
+            SOCKET tryConnect(osiSockAddr* address, int tries);
+
+        };
+
     }
 }
 

pvAccessApp/remote/blockingTCPConnector.cpp

@@ -0,0 +1,174 @@
+/*
+ * blockingTCPConnector.cpp
+ *
+ *  Created on: Jan 4, 2011
+ *      Author: Miha Vitorovic
+ */
+
+#include "blockingTCP.h"
+#include "remote.h"
+
+#include <epicsThread.h>
+#include <osiSock.h>
+#include <errlog.h>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sstream>
+
+namespace epics {
+    namespace pvAccess {
+
+        BlockingTCPConnector::BlockingTCPConnector(Context* context,
+                int receiveBufferSize, float beaconInterval) :
+            _context(context), _receiveBufferSize(receiveBufferSize),
+                    _beaconInterval(beaconInterval)
+        //TODO , _namedLocker(new NamedLockPattern())
+        {
+        }
+
+        BlockingTCPConnector::~BlockingTCPConnector() {
+            // TODO delete _namedLocker;
+        }
+
+        SOCKET BlockingTCPConnector::tryConnect(osiSockAddr* address, int tries) {
+            for(int tryCount = 0; tryCount<tries; tryCount++) {
+                // sleep for a while
+                if(tryCount>0) epicsThreadSleep(0.1);
+
+                char strBuffer[64];
+                ipAddrToA(&address->ia, strBuffer, sizeof(strBuffer));
+
+                errlogSevPrintf(errlogInfo,
+                        "Opening socket to CA server %s, attempt %d.",
+                        strBuffer, tryCount+1);
+
+                SOCKET socket = epicsSocketCreate(AF_INET, SOCK_STREAM,
+                        IPPROTO_TCP);
+                if(socket==INVALID_SOCKET) {
+                    epicsSocketConvertErrnoToString(strBuffer,
+                            sizeof(strBuffer));
+                    errlogSevPrintf(errlogMinor, "Socket create error: %s",
+                            strBuffer);
+                }
+                else {
+                    if(::connect(socket, &address->sa, sizeof(sockaddr))==0)
+                        return socket;
+                    else {
+                        epicsSocketConvertErrnoToString(strBuffer,
+                                sizeof(strBuffer));
+                        errlogSevPrintf(errlogMinor,
+                                "Socket connect error: %s", strBuffer);
+                    }
+                }
+            }
+            return INVALID_SOCKET;
+        }
+
+        Transport* BlockingTCPConnector::connect(TransportClient* client,
+                ResponseHandler* responseHandler, osiSockAddr* address,
+                short transportRevision, int16 priority) {
+
+            SOCKET socket = INVALID_SOCKET;
+
+            char ipAddrStr[64];
+            ipAddrToA(&address->ia, ipAddrStr, sizeof(ipAddrStr));
+
+            // first try to check cache w/o named lock...
+            BlockingClientTCPTransport
+                    * transport =
+                            (BlockingClientTCPTransport*)(_context->getTransportRegistry()->get(
+                                    "TCP", address, priority));
+            if(transport!=NULL) {
+                errlogSevPrintf(errlogInfo,
+                        "Reusing existing connection to CA server: %s",
+                        ipAddrStr);
+                if(transport->acquire(client)) return transport;
+            }
+
+            bool lockAcquired = true;
+            // TODO comment out
+            //bool lockAcquired = _namedLocker->acquireSynchronizationObject(
+            //        address, LOCK_TIMEOUT);
+            if(lockAcquired) {
+                try {
+                    // ... transport created during waiting in lock
+                    transport
+                            = (BlockingClientTCPTransport*)(_context->getTransportRegistry()->get(
+                                    "TCP", address, priority));
+                    if(transport!=NULL) {
+                        errlogSevPrintf(errlogInfo,
+                                "Reusing existing connection to CA server: %s",
+                                ipAddrStr);
+                        if(transport->acquire(client)) return transport;
+                    }
+
+                    errlogSevPrintf(errlogInfo, "Connecting to CA server: %s",
+                            ipAddrStr);
+
+                    socket = tryConnect(address, 3);
+
+                    // use blocking channel
+                    // socket is blocking bya default
+                    //socket.configureBlocking(true);
+
+                    // enable TCP_NODELAY (disable Nagle's algorithm)
+                    int optval = 1; // true
+                    int retval = ::setsockopt(socket, IPPROTO_TCP, TCP_NODELAY,
+                            &optval, sizeof(optval));
+                    if(retval<0) errlogSevPrintf(errlogMajor,
+                            "Error setting TCP_NODELAY: %s", strerror(errno));
+
+                    // enable TCP_KEEPALIVE
+                    retval = ::setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE,
+                            &optval, sizeof(optval));
+                    if(retval<0) errlogSevPrintf(errlogMinor,
+                            "Error setting SO_KEEPALIVE: %s", strerror(errno));
+
+                    // TODO tune buffer sizes?! Win32 defaults are 8k, which is OK
+                    //socket.socket().setReceiveBufferSize();
+                    //socket.socket().setSendBufferSize();
+
+                    // create transport
+                    transport = new BlockingClientTCPTransport(_context, socket,
+                            responseHandler, _receiveBufferSize, client,
+                            transportRevision, _beaconInterval, priority);
+
+                    // verify
+                    if(!transport->waitUntilVerified(3.0)) {
+                        transport->close(true);
+                        errlogSevPrintf(
+                                errlogInfo,
+                                "Connection to CA client %s failed to be validated, closing it.",
+                                ipAddrStr);
+                        ostringstream temp;
+                        temp<<"Failed to verify TCP connection to '"<<ipAddrStr
+                                <<"'.";
+                        THROW_BASE_EXCEPTION(temp.str().c_str());
+                    }
+
+                    // TODO send security token
+
+                    errlogSevPrintf(errlogInfo, "Connected to CA server: %s",
+                            ipAddrStr);
+
+                    return transport;
+                } catch(...) {
+                    // close socket, if open
+                    if(socket!=INVALID_SOCKET) epicsSocketDestroy(socket);
+
+                    // TODO namedLocker.releaseSynchronizationObject(address);
+
+                    throw;
+                }
+            }
+            else {
+                ostringstream temp;
+                temp<<"Failed to obtain synchronization lock for '"<<ipAddrStr;
+                temp<<"', possible deadlock.";
+                THROW_BASE_EXCEPTION(temp.str().c_str());
+            }
+        }
+
+    }
+}

pvAccessApp/remote/blockingTCPTransport.cpp

@@ -62,9 +62,9 @@ namespace epics {
             GrowingCircularBuffer<TransportSender*>* _monitorSendQueue;
         };
 
-        BlockingTCPTransport::BlockingTCPTransport(SOCKET channel,
-                ResponseHandler* responseHandler, int receiveBufferSize,
-                short priority, TransportRegistry* transportRegistry) :
+        BlockingTCPTransport::BlockingTCPTransport(Context* context,
+                SOCKET channel, ResponseHandler* responseHandler,
+                int receiveBufferSize, int16 priority) :
             _closed(false), _channel(channel), _remoteTransportRevision(0),
                     _remoteTransportReceiveBufferSize(MAX_TCP_RECV),
                     _remoteTransportSocketReceiveBufferSize(MAX_TCP_RECV),
@@ -86,8 +86,7 @@ namespace epics {
                     _rcvThreadId(NULL), _sendThreadId(NULL), _monitorSendQueue(
                             new GrowingCircularBuffer<TransportSender*> (100)),
                     _monitorSender(new MonitorSender(_monitorMutex,
-                            _monitorSendQueue)), _transportRegistry(
-                            transportRegistry) {
+                            _monitorSendQueue)), _context(context) {
 
             _socketBuffer = new ByteBuffer(max(MAX_TCP_RECV
                     +MAX_ENSURE_DATA_BUFFER_SIZE, receiveBufferSize));
@@ -124,7 +123,7 @@ namespace epics {
             clearAndReleaseBuffer();
 
             // add to registry
-            _transportRegistry->put(this);
+            _context->getTransportRegistry()->put(this);
         }
 
         BlockingTCPTransport::~BlockingTCPTransport() {
@@ -195,7 +194,7 @@ namespace epics {
             _closed = true;
 
             // remove from registry
-            _transportRegistry->remove(this);
+            _context->getTransportRegistry()->remove(this);
 
             // clean resources
             internalClose(force);

pvAccessApp/remote/blockingUDP.h

@@ -17,6 +17,7 @@
 #include <noDefaultMethods.h>
 #include <byteBuffer.h>
 #include <lock.h>
+#include <epicsException.h>
 
 /* EPICSv3 */
 #include <osdSock.h>
@@ -160,6 +161,10 @@ namespace epics {
                 _sendAddresses = addresses;
             }
 
+            virtual IntrospectionRegistry* getIntrospectionRegistry() {
+                THROW_BASE_EXCEPTION("not supported by UDP transport");
+            }
+
         protected:
             bool volatile _closed;
 
@@ -237,7 +242,7 @@ namespace epics {
         };
 
         class BlockingUDPConnector : public Connector,
-                epics::pvData::NoDefaultMethods {
+                public epics::pvData::NoDefaultMethods {
         public:
 
             BlockingUDPConnector(bool reuseSocket,
@@ -255,7 +260,7 @@ namespace epics {
              */
             virtual Transport* connect(TransportClient* client,
                     ResponseHandler* responseHandler, osiSockAddr* bindAddress,
-                    short transportRevision, short priority);
+                    short transportRevision, int16 priority);
 
         private:
 

pvAccessApp/remote/blockingUDPConnector.cpp

@@ -25,7 +25,7 @@ namespace epics {
 
         Transport* BlockingUDPConnector::connect(TransportClient* client,
                 ResponseHandler* responseHandler, osiSockAddr* bindAddress,
-                short transportRevision, short priority) {
+                short transportRevision, int16 priority) {
             errlogSevPrintf(errlogInfo, "Creating datagram socket to: %s",
                     inetAddressToString(bindAddress).c_str());
 
@@ -59,11 +59,13 @@ namespace epics {
             retval = ::setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, &optval,
                     sizeof(optval));
             if(retval<0) errlogSevPrintf(errlogMajor,
-                    "Error binding socket: %s", strerror(errno));
+                    "Error setting SO_REUSEADDR: %s", strerror(errno));
 
             optval = _broadcast ? 1 : 0;
             retval = ::setsockopt(socket, SOL_SOCKET, SO_BROADCAST, &optval,
                     sizeof(optval));
+            if(retval<0) errlogSevPrintf(errlogMajor,
+                    "Error setting SO_BROADCAST: %s", strerror(errno));
 
             // sockets are blocking by default
 

pvAccessApp/remote/remote.h

@@ -9,10 +9,13 @@
 #define REMOTE_H_
 
 #include "caConstants.h"
+#include "transportRegistry.h"
+#include "introspectionRegistry.h"
 
 #include <serialize.h>
 #include <pvType.h>
 #include <byteBuffer.h>
+#include <timer.h>
 
 #include <osiSock.h>
 #include <osdSock.h>
@@ -20,6 +23,8 @@
 namespace epics {
     namespace pvAccess {
 
+        class TransportRegistry;
+
         enum ProtocolType {
             TCP, UDP, SSL
         };
@@ -54,7 +59,8 @@ namespace epics {
              * NOTE: these limitations allows efficient implementation.
              */
             virtual void
-            send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) =0;
+            send(epics::pvData::ByteBuffer* buffer,
+                    TransportSendControl* control) =0;
 
             virtual void lock() =0;
             virtual void unlock() =0;
@@ -156,7 +162,7 @@ namespace epics {
              * Get introspection registry for transport.
              * @return <code>IntrospectionRegistry</code> instance.
              */
-            //virtual IntrospectionRegistry getIntrospectionRegistry() =0;
+            virtual IntrospectionRegistry* getIntrospectionRegistry() =0;
 
             /**
              * Close transport.
@@ -206,9 +212,10 @@ namespace epics {
              *                      Note that this might not be the only message in the buffer.
              *                      Code must not manipulate buffer.
              */
-            virtual void handleResponse(osiSockAddr* responseFrom,
-                    Transport* transport, int8 version, int8 command,
-                    int payloadSize, epics::pvData::ByteBuffer* payloadBuffer) =0;
+            virtual void
+            handleResponse(osiSockAddr* responseFrom, Transport* transport,
+                    int8 version, int8 command, int payloadSize,
+                    epics::pvData::ByteBuffer* payloadBuffer) =0;
         };
 
         /**
@@ -261,10 +268,46 @@ namespace epics {
              */
             virtual Transport* connect(TransportClient* client,
                     ResponseHandler* responseHandler, osiSockAddr* address,
-                    short transportRevision, short priority) =0;
+                    short transportRevision, int16 priority) =0;
 
         };
 
+        class Context {
+        public:
+            /**
+             * Get timer.
+             * @return timer.
+             */
+            virtual Timer* getTimer() =0;
+
+            /**
+             * Get transport (virtual circuit) registry.
+             * @return transport (virtual circuit) registry.
+             */
+            virtual TransportRegistry* getTransportRegistry() =0;
+        };
+
+        /**
+         * Interface defining reference counting transport IF.
+         * @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
+         * @version $Id: ReferenceCountingTransport.java,v 1.1 2010/05/03 14:45:39 mrkraimer Exp $
+         */
+        class ReferenceCountingTransport {
+        public:
+            /**
+             * 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* client) =0;
+
+            /**
+             * Releases transport.
+             * @param client client (channel) releasing the transport
+             */
+            virtual void release(TransportClient* client) =0;
+        };
+
     }
 }
 

pvAccessApp/utils/transportRegistry.h

@@ -23,6 +23,7 @@ using namespace std;
 
 namespace epics { namespace pvAccess {
 
+class Transport;
 //TODO if unordered map is used instead of map we can use sockAddrAreIdentical routine from osiSock.h
 struct comp_osiSockAddr
 {

testApp/remote/testRemoteClientImpl.cpp

@@ -370,8 +370,6 @@ class MockMonitor : public Monitor, public MonitorElement
     
 };
 
-
-class TransportRegistry;
 class ChannelSearchManager;
 class BlockingTCPConnector;
 class NamedLockPattern;

testApp/utils/transportRegistryTest.cpp

@@ -4,6 +4,7 @@
  */
 
 #include "transportRegistry.h"
+#include "introspectionRegistry.h"
 #include "showConstructDestruct.h"
 
 #include <epicsAssert.h>
@@ -39,6 +40,7 @@ namespace epics {
         virtual void verified(){};
         virtual void enqueueSendRequest(TransportSender* sender){};
         virtual void ensureData(int) {};
+        virtual IntrospectionRegistry* getIntrospectionRegistry() {return NULL;};
 	private:
 		string _type;
 		int16 _priority;