[QC-450] Allow for output proxy channels which aren't "downstream" (#4631)

knopers8 · web-flow · commit ade60fa111d6 · 2020-10-20T09:06:55.000+02:00
diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -397,10 +397,10 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,
 
 namespace
 {
-// Decide where to sent the output. Everything to "downstream" for now
-std::string decideChannel(InputSpec const&)
+// Decide where to sent the output. Everything to "downstream" if there is such a channel.
+std::string decideChannel(InputSpec const& input, const std::unordered_map<std::string, std::vector<FairMQChannel>>& channels)
 {
-  return "downstream";
+  return channels.count("downstream") ? "downstream" : input.binding;
 }
 } // namespace
 
@@ -420,7 +420,7 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,
     auto channelConfigurationChecker = [inputSpecs = std::move(inputSpecs), device]() {
       LOG(INFO) << "checking channel configuration";
       for (auto const& spec : inputSpecs) {
-        auto channel = decideChannel(spec);
+        auto channel = decideChannel(spec, device->fChannels);
         if (device->fChannels.count(channel) == 0) {
           throw std::runtime_error("no corresponding output channel found for input '" + channel + "'");
         }
@@ -453,7 +453,7 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,
           }
           size_t payloadMsgSize = dh->payloadSize;
 
-          auto channel = decideChannel(*first.spec);
+          auto channel = decideChannel(*first.spec, device.fChannels);
           auto headerMessage = device.NewMessageFor(channel, index, headerMsgSize);
           memcpy(headerMessage->GetData(), part.header, headerMsgSize);
           auto payloadMessage = device.NewMessageFor(channel, index, payloadMsgSize);

Original file line number	Diff line number	Diff line change
`@@ -397,10 +397,10 @@ DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* name,`
`397`	`397`
`398`	`398`	`namespace`
`399`	`399`	`{`
`400`		`-// Decide where to sent the output. Everything to "downstream" for now`
`401`		`-std::string decideChannel(InputSpec const&)`
	`400`	`+// Decide where to sent the output. Everything to "downstream" if there is such a channel.`
	`401`	`+std::string decideChannel(InputSpec const& input, const std::unordered_map<std::string, std::vector<FairMQChannel>>& channels)`
`402`	`402`	`{`
`403`		`- return "downstream";`
	`403`	`+ return channels.count("downstream") ? "downstream" : input.binding;`
`404`	`404`	`}`
`405`	`405`	`} // namespace`
`406`	`406`
`@@ -420,7 +420,7 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,`
`420`	`420`	`auto channelConfigurationChecker = [inputSpecs = std::move(inputSpecs), device]() {`
`421`	`421`	`LOG(INFO) << "checking channel configuration";`
`422`	`422`	`for (auto const& spec : inputSpecs) {`
`423`		`- auto channel = decideChannel(spec);`
	`423`	`+ auto channel = decideChannel(spec, device->fChannels);`
`424`	`424`	`if (device->fChannels.count(channel) == 0) {`
`425`	`425`	`throw std::runtime_error("no corresponding output channel found for input '" + channel + "'");`
`426`	`426`	`}`
`@@ -453,7 +453,7 @@ DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* name,`
`453`	`453`	`}`
`454`	`454`	`size_t payloadMsgSize = dh->payloadSize;`
`455`	`455`
`456`		`- auto channel = decideChannel(*first.spec);`
	`456`	`+ auto channel = decideChannel(*first.spec, device.fChannels);`
`457`	`457`	`auto headerMessage = device.NewMessageFor(channel, index, headerMsgSize);`
`458`	`458`	`memcpy(headerMessage->GetData(), part.header, headerMsgSize);`
`459`	`459`	`auto payloadMessage = device.NewMessageFor(channel, index, payloadMsgSize);`