GeezerVison is showing github.com — AliceO2/run/SimPublishChannelHelper.h at dev · AliceO2Group/AliceO2

Expand file tree

SimPublishChannelHelper.h

Copy path

More file actions

Latest commit

History

79 lines (70 loc) · 2.77 KB

SimPublishChannelHelper.h

Copy path

File metadata and controls

79 lines (70 loc) · 2.77 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.

// All rights not expressly granted are reserved.

// This software is distributed under the terms of the GNU General Public

// License v3 (GPL Version 3), copied verbatim in the file "COPYING".

// In applying this license CERN does not waive the privileges and immunities

// granted to it by virtue of its status as an Intergovernmental Organization

// or submit itself to any jurisdiction.

#ifndef O2_SIMPUBLISHCHANNELHELPER_H

#define O2_SIMPUBLISHCHANNELHELPER_H

#include <string>

#include <fairmq/Channel.h>

#include <fairmq/Message.h>

#include <fairmq/TransportFactory.h>

namespace o2::simpubsub

{

// create an IPC socket name of the type

// ipc:///tmp/base-PID

// base should be for example "o2sim-worker" or "o2sim-merger"

inline std::string getPublishAddress(std::string const& base, int pid = getpid())

{

std::stringstream publishsocketname;

publishsocketname << "ipc:///tmp/" << base << "-" << pid;

return publishsocketname.str();

}

// some standard format for pub-sub subscribers

inline std::string simStatusString(std::string const& origin, std::string const& topic, std::string const& message)

{

return origin + std::string("[") + topic + std::string("] : ") + message;

}

// helper function to publish a message to an outside subscriber

inline bool publishMessage(fair::mq::Channel& channel, std::string const& message)

{

if (channel.IsValid()) {

auto text = new std::string(message);

std::unique_ptr<fair::mq::Message> payload(channel.NewMessage(

const_cast<char*>(text->c_str()),

text->length(), [](void* data, void* hint) { delete static_cast<std::string*>(hint); }, text));

if (channel.Send(payload) > 0) {

return true;

}

} else {

LOG(error) << "CHANNEL NOT VALID";

}

return false;

}

// make channel (transport factory needs to be injected)

inline fair::mq::Channel createPUBChannel(std::string const& address,

std::string const& type = "pub")

{

auto factory = fair::mq::TransportFactory::CreateTransportFactory("zeromq");

static int i = 0;

std::stringstream str;

str << "channel" << i++;

auto externalpublishchannel = fair::mq::Channel{str.str(), type, factory};

externalpublishchannel.Init();

if ((type.compare("pub") == 0) || (type.compare("pull") == 0)) {

externalpublishchannel.Bind(address);

LOG(info) << "BINDING TO ADDRESS " << address << " type " << type;

} else {

externalpublishchannel.Connect(address);

LOG(info) << "CONNECTING TO ADDRESS " << address << " type " << type;

}

externalpublishchannel.Validate();

return externalpublishchannel;

}

} // namespace o2::simpubsub

#endif //O2_SIMPUBLISHCHANNELHELPER_H