Source · fairmq/devices/
Proxy.hheader
1/********************************************************************************2 * Copyright (C) 2014-2021 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_PROXY_H10#define FAIR_MQ_PROXY_H11 12#include <fairmq/Device.h>13 14#include <string>15 16namespace fair::mq17{18 19class Proxy : public Device20{21 protected:22 bool fMultipart = true;23 std::string fInChannelName;24 std::string fOutChannelName;25 26 void InitTask() override27 {28 fMultipart = fConfig->GetProperty<bool>("multipart");29 fInChannelName = fConfig->GetProperty<std::string>("in-channel");30 fOutChannelName = fConfig->GetProperty<std::string>("out-channel");31 }32 33 void Run() override34 {35 if (fMultipart) {36 while (!NewStatePending()) {37 Parts payload;38 if (Receive(payload, fInChannelName) >= 0) {39 if (Send(payload, fOutChannelName) < 0) {40 LOG(debug) << "Transfer interrupted";41 break;42 }43 } else {44 LOG(debug) << "Transfer interrupted";45 break;46 }47 }48 } else {49 while (!NewStatePending()) {50 MessagePtr payload(fTransportFactory->CreateMessage());51 if (Receive(payload, fInChannelName) >= 0) {52 if (Send(payload, fOutChannelName) < 0) {53 LOG(debug) << "Transfer interrupted";54 break;55 }56 } else {57 LOG(debug) << "Transfer interrupted";58 break;59 }60 }61 }62 }63};64 65} // namespace fair::mq66 67#endif /* FAIR_MQ_PROXY_H */