FairMQChannel.h

/********************************************************************************
 * Copyright (C) 2014-2018 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH  *
 *                                                                              *
 *              This software is distributed under the terms of the             *
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
 
#ifndef FAIRMQCHANNEL_H_
#define FAIRMQCHANNEL_H_
 
#include <FairMQTransportFactory.h>
#include <FairMQUnmanagedRegion.h>
#include <FairMQSocket.h>
#include <fairmq/Transports.h>
#include <FairMQParts.h>
#include <fairmq/Properties.h>
#include <FairMQMessage.h>
 
#include <string>
#include <memory> // unique_ptr, shared_ptr
#include <vector>
#include <mutex>
#include <stdexcept>
#include <utility> // std::move
#include <cstdint> // int64_t
 
class FairMQChannel
{
    friend class FairMQDevice;
 
  public:
    FairMQChannel();
 
    FairMQChannel(const std::string& name);
 
    FairMQChannel(const std::string& type, const std::string& method, const std::string& address);
 
    FairMQChannel(const std::string& name, const std::string& type, std::shared_ptr<FairMQTransportFactory> factory);
 
    FairMQChannel(const std::string& name, const std::string& type, const std::string& method, const std::string& address, std::shared_ptr<FairMQTransportFactory> factory);
 
    FairMQChannel(const std::string& name, int index, const fair::mq::Properties& properties);
 
    FairMQChannel(const FairMQChannel&);
 
    FairMQChannel(const FairMQChannel&, const std::string& name);
 
    // FairMQChannel(FairMQChannel&&) = delete;
 
    FairMQChannel& operator=(const FairMQChannel&);
 
    // FairMQChannel& operator=(FairMQChannel&&) = delete;
 
    virtual ~FairMQChannel() { /* LOG(warn) << "Destroying channel '" << fName << "'"; */ }
 
    struct ChannelConfigurationError : std::runtime_error { using std::runtime_error::runtime_error; };
 
    FairMQSocket& GetSocket() const { assert(fSocket); return *fSocket; }
 
    bool Bind(const std::string& address)
    {
        fMethod = "bind";
        fAddress = address;
        return fSocket->Bind(address);
    }
 
    bool Connect(const std::string& address)
    {
        fMethod = "connect";
        fAddress = address;
        return fSocket->Connect(address);
    }
 
    std::string GetName() const { return fName; }
 
    std::string GetPrefix() const
    {
        std::string prefix = fName;
        prefix = prefix.erase(fName.rfind('['));
        return prefix;
    }
 
    std::string GetIndex() const
    {
        std::string indexStr = fName;
        indexStr.erase(indexStr.rfind(']'));
        indexStr.erase(0, indexStr.rfind('[') + 1);
        return indexStr;
    }
 
    std::string GetType() const { return fType; }
 
    std::string GetMethod() const { return fMethod; }
 
    std::string GetAddress() const { return fAddress; }
 
    std::string GetTransportName() const { return fair::mq::TransportName(fTransportType); }
 
    fair::mq::Transport GetTransportType() const { return fTransportType; }
 
    int GetSndBufSize() const { return fSndBufSize; }
 
    int GetRcvBufSize() const { return fRcvBufSize; }
 
    int GetSndKernelSize() const { return fSndKernelSize; }
 
    int GetRcvKernelSize() const { return fRcvKernelSize; }
 
    int GetLinger() const { return fLinger; }
 
    int GetRateLogging() const { return fRateLogging; }
 
    int GetPortRangeMin() const { return fPortRangeMin; }
 
    int GetPortRangeMax() const { return fPortRangeMax; }
 
    bool GetAutoBind() const { return fAutoBind; }
 
    void UpdateName(const std::string& name) { fName = name; Invalidate(); }
 
    void UpdateType(const std::string& type) { fType = type; Invalidate(); }
 
    void UpdateMethod(const std::string& method) { fMethod = method; Invalidate(); }
 
    void UpdateAddress(const std::string& address) { fAddress = address; Invalidate(); }
 
    void UpdateTransport(const std::string& transport) { fTransportType = fair::mq::TransportType(transport); Invalidate(); }
 
    void UpdateSndBufSize(const int sndBufSize) { fSndBufSize = sndBufSize; Invalidate(); }
 
    void UpdateRcvBufSize(const int rcvBufSize) { fRcvBufSize = rcvBufSize; Invalidate(); }
 
    void UpdateSndKernelSize(const int sndKernelSize) { fSndKernelSize = sndKernelSize; Invalidate(); }
 
    void UpdateRcvKernelSize(const int rcvKernelSize) { fRcvKernelSize = rcvKernelSize; Invalidate(); }
 
    void UpdateLinger(const int duration) { fLinger = duration; Invalidate(); }
 
    void UpdateRateLogging(const int rateLogging) { fRateLogging = rateLogging; Invalidate(); }
 
    void UpdatePortRangeMin(const int minPort) { fPortRangeMin = minPort; Invalidate(); }
 
    void UpdatePortRangeMax(const int maxPort) { fPortRangeMax = maxPort; Invalidate(); }
 
    void UpdateAutoBind(const bool autobind) { fAutoBind = autobind; Invalidate(); }
 
    bool IsValid() const { return fValid; }
 
    bool Validate();
 
    void Init();
 
    bool ConnectEndpoint(const std::string& endpoint);
 
    bool BindEndpoint(std::string& endpoint);
 
    void Invalidate() { fValid = false; }
 
    int64_t Send(FairMQMessagePtr& msg, int sndTimeoutInMs = -1)
    {
        CheckSendCompatibility(msg);
        return fSocket->Send(msg, sndTimeoutInMs);
    }
 
    int64_t Receive(FairMQMessagePtr& msg, int rcvTimeoutInMs = -1)
    {
        CheckReceiveCompatibility(msg);
        return fSocket->Receive(msg, rcvTimeoutInMs);
    }
 
    int64_t Send(std::vector<FairMQMessagePtr>& msgVec, int sndTimeoutInMs = -1)
    {
        CheckSendCompatibility(msgVec);
        return fSocket->Send(msgVec, sndTimeoutInMs);
    }
 
    int64_t Receive(std::vector<FairMQMessagePtr>& msgVec, int rcvTimeoutInMs = -1)
    {
        CheckReceiveCompatibility(msgVec);
        return fSocket->Receive(msgVec, rcvTimeoutInMs);
    }
 
    int64_t Send(FairMQParts& parts, int sndTimeoutInMs = -1)
    {
        return Send(parts.fParts, sndTimeoutInMs);
    }
 
    int64_t Receive(FairMQParts& parts, int rcvTimeoutInMs = -1)
    {
        return Receive(parts.fParts, rcvTimeoutInMs);
    }
 
    unsigned long GetBytesTx() const { return fSocket->GetBytesTx(); }
    unsigned long GetBytesRx() const { return fSocket->GetBytesRx(); }
    unsigned long GetMessagesTx() const { return fSocket->GetMessagesTx(); }
    unsigned long GetMessagesRx() const { return fSocket->GetMessagesRx(); }
 
    auto Transport() -> FairMQTransportFactory* { return fTransportFactory.get(); };
 
    template<typename... Args>
    FairMQMessagePtr NewMessage(Args&&... args)
    {
        return Transport()->CreateMessage(std::forward<Args>(args)...);
    }
 
    template<typename T>
    FairMQMessagePtr NewSimpleMessage(const T& data)
    {
        return Transport()->NewSimpleMessage(data);
    }
 
    template<typename T>
    FairMQMessagePtr NewStaticMessage(const T& data)
    {
        return Transport()->NewStaticMessage(data);
    }
 
    template<typename... Args>
    FairMQUnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)
    {
        return Transport()->CreateUnmanagedRegion(std::forward<Args>(args)...);
    }
 
    static constexpr fair::mq::Transport DefaultTransportType = fair::mq::Transport::DEFAULT;
    static constexpr const char* DefaultTransportName = "default";
    static constexpr const char* DefaultName = "";
    static constexpr const char* DefaultType = "unspecified";
    static constexpr const char* DefaultMethod = "unspecified";
    static constexpr const char* DefaultAddress = "unspecified";
    static constexpr int DefaultSndBufSize = 1000;
    static constexpr int DefaultRcvBufSize = 1000;
    static constexpr int DefaultSndKernelSize = 0;
    static constexpr int DefaultRcvKernelSize = 0;
    static constexpr int DefaultLinger = 500;
    static constexpr int DefaultRateLogging = 1;
    static constexpr int DefaultPortRangeMin = 22000;
    static constexpr int DefaultPortRangeMax = 23000;
    static constexpr bool DefaultAutoBind = true;
 
  private:
    std::shared_ptr<FairMQTransportFactory> fTransportFactory;
    fair::mq::Transport fTransportType;
    std::unique_ptr<FairMQSocket> fSocket;
 
    std::string fName;
    std::string fType;
    std::string fMethod;
    std::string fAddress;
    int fSndBufSize;
    int fRcvBufSize;
    int fSndKernelSize;
    int fRcvKernelSize;
    int fLinger;
    int fRateLogging;
    int fPortRangeMin;
    int fPortRangeMax;
    bool fAutoBind;
 
    bool fValid;
 
    bool fMultipart;
 
    void CheckSendCompatibility(FairMQMessagePtr& msg)
    {
        if (fTransportType != msg->GetType()) {
            FairMQMessagePtr msgWrapper(NewMessage(
                msg->GetData(),
                msg->GetSize(),
                [](void* /*data*/, void* _msg) { delete static_cast<FairMQMessage*>(_msg); },
                msg.get()
            ));
            msg.release();
            msg = move(msgWrapper);
        }
    }
 
    void CheckSendCompatibility(std::vector<FairMQMessagePtr>& msgVec)
    {
        for (auto& msg : msgVec) {
            if (fTransportType != msg->GetType()) {
 
                FairMQMessagePtr msgWrapper(NewMessage(
                    msg->GetData(),
                    msg->GetSize(),
                    [](void* /*data*/, void* _msg) { delete static_cast<FairMQMessage*>(_msg); },
                    msg.get()
                ));
                msg.release();
                msg = move(msgWrapper);
            }
        }
    }
 
    void CheckReceiveCompatibility(FairMQMessagePtr& msg)
    {
        if (fTransportType != msg->GetType()) {
            FairMQMessagePtr newMsg(NewMessage());
            msg = move(newMsg);
        }
    }
 
    void CheckReceiveCompatibility(std::vector<FairMQMessagePtr>& msgVec)
    {
        for (auto& msg : msgVec) {
            if (fTransportType != msg->GetType()) {
 
                FairMQMessagePtr newMsg(NewMessage());
                msg = move(newMsg);
            }
        }
    }
 
    void InitTransport(std::shared_ptr<FairMQTransportFactory> factory)
    {
        fTransportFactory = factory;
        fTransportType = factory->GetType();
    }
};
 
#endif /* FAIRMQCHANNEL_H_ */