blockingUDP.h

 
#ifndef BLOCKINGUDP_H_
#define BLOCKINGUDP_H_
 
#ifdef epicsExportSharedSymbols
#   define blockingUDPEpicsExportSharedSymbols
#   undef epicsExportSharedSymbols
#endif
 
#include <shareLib.h>
#include <osiSock.h>
#include <epicsThread.h>
 
#include <pv/noDefaultMethods.h>
#include <pv/byteBuffer.h>
#include <pv/lock.h>
#include <pv/event.h>
#include <pv/pvIntrospect.h>
 
#ifdef blockingUDPEpicsExportSharedSymbols
#   define epicsExportSharedSymbols
#       undef blockingUDPEpicsExportSharedSymbols
#endif
 
#include <shareLib.h>
 
#include <pv/remote.h>
#include <pv/pvaConstants.h>
#include <pv/inetAddressUtil.h>
 
namespace epics {
namespace pvAccess {
 
class ClientChannelImpl;
class BlockingUDPConnector;
 
enum InetAddressType { inetAddressType_all, inetAddressType_unicast, inetAddressType_broadcast_multicast };
 
class BlockingUDPTransport :
    public Transport,
    public TransportSendControl,
    public epicsThreadRunable
{
    EPICS_NOT_COPYABLE(BlockingUDPTransport)
public:
    POINTER_DEFINITIONS(BlockingUDPTransport);
 
    static size_t num_instances;
 
private:
    std::tr1::weak_ptr<BlockingUDPTransport> internal_this;
    friend class BlockingUDPConnector;
    BlockingUDPTransport(bool serverFlag,
                         ResponseHandler::shared_pointer const & responseHandler,
                         SOCKET channel, osiSockAddr &bindAddress,
                         short remoteTransportRevision);
public:
 
    virtual ~BlockingUDPTransport();
 
    virtual bool isClosed() OVERRIDE FINAL {
        return _closed.get();
    }
 
    virtual const osiSockAddr& getRemoteAddress() const OVERRIDE FINAL {
        return _remoteAddress;
    }
 
    virtual const std::string& getRemoteName() const OVERRIDE FINAL {
        return _remoteName;
    }
 
    virtual std::string getType() const OVERRIDE FINAL {
        return std::string("udp");
    }
 
    virtual std::size_t getReceiveBufferSize() const OVERRIDE FINAL {
        return _receiveBuffer.getSize();
    }
 
    virtual epics::pvData::int16 getPriority() const OVERRIDE FINAL {
        return PVA_DEFAULT_PRIORITY;
    }
 
    virtual void setRemoteTransportReceiveBufferSize(
        std::size_t /*receiveBufferSize*/) OVERRIDE FINAL {
        // noop for UDP (limited by 64k; MAX_UDP_SEND for PVA)
    }
 
    virtual void setRemoteTransportSocketReceiveBufferSize(
        std::size_t /*socketReceiveBufferSize*/) OVERRIDE FINAL {
        // noop for UDP (limited by 64k; MAX_UDP_SEND for PVA)
    }
 
    virtual bool verify(epics::pvData::int32 /*timeoutMs*/) OVERRIDE FINAL {
        // noop
        return true;
    }
 
    virtual void verified(epics::pvData::Status const & /*status*/) OVERRIDE FINAL {
        // noop
    }
 
    virtual void authNZMessage(epics::pvData::PVStructure::shared_pointer const & data) OVERRIDE FINAL {
        // noop
    }
 
    // NOTE: this is not yet used for UDP
    virtual void setByteOrder(int byteOrder) OVERRIDE FINAL  {
        // called from receive thread... or before processing
        _receiveBuffer.setEndianess(byteOrder);
 
        // sync?!
        _sendBuffer.setEndianess(byteOrder);
    }
 
    virtual void enqueueSendRequest(TransportSender::shared_pointer const & sender) OVERRIDE FINAL;
 
    virtual void flushSendQueue() OVERRIDE FINAL;
 
    void start();
 
    virtual void close() OVERRIDE FINAL;
 
    virtual void ensureData(std::size_t size) OVERRIDE FINAL;
 
    virtual bool directSerialize(epics::pvData::ByteBuffer* /*existingBuffer*/, const char* /*toSerialize*/,
                                 std::size_t /*elementCount*/, std::size_t /*elementSize*/) OVERRIDE FINAL
    {
        return false;
    }
 
    virtual bool directDeserialize(epics::pvData::ByteBuffer* /*existingBuffer*/, char* /*deserializeTo*/,
                                   std::size_t /*elementCount*/, std::size_t /*elementSize*/) OVERRIDE FINAL
    {
        return false;
    }
 
    virtual void startMessage(epics::pvData::int8 command, std::size_t ensureCapacity, epics::pvData::int32 payloadSize = 0) OVERRIDE FINAL;
    virtual void endMessage() OVERRIDE FINAL;
 
    virtual void flush(bool /*lastMessageCompleted*/) OVERRIDE FINAL {
        // noop since all UDP requests are sent immediately
    }
 
    virtual void setRecipient(const osiSockAddr& sendTo) OVERRIDE FINAL {
        _sendToEnabled = true;
        _sendTo = sendTo;
    }
 
    void setLocalMulticastAddress(const osiSockAddr& sendTo) {
        _localMulticastAddressEnabled = true;
        _localMulticastAddress = sendTo;
    }
 
    bool hasLocalMulticastAddress() const {
        return _localMulticastAddressEnabled;
    }
 
    const osiSockAddr& getLocalMulticastAddress() const {
        return _localMulticastAddress;
    }
 
    virtual void flushSerializeBuffer() OVERRIDE FINAL {
        // noop
    }
 
    virtual void ensureBuffer(std::size_t /*size*/) OVERRIDE FINAL {
        // noop
    }
 
    virtual void cachedSerialize(
        const std::tr1::shared_ptr<const epics::pvData::Field>& field, epics::pvData::ByteBuffer* buffer) OVERRIDE FINAL
    {
        // no cache
        field->serialize(buffer, this);
    }
 
    virtual std::tr1::shared_ptr<const epics::pvData::Field>
    cachedDeserialize(epics::pvData::ByteBuffer* buffer) OVERRIDE FINAL
    {
        // no cache
        // TODO
        return epics::pvData::getFieldCreate()->deserialize(buffer, this);
    }
 
    virtual bool acquire(std::tr1::shared_ptr<ClientChannelImpl> const & /*client*/) OVERRIDE FINAL
    {
        return false;
    }
 
    virtual void release(pvAccessID /*clientId*/) OVERRIDE FINAL {}
 
    void setIgnoredAddresses(const InetAddrVector& addresses) {
        _ignoredAddresses = addresses;
    }
 
    const InetAddrVector& getIgnoredAddresses() const {
        return _ignoredAddresses;
    }
 
    void setTappedNIF(const InetAddrVector& addresses) {
        _tappedNIF = addresses;
    }
 
    const InetAddrVector& getTappedNIF() const {
        return _tappedNIF;
    }
 
    bool send(const char* buffer, size_t length, const osiSockAddr& address);
 
    bool send(epics::pvData::ByteBuffer* buffer, const osiSockAddr& address);
 
    bool send(epics::pvData::ByteBuffer* buffer, InetAddressType target = inetAddressType_all);
 
    const InetAddrVector& getSendAddresses() {
        return _sendAddresses;
    }
 
    const osiSockAddr* getBindAddress() const {
        return &_bindAddress;
    }
 
    bool isBroadcastAddress(const osiSockAddr* address, const InetAddrVector& broadcastAddresses)
    {
        for (size_t i = 0; i < broadcastAddresses.size(); i++)
            if (broadcastAddresses[i].ia.sin_addr.s_addr == address->ia.sin_addr.s_addr)
                return true;
        return false;
    }
 
    // consumes arguments
    void setSendAddresses(InetAddrVector& addresses, std::vector<bool>& address_types) {
        _sendAddresses.swap(addresses);
        _isSendAddressUnicast.swap(address_types);
    }
 
    void join(const osiSockAddr & mcastAddr, const osiSockAddr & nifAddr);
 
    void setMutlicastNIF(const osiSockAddr & nifAddr, bool loopback);
 
protected:
    AtomicBoolean _closed;
 
    ResponseHandler::shared_pointer _responseHandler;
 
    virtual void run() OVERRIDE FINAL;
 
private:
    bool processBuffer(Transport::shared_pointer const & transport, osiSockAddr& fromAddress, epics::pvData::ByteBuffer* receiveBuffer);
 
    void close(bool waitForThreadToComplete);
 
    // Context only used for logging in this class
 
    SOCKET _channel;
 
    /* When provided, this transport is used for replies (passed to handler)
     * instead of *this.  This feature is used in the situation where broadcast
     * traffic is received on one socket, but a different socket must be used
     * for unicast replies.
     *
    Transport::shared_pointer _replyTransport;
    */
 
    osiSockAddr _bindAddress;
 
    osiSockAddr _remoteAddress;
    std::string _remoteName;
 
    InetAddrVector _sendAddresses;
 
    std::vector<bool> _isSendAddressUnicast;
 
    InetAddrVector _ignoredAddresses;
 
    InetAddrVector _tappedNIF;
 
    osiSockAddr _sendTo;
    bool _sendToEnabled;
 
    osiSockAddr _localMulticastAddress;
    bool _localMulticastAddressEnabled;
 
    epics::pvData::ByteBuffer _receiveBuffer;
 
    epics::pvData::ByteBuffer _sendBuffer;
 
    int _lastMessageStartPosition;
 
    epics::pvData::Mutex _mutex;
    epics::pvData::Mutex _sendMutex;
 
    epics::auto_ptr<epicsThread> _thread;
 
    epics::pvData::int8 _clientServerWithEndianFlag;
 
};
 
class BlockingUDPConnector{
public:
    POINTER_DEFINITIONS(BlockingUDPConnector);
 
    BlockingUDPConnector(bool serverFlag) :_serverFlag(serverFlag) {}
 
    BlockingUDPTransport::shared_pointer connect(
            ResponseHandler::shared_pointer const & responseHandler,
            osiSockAddr& bindAddress,
            epics::pvData::int8 transportRevision);
 
private:
 
    bool _serverFlag;
 
    EPICS_NOT_COPYABLE(BlockingUDPConnector)
};
 
typedef std::vector<BlockingUDPTransport::shared_pointer> BlockingUDPTransportVector;
 
void initializeUDPTransports(
    bool serverFlag,
    BlockingUDPTransportVector& udpTransports,
    const IfaceNodeVector& ifaceList,
    const ResponseHandler::shared_pointer& responseHandler,
    BlockingUDPTransport::shared_pointer& sendTransport,
    epics::pvData::int32& listenPort,
    bool autoAddressList,
    const std::string& addressList,
    const std::string& ignoreAddressList);
 
 
}
}
 
#endif /* BLOCKINGUDP_H_ */