DPL: add allocator to forward a fair::mq::Message payload by shallow copy

Author: ehellbar

Framework/Core/include/Framework/DataAllocator.h

@@ -455,6 +455,11 @@ class DataAllocator
   void snapshot(const Output& spec, const char* payload, size_t payloadSize,
                 o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
 
+  /// create a shallow copy of the @a inputPayload and forward it to the output route of @a spec
+  /// if the transport types of the input and output routes are different, a real copy is created as fallback
+  void forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
+                      o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
+
   /// make an object of type T and route to output specified by OutputRef
   /// The object is owned by the framework, returned reference can be used to fill the object.
   ///

Framework/Core/src/DataAllocator.cxx

@@ -342,6 +342,26 @@ void DataAllocator::snapshot(const Output& spec, const char* payload, size_t pay
   addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
 }
 
+void DataAllocator::forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
+                                   o2::header::SerializationMethod serializationMethod)
+{
+  auto& proxy = mRegistry.get<FairMQDeviceProxy>();
+  auto& timingInfo = mRegistry.get<TimingInfo>();
+
+  RouteIndex routeIndex = matchDataHeader(spec, timingInfo.timeslice);
+  auto* transport = proxy.getOutputTransport(routeIndex);
+
+  if (inputPayload.GetTransport() == transport) {
+    auto payloadMessage = transport->CreateMessage();
+    payloadMessage->Copy(inputPayload);
+    addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
+  } else {
+    auto payloadMessage = transport->CreateMessage(inputPayload.GetSize(), fair::mq::Alignment{64});
+    memcpy(payloadMessage->GetData(), inputPayload.GetData(), inputPayload.GetSize());
+    addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
+  }
+}
+
 Output DataAllocator::getOutputByBind(OutputRef&& ref)
 {
   if (ref.label.empty()) {