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 <string>
#include <memory> // unique_ptr
#include <vector>
#include <atomic>
#include <mutex>

#include <FairMQTransportFactory.h>
#include <FairMQSocket.h>
#include <fairmq/Transports.h>
#include <FairMQLogger.h>
#include <FairMQParts.h>

class FairMQChannel
{
    friend class FairMQDevice;

  public:
    FairMQChannel();

    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 FairMQChannel&);

    FairMQChannel& operator=(const FairMQChannel&);

    virtual ~FairMQChannel();

    FairMQSocket const & GetSocket() const;

    auto Bind(const std::string& address) -> bool
    {
        fMethod = "bind";
        fAddress = address;
        return fSocket->Bind(address);
    }

    auto Connect(const std::string& address) -> void
    {
        fMethod = "connect";
        fAddress = address;
        return fSocket->Connect(address);
    }

    std::string GetChannelName() const;

    std::string GetChannelPrefix() const;

    std::string GetChannelIndex() const;

    std::string GetType() const;

    std::string GetMethod() const;

    std::string GetAddress() const;

    std::string GetTransport() const;

    int GetSndBufSize() const;

    int GetRcvBufSize() const;

    int GetSndKernelSize() const;

    int GetRcvKernelSize() const;

    int GetRateLogging() const;

    void UpdateType(const std::string& type);

    void UpdateMethod(const std::string& method);

    void UpdateAddress(const std::string& address);

    void UpdateTransport(const std::string& transport);

    void UpdateSndBufSize(const int sndBufSize);

    void UpdateRcvBufSize(const int rcvBufSize);

    void UpdateSndKernelSize(const int sndKernelSize);

    void UpdateRcvKernelSize(const int rcvKernelSize);

    void UpdateRateLogging(const int rateLogging);

    void UpdateChannelName(const std::string& name);

    bool IsValid() const;

    bool ValidateChannel();

    void ResetChannel();

    int Send(std::unique_ptr<FairMQMessage>& msg) const;
    int Receive(std::unique_ptr<FairMQMessage>& msg) const;

    int Send(std::unique_ptr<FairMQMessage>& msg, int sndTimeoutInMs) const;

    int Receive(std::unique_ptr<FairMQMessage>& msg, int rcvTimeoutInMs) const;

    int SendAsync(std::unique_ptr<FairMQMessage>& msg) const;

    int ReceiveAsync(std::unique_ptr<FairMQMessage>& msg) const;

    int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;
    int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;

    int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int sndTimeoutInMs) const;

    int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, int rcvTimeoutInMs) const;

    int64_t SendAsync(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;

    int64_t ReceiveAsync(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;

    int64_t Send(FairMQParts& parts) const
    {
        return Send(parts.fParts);
    }

    int64_t Receive(FairMQParts& parts) const
    {
        return Receive(parts.fParts);
    }

    int64_t Send(FairMQParts& parts, int sndTimeoutInMs) const
    {
        return Send(parts.fParts, sndTimeoutInMs);
    }

    int64_t Receive(FairMQParts& parts, int rcvTimeoutInMs) const
    {
        return Receive(parts.fParts, rcvTimeoutInMs);
    }

    int64_t SendAsync(FairMQParts& parts) const
    {
        return SendAsync(parts.fParts);
    }

    int64_t ReceiveAsync(FairMQParts& parts) const
    {
        return ReceiveAsync(parts.fParts);
    }

    unsigned long GetBytesTx() const;
    unsigned long GetBytesRx() const;
    unsigned long GetMessagesTx() const;
    unsigned long GetMessagesRx() const;

    auto Transport() const -> const FairMQTransportFactory*
    {
        return fTransportFactory.get();
    };

    template<typename... Args>
    FairMQMessagePtr NewMessage(Args&&... args) const
    {
        return Transport()->CreateMessage(std::forward<Args>(args)...);
    }

    template<typename T>
    FairMQMessagePtr NewSimpleMessage(const T& data) const
    {
        return Transport()->NewSimpleMessage(data);
    }

    template<typename T>
    FairMQMessagePtr NewStaticMessage(const T& data) const
    {
        return Transport()->NewStaticMessage(data);
    }

  private:
    std::unique_ptr<FairMQSocket> fSocket;

    std::string fType;
    std::string fMethod;
    std::string fAddress;
    std::string fTransport;
    int fSndBufSize;
    int fRcvBufSize;
    int fSndKernelSize;
    int fRcvKernelSize;
    int fRateLogging;

    std::string fName;
    std::atomic<bool> fIsValid;

    FairMQ::Transport fTransportType;
    std::shared_ptr<FairMQTransportFactory> fTransportFactory;

    bool CheckCompatibility(std::unique_ptr<FairMQMessage>& msg) const;
    bool CheckCompatibility(std::vector<std::unique_ptr<FairMQMessage>>& msgVec) const;

    void InitTransport(std::shared_ptr<FairMQTransportFactory> factory);

    // use static mutex to make the class easily copyable
    // implication: same mutex is used for all instances of the class
    // this does not hurt much, because mutex is used only during initialization with very low contention
    // possible TODO: improve this
    static std::mutex fChannelMutex;

    bool fMultipart;
    bool fModified;
    auto SetModified(const bool modified) -> void;
    bool fReset;
};

#endif /* FAIRMQCHANNEL_H_ */