Source · fairmq/devices/
Multiplier.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_MULTIPLIER_H10#define FAIR_MQ_MULTIPLIER_H11 12#include <fairmq/Device.h>13 14#include <string>15#include <vector>16 17namespace fair::mq18{19 20class Multiplier : public Device21{22 protected:23 bool fMultipart = true;24 int fNumOutputs = 0;25 std::string fInChannelName;26 std::vector<std::string> fOutChannelNames;27 28 void InitTask() override29 {30 fMultipart = fConfig->GetProperty<bool>("multipart");31 fInChannelName = fConfig->GetProperty<std::string>("in-channel");32 fOutChannelNames = fConfig->GetProperty<std::vector<std::string>>("out-channel");33 fNumOutputs = GetNumSubChannels(fOutChannelNames.at(0));34 35 if (fMultipart) {36 OnData(fInChannelName, &Multiplier::HandleMultipartData);37 } else {38 OnData(fInChannelName, &Multiplier::HandleSingleData);39 }40 }41 42 43 bool HandleSingleData(std::unique_ptr<Message>& payload, int)44 {45 for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel46 for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel47 MessagePtr msgCopy(fTransportFactory->CreateMessage());48 msgCopy->Copy(*payload);49 50 Send(msgCopy, fOutChannelNames.at(i), j);51 }52 }53 54 unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back());55 56 for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel57 MessagePtr msgCopy(fTransportFactory->CreateMessage());58 msgCopy->Copy(*payload);59 60 Send(msgCopy, fOutChannelNames.back(), i);61 }62 63 Send(payload, fOutChannelNames.back(), lastChannelSize - 1); // send final message to last subChannel of last channel64 65 return true;66 }67 68 bool HandleMultipartData(Parts& payload, int)69 {70 for (unsigned int i = 0; i < fOutChannelNames.size() - 1; ++i) { // all except last channel71 for (unsigned int j = 0; j < GetNumSubChannels(fOutChannelNames.at(i)); ++j) { // all subChannels in a channel72 Parts parts;73 74 for (unsigned int k = 0; k < payload.Size(); ++k) {75 MessagePtr msgCopy(fTransportFactory->CreateMessage());76 msgCopy->Copy(*(payload.At(k)));77 parts.AddPart(std::move(msgCopy));78 }79 80 Send(parts, fOutChannelNames.at(i), j);81 }82 }83 84 unsigned int lastChannelSize = GetNumSubChannels(fOutChannelNames.back());85 86 for (unsigned int i = 0; i < lastChannelSize - 1; ++i) { // iterate over all except last subChannels of the last channel87 Parts parts;88 89 for (unsigned int k = 0; k < payload.Size(); ++k) {90 MessagePtr msgCopy(fTransportFactory->CreateMessage());91 msgCopy->Copy(*(payload.At(k)));92 parts.AddPart(std::move(msgCopy));93 }94 95 Send(parts, fOutChannelNames.back(), i);96 }97 98 Send(payload, fOutChannelNames.back(), lastChannelSize - 1); // send final message to last subChannel of last channel99 100 return true;101 }102};103 104} // namespace fair::mq105 106#endif /* FAIR_MQ_MULTIPLIER_H */