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 GetTransportName() 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;
fair::mq::Transport fTransportType;
int fSndBufSize;
int fRcvBufSize;
int fSndKernelSize;
int fRcvKernelSize;
int fRateLogging;
std::string fName;
std::atomic<bool> fIsValid;
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_ */
FairMQChannel::GetSndBufSize
int GetSndBufSize() const
Definition: FairMQChannel.cxx:206
FairMQChannel::GetTransportName
std::string GetTransportName() const
Definition: FairMQChannel.cxx:192
FairMQChannel::GetType
std::string GetType() const
Definition: FairMQChannel.cxx:150
FairMQChannel::ReceiveAsync
int ReceiveAsync(std::unique_ptr< FairMQMessage > &msg) const
FairMQChannel::SendAsync
int SendAsync(std::unique_ptr< FairMQMessage > &msg) const
FairMQChannel::~FairMQChannel
virtual ~FairMQChannel()
Default destructor.
Definition: FairMQChannel.cxx:729
FairMQChannel::GetRateLogging
int GetRateLogging() const
Definition: FairMQChannel.cxx:262
FairMQChannel::GetAddress
std::string GetAddress() const
Definition: FairMQChannel.cxx:178
FairMQChannel::GetRcvKernelSize
int GetRcvKernelSize() const
Definition: FairMQChannel.cxx:248
FairMQTransportFactory
Definition: FairMQTransportFactory.h:27
FairMQDevice::Transport
auto Transport() const -> const FairMQTransportFactory *
Getter for default transport factory.
Definition: FairMQDevice.h:197
FairMQChannel::GetChannelPrefix
std::string GetChannelPrefix() const
Definition: FairMQChannel.cxx:135
FairMQChannel::UpdateRcvBufSize
void UpdateRcvBufSize(const int rcvBufSize)
Definition: FairMQChannel.cxx:358
FairMQChannel::operator=
FairMQChannel & operator=(const FairMQChannel &)
Assignment operator.
Definition: FairMQChannel.cxx:105
FairMQChannel
Definition: FairMQChannel.h:24
FairMQChannel::GetSndKernelSize
int GetSndKernelSize() const
Definition: FairMQChannel.cxx:234
FairMQChannel::IsValid
bool IsValid() const
Definition: FairMQChannel.cxx:452
FairMQChannel::UpdateRcvKernelSize
void UpdateRcvKernelSize(const int rcvKernelSize)
Definition: FairMQChannel.cxx:390
FairMQChannel::UpdateAddress
void UpdateAddress(const std::string &address)
Definition: FairMQChannel.cxx:308
FairMQChannel::UpdateTransport
void UpdateTransport(const std::string &transport)
Definition: FairMQChannel.cxx:324
FairMQChannel::GetChannelName
std::string GetChannelName() const
Definition: FairMQChannel.cxx:130
FairMQSocket
Definition: FairMQSocket.h:18
FairMQChannel::UpdateRateLogging
void UpdateRateLogging(const int rateLogging)
Definition: FairMQChannel.cxx:406
FairMQChannel::GetChannelIndex
std::string GetChannelIndex() const
Definition: FairMQChannel.cxx:142
FairMQChannel::UpdateSndBufSize
void UpdateSndBufSize(const int sndBufSize)
Definition: FairMQChannel.cxx:342
FairMQChannel::UpdateMethod
void UpdateMethod(const std::string &method)
Definition: FairMQChannel.cxx:292
FairMQChannel::UpdateChannelName
void UpdateChannelName(const std::string &name)
Definition: FairMQChannel.cxx:436
FairMQChannel::GetMethod
std::string GetMethod() const
Definition: FairMQChannel.cxx:164
FairMQParts
FairMQParts is a lightweight convenience wrapper around a vector of unique pointers to FairMQMessage...
Definition: FairMQParts.h:20
FairMQDevice::fTransportFactory
std::shared_ptr< FairMQTransportFactory > fTransportFactory
Default transport factory.
Definition: FairMQDevice.h:417
FairMQChannel::GetRcvBufSize
int GetRcvBufSize() const
Definition: FairMQChannel.cxx:220
FairMQChannel::ResetChannel
void ResetChannel()
Resets the channel (requires validation to be used again).
Definition: FairMQChannel.cxx:649
FairMQChannel::ValidateChannel
bool ValidateChannel()
Definition: FairMQChannel.cxx:466
FairMQDevice
Definition: FairMQDevice.h:44
FairMQChannel::UpdateType
void UpdateType(const std::string &type)
Definition: FairMQChannel.cxx:276
FairMQChannel::UpdateSndKernelSize
void UpdateSndKernelSize(const int sndKernelSize)
Definition: FairMQChannel.cxx:374
FairMQChannel::FairMQChannel
FairMQChannel()
Default constructor.
Definition: FairMQChannel.cxx:26