Improved testing of the DPL proxies

matthiasrichter · ktf · commit 13fef42e6bc6 · 2022-02-28T09:57:01.000+01:00
The default input and output proxies can not be tested together in the
same workflow configuration because data descriptions on the producer
and consumer side can not be the same. There are now two workflow
configurations:
1. For testing the default input proxy, the producer is opening the
out of band channel directly and the output proxy is skipped.

2. For testing the default output proxy, the input proxy is created
with a custom converter function to adjust data description. A specific
unit test is added with the necessary command line option
diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
@@ -348,3 +348,17 @@ o2_add_test(
     --consumer
     "--global-config consumer-config --local-option hello-aliceo2 --a-boolean3 --an-int2 20 --a-double2 22. --an-int64-2 50000000000000"
   )
+
+# the test is compiled from the ExternalFairMQDeviceWorkflow test and run with
+# command line option to include the output proxy
+o2_add_test(
+  ExternalFairMQOutputProxyWorkflow NAME test_Framework_test_ExternalFairMQOutputProxyWorkflow
+  SOURCES test/test_ExternalFairMQDeviceWorkflow.cxx
+  COMPONENT_NAME Framework
+  LABELS framework workflow
+  TIMEOUT 60
+  PUBLIC_LINK_LIBRARIES O2::Framework
+  NO_BOOST_TEST
+  COMMAND_LINE_ARGS
+    --proxy-mode all --run ${DPL_WORKFLOW_TESTS_EXTRA_OPTIONS}
+  )
diff --git a/Framework/Core/test/test_ExternalFairMQDeviceWorkflow.cxx b/Framework/Core/test/test_ExternalFairMQDeviceWorkflow.cxx
@@ -373,17 +373,28 @@ std::vector<DataProcessorSpec> defineDataProcessing(ConfigContext const& config)
     if (channelName.empty()) {
       return;
     }
-    // make a copy of the header message, get the data header and change origin
-    auto outHeaderMessage = device.NewMessageFor(channelName, 0, inputs.At(msgidx)->GetSize());
-    memcpy(outHeaderMessage->GetData(), inputs.At(msgidx)->GetData(), inputs.At(msgidx)->GetSize());
-    // this we obviously need to fix in the get API, const'ness of the returned header pointer
-    // should depend on const'ness of the buffer
-    auto odh = const_cast<o2::header::DataHeader*>(o2::header::get<o2::header::DataHeader*>(outHeaderMessage->GetData()));
-    odh->dataOrigin = o2::header::DataOrigin("PRX");
     FairMQParts output;
-    output.AddPart(std::move(outHeaderMessage));
-    output.AddPart(std::move(inputs.At(msgidx + 1)));
-    LOG(debug) << "sending " << DataSpecUtils::describe(OutputSpec{odh->dataOrigin, odh->dataDescription, odh->subSpecification});
+    for (; msgidx < inputs.Size(); ++msgidx) {
+      auto const* dh = o2::header::get<o2::header::DataHeader*>(inputs.At(msgidx)->GetData());
+      if (dh) {
+        LOGP(debug, "{}/{}/{} with {} part(s), index {}",
+             dh->dataOrigin.as<std::string>(),
+             dh->dataDescription.as<std::string>(),
+             dh->subSpecification,
+             dh->splitPayloadParts,
+             dh->splitPayloadIndex);
+        // make a copy of the header message, get the data header and change origin
+        auto outHeaderMessage = device.NewMessageFor(channelName, 0, inputs.At(msgidx)->GetSize());
+        memcpy(outHeaderMessage->GetData(), inputs.At(msgidx)->GetData(), inputs.At(msgidx)->GetSize());
+        // this we obviously need to fix in the get API, const'ness of the returned header pointer
+        // should depend on const'ness of the buffer
+        auto odh = const_cast<o2::header::DataHeader*>(o2::header::get<o2::header::DataHeader*>(outHeaderMessage->GetData()));
+        odh->dataOrigin = o2::header::DataOrigin("PRX");
+        output.AddPart(std::move(outHeaderMessage));
+      } else {
+        output.AddPart(std::move(inputs.At(msgidx)));
+      }
+    }
     o2::framework::sendOnChannel(device, output, channelName);
   };
 
