Source · fairmq/devices/
Merger.hheader
Described in builtin-devices: a topology of built-in devices, Design a topology, Built-in devices.
1/********************************************************************************2 * Copyright (C) 2014-2023 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_MERGER_H10#define FAIR_MQ_MERGER_H11 12#include <fairmq/Poller.h>13#include <fairmq/Device.h>14 15#include <fairlogger/Logger.h>16#include <string>17#include <vector>18 19namespace fair::mq20{21 22class Merger : public Device23{24 protected:25 bool fMultipart = true;26 std::string fInChannelName{"data-in"};27 std::string fOutChannelName{"data-out"};28 29 void InitTask() override30 {31 fMultipart = fConfig->GetProperty<bool>("multipart");32 fInChannelName = fConfig->GetProperty<std::string>("in-channel");33 fOutChannelName = fConfig->GetProperty<std::string>("out-channel");34 }35 36 void RegisterChannelEndpoints() override37 {38 RegisterChannelEndpoint(fInChannelName, 1, 10000);39 RegisterChannelEndpoint(fOutChannelName, 1, 1);40 41 PrintRegisteredChannels();42 }43 44 void Run() override45 {46 int numInputs = GetNumSubChannels(fInChannelName);47 48 std::vector<Channel*> chans;49 50 for (auto& chan : GetChannels().at(fInChannelName)) {51 chans.push_back(&chan);52 }53 54 PollerPtr poller(NewPoller(chans));55 56 if (fMultipart) {57 while (!NewStatePending()) {58 poller->Poll(100);59 60 // Loop over the data input channels.61 for (int i = 0; i < numInputs; ++i) {62 // Check if the channel has data ready to be received.63 if (poller->CheckInput(i)) {64 Parts payload;65 66 if (Receive(payload, fInChannelName, i) >= 0) {67 if (Send(payload, fOutChannelName) < 0) {68 LOG(debug) << "Transfer interrupted";69 break;70 }71 } else {72 LOG(debug) << "Transfer interrupted";73 break;74 }75 }76 }77 }78 } else {79 while (!NewStatePending()) {80 poller->Poll(100);81 82 // Loop over the data input channels.83 for (int i = 0; i < numInputs; ++i) {84 // Check if the channel has data ready to be received.85 if (poller->CheckInput(i)) {86 MessagePtr payload(fTransportFactory->CreateMessage());87 88 if (Receive(payload, fInChannelName, i) >= 0) {89 if (Send(payload, fOutChannelName) < 0) {90 LOG(debug) << "Transfer interrupted";91 break;92 }93 } else {94 LOG(debug) << "Transfer interrupted";95 break;96 }97 }98 }99 }100 }101 }102};103 104} // namespace fair::mq105 106#endif /* FAIR_MQ_MERGER_H */