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, shared_ptr
#include <vector>
#include <atomic>
#include <mutex>
#include <stdexcept>
#include <utility> // std::move
#include <FairMQTransportFactory.h>
#include <FairMQSocket.h>
#include <fairmq/Transports.h>
#include <FairMQLogger.h>
#include <FairMQParts.h>
#include <FairMQMessage.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 std::string& name, const std::string& type, const std::string& method, const std::string& address, std::shared_ptr<FairMQTransportFactory> factory);
FairMQChannel(const FairMQChannel&);
FairMQChannel& operator=(const FairMQChannel&);
virtual ~FairMQChannel() {}
struct ChannelConfigurationError : std::runtime_error { using std::runtime_error::runtime_error; };
FairMQSocket& GetSocket() const;
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 GetChannelName() const { return GetName(); } // TODO: deprecate this in favor of following
std::string GetName() const;
std::string GetChannelPrefix() const { return GetPrefix(); } // TODO: deprecate this in favor of following
std::string GetPrefix() const;
std::string GetChannelIndex() const { return GetIndex(); } // TODO: deprecate this in favor of following
std::string GetIndex() 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 GetLinger() const;
int GetRateLogging() const;
int GetPortRangeMin() const;
int GetPortRangeMax() const;
bool GetAutoBind() 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 UpdateLinger(const int duration);
void UpdateRateLogging(const int rateLogging);
void UpdatePortRangeMin(const int minPort);
void UpdatePortRangeMax(const int maxPort);
void UpdateAutoBind(const bool autobind);
void UpdateChannelName(const std::string& name) { UpdateName(name); } // TODO: deprecate this in favor of following
void UpdateName(const std::string& name);
bool IsValid() const;
bool ValidateChannel() // TODO: deprecate this
{
return Validate();
}
bool Validate();
void Init();
bool ConnectEndpoint(const std::string& endpoint);
bool BindEndpoint(std::string& endpoint);
void ResetChannel();
int Send(FairMQMessagePtr& msg, int sndTimeoutInMs = -1)
{
CheckSendCompatibility(msg);
return fSocket->Send(msg, sndTimeoutInMs);
}
int Receive(FairMQMessagePtr& msg, int rcvTimeoutInMs = -1)
{
CheckReceiveCompatibility(msg);
return fSocket->Receive(msg, rcvTimeoutInMs);
}
int SendAsync(FairMQMessagePtr& msg) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msg, timeout);")))
{
CheckSendCompatibility(msg);
return fSocket->Send(msg, 0);
}
int ReceiveAsync(FairMQMessagePtr& msg) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msg, timeout);")))
{
CheckReceiveCompatibility(msg);
return fSocket->Receive(msg, 0);
}
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 SendAsync(std::vector<FairMQMessagePtr>& msgVec) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(msgVec, timeout);")))
{
CheckSendCompatibility(msgVec);
return fSocket->Send(msgVec, 0);
}
int64_t ReceiveAsync(std::vector<FairMQMessagePtr>& msgVec) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(msgVec, timeout);")))
{
CheckReceiveCompatibility(msgVec);
return fSocket->Receive(msgVec, 0);
}
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);
}
int64_t SendAsync(FairMQParts& parts) __attribute__((deprecated("For non-blocking Send, use timeout version with timeout of 0: Send(parts, timeout);")))
{
return Send(parts.fParts, 0);
}
int64_t ReceiveAsync(FairMQParts& parts) __attribute__((deprecated("For non-blocking Receive, use timeout version with timeout of 0: Receive(parts, timeout);")))
{
return Receive(parts.fParts, 0);
}
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);
}
private:
std::shared_ptr<FairMQTransportFactory> fTransportFactory;
fair::mq::Transport fTransportType;
std::unique_ptr<FairMQSocket> fSocket;
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;
std::string fName;
std::atomic<bool> fIsValid;
// 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;
bool fReset;
void CheckSendCompatibility(FairMQMessagePtr& msg)
{
if (fTransportType != msg->GetType()) {
// LOG(debug) << "Channel type does not match message type. Creating wrapper";
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()) {
// LOG(debug) << "Channel type does not match message type. Creating wrapper";
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()) {
// LOG(debug) << "Channel type does not match message type. Creating wrapper";
FairMQMessagePtr newMsg(NewMessage());
msg = move(newMsg);
}
}
void CheckReceiveCompatibility(std::vector<FairMQMessagePtr>& msgVec)
{
for (auto& msg : msgVec) {
if (fTransportType != msg->GetType()) {
// LOG(debug) << "Channel type does not match message type. Creating wrapper";
FairMQMessagePtr newMsg(NewMessage());
msg = move(newMsg);
}
}
}
void InitTransport(std::shared_ptr<FairMQTransportFactory> factory)
{
fTransportFactory = factory;
fTransportType = factory->GetType();
}
auto SetModified(const bool modified) -> void;
};
#endif /* FAIRMQCHANNEL_H_ */
FairMQChannel::GetChannelIndex
std::string GetChannelIndex() const
Definition: FairMQChannel.h:94
FairMQChannel::GetPortRangeMax
int GetPortRangeMax() const
Definition: FairMQChannel.cxx:239
FairMQChannel::GetSndBufSize
int GetSndBufSize() const
Definition: FairMQChannel.cxx:176
FairMQChannel::GetTransportName
std::string GetTransportName() const
Definition: FairMQChannel.cxx:167
FairMQChannel::Send
int Send(FairMQMessagePtr &msg, int sndTimeoutInMs=-1)
Definition: FairMQChannel.h:234
FairMQChannel::UpdateAutoBind
void UpdateAutoBind(const bool autobind)
Definition: FairMQChannel.cxx:389
FairMQChannel::GetAutoBind
bool GetAutoBind() const
Definition: FairMQChannel.cxx:248
FairMQChannel::GetType
std::string GetType() const
Definition: FairMQChannel.cxx:140
FairMQChannel::Receive
int Receive(FairMQMessagePtr &msg, int rcvTimeoutInMs=-1)
Definition: FairMQChannel.h:244
FairMQChannel::UpdatePortRangeMin
void UpdatePortRangeMin(const int minPort)
Definition: FairMQChannel.cxx:367
FairMQChannel::GetChannelPrefix
std::string GetChannelPrefix() const
Definition: FairMQChannel.h:89
FairMQChannel::Send
int64_t Send(std::vector< FairMQMessagePtr > &msgVec, int sndTimeoutInMs=-1)
Definition: FairMQChannel.h:265
FairMQChannel::GetRateLogging
int GetRateLogging() const
Definition: FairMQChannel.cxx:221
FairMQChannel::GetAddress
std::string GetAddress() const
Definition: FairMQChannel.cxx:158
FairMQChannel::GetRcvKernelSize
int GetRcvKernelSize() const
Definition: FairMQChannel.cxx:203
FairMQTransportFactory
Definition: FairMQTransportFactory.h:28
FairMQChannel::GetPortRangeMin
int GetPortRangeMin() const
Definition: FairMQChannel.cxx:230
FairMQChannel::GetChannelName
std::string GetChannelName() const
Definition: FairMQChannel.h:84
FairMQChannel::UpdateRcvBufSize
void UpdateRcvBufSize(const int rcvBufSize)
Definition: FairMQChannel.cxx:312
FairMQChannel::operator=
FairMQChannel & operator=(const FairMQChannel &)
Assignment operator.
Definition: FairMQChannel.cxx:85
FairMQChannel
Definition: FairMQChannel.h:27
FairMQChannel::GetSndKernelSize
int GetSndKernelSize() const
Definition: FairMQChannel.cxx:194
FairMQChannel::IsValid
bool IsValid() const
Definition: FairMQChannel.cxx:420
FairMQChannel::UpdateRcvKernelSize
void UpdateRcvKernelSize(const int rcvKernelSize)
Definition: FairMQChannel.cxx:334
FairMQChannel::UpdateAddress
void UpdateAddress(const std::string &address)
Definition: FairMQChannel.cxx:279
FairMQChannel::UpdateTransport
void UpdateTransport(const std::string &transport)
Definition: FairMQChannel.cxx:290
FairMQChannel::Receive
int64_t Receive(FairMQParts &parts, int rcvTimeoutInMs=-1)
Definition: FairMQChannel.h:305
FairMQChannel::Receive
int64_t Receive(std::vector< FairMQMessagePtr > &msgVec, int rcvTimeoutInMs=-1)
Definition: FairMQChannel.h:275
FairMQSocket
Definition: FairMQSocket.h:19
FairMQChannel::UpdateRateLogging
void UpdateRateLogging(const int rateLogging)
Definition: FairMQChannel.cxx:356
FairMQChannel::GetLinger
int GetLinger() const
Definition: FairMQChannel.cxx:212
FairMQChannel::UpdateSndBufSize
void UpdateSndBufSize(const int sndBufSize)
Definition: FairMQChannel.cxx:301
FairMQChannel::Validate
bool Validate()
Definition: FairMQChannel.cxx:429
FairMQChannel::UpdateMethod
void UpdateMethod(const std::string &method)
Definition: FairMQChannel.cxx:268
FairMQChannel::UpdateChannelName
void UpdateChannelName(const std::string &name)
Definition: FairMQChannel.h:203
FairMQChannel::GetMethod
std::string GetMethod() const
Definition: FairMQChannel.cxx:149
FairMQParts
FairMQParts is a lightweight convenience wrapper around a vector of unique pointers to FairMQMessage...
Definition: FairMQParts.h:20
FairMQChannel::GetRcvBufSize
int GetRcvBufSize() const
Definition: FairMQChannel.cxx:185
FairMQChannel::UpdateLinger
void UpdateLinger(const int duration)
Definition: FairMQChannel.cxx:345
FairMQChannel::ResetChannel
void ResetChannel()
Resets the channel (requires validation to be used again).
Definition: FairMQChannel.cxx:637
FairMQChannel::ValidateChannel
bool ValidateChannel()
Definition: FairMQChannel.h:212
FairMQDevice
Definition: FairMQDevice.h:46
FairMQChannel::UpdatePortRangeMax
void UpdatePortRangeMax(const int maxPort)
Definition: FairMQChannel.cxx:378
FairMQChannel::UpdateType
void UpdateType(const std::string &type)
Definition: FairMQChannel.cxx:257
FairMQMessage
Definition: FairMQMessage.h:20
FairMQChannel::UpdateSndKernelSize
void UpdateSndKernelSize(const int sndKernelSize)
Definition: FairMQChannel.cxx:323
FairMQChannel::FairMQChannel
FairMQChannel()
Default constructor.
Definition: FairMQChannel.cxx:27
FairMQChannel::~FairMQChannel
virtual ~FairMQChannel()
Default destructor.
Definition: FairMQChannel.h:62
FairMQChannel::Send
int64_t Send(FairMQParts &parts, int sndTimeoutInMs=-1)
Definition: FairMQChannel.h:296
FairMQChannel::ChannelConfigurationError
Definition: FairMQChannel.h:64

privacy