DPL: extend DataRefUtils::match to support multiple headers (#14160)

Author: ktf

Framework/Core/include/Framework/DataRefUtils.h

@@ -11,6 +11,7 @@
 #ifndef O2_FRAMEWORK_DATAREFUTILS_H_
 #define O2_FRAMEWORK_DATAREFUTILS_H_
 
+#include "Framework/DataDescriptorMatcher.h"
 #include "Framework/DataRef.h"
 #include "Framework/RootSerializationSupport.h"
 #include "Framework/SerializationMethods.h"
@@ -33,6 +34,9 @@ class ConfigurableParam;
 namespace o2::framework
 {
 
+template <typename H>
+concept DataHeaderLike = requires(H& dh) {dh.dataOrigin; dh.dataDescription; dh.subSpecification; };
+
 // FIXME: Should enforce the fact that DataRefs are read only...
 struct DataRefUtils {
 
@@ -52,7 +56,7 @@ struct DataRefUtils {
       if ((payloadSize % sizeof(T)) != 0) {
         throw runtime_error("Cannot extract POD from message as size do not match");
       }
-      //FIXME: provide a const collection
+      // FIXME: provide a const collection
       return gsl::span<T>(reinterpret_cast<T*>(const_cast<char*>(ref.payload)), payloadSize / sizeof(T));
     } else if constexpr (has_root_dictionary<T>::value == true &&
                          is_messageable<T>::value == false) {
@@ -220,17 +224,24 @@ struct DataRefUtils {
     return ref.spec != nullptr && ref.spec->binding == binding;
   }
 
-  /// check if the O2 message referred by DataRef matches a particular
-  /// input spec. The DataHeader is retrieved from the header message and matched
-  /// against @ref spec parameter.
-  static bool match(DataRef const& ref, InputSpec const& spec)
+  template <DataHeaderLike H>
+  static bool matchHeader(DataRef const& ref, InputSpec const& spec)
   {
-    auto dh = DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    auto const* dh = o2::header::get<H*>(ref.header);
     if (dh == nullptr) {
       return false;
     }
     return DataSpecUtils::match(spec, dh->dataOrigin, dh->dataDescription, dh->subSpecification);
   }
+
+  /// check if the O2 message referred by DataRef matches a particular
+  /// input spec. The DataHeader is retrieved from the header message and matched
+  /// against @ref spec parameter.
+  template <DataHeaderLike... H>
+  static bool match(DataRef const& ref, InputSpec const& spec)
+  {
+    return (DataRefUtils::matchHeader<H>(ref, spec) || ... || matchHeader<o2::header::DataHeader>(ref, spec));
+  }
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/InputRecordWalker.h

@@ -12,11 +12,11 @@
 #define FRAMEWORK_INPUTRECORDWALKER_H
 
 /// @file   InputRecordWalker.h
-/// @author Matthias Richter
 /// @since  2020-03-25
 /// @brief  A helper class to iteratate over all parts of all input routes
 
 #include "Framework/InputRecord.h"
+#include "Framework/DataRefUtils.h"
 
 namespace o2::framework
 {
@@ -49,6 +49,7 @@ namespace o2::framework
 ///   for (auto const& ref : InputRecordWalker(inputs, filter)) {
 ///     // do something with the data
 ///   }
+template <DataHeaderLike... EXTRA_HEADERS>
 class InputRecordWalker
 {
  public:
@@ -131,7 +132,7 @@ class InputRecordWalker
           if (mFilterSpecs.size() > 0) {
             bool isSelected = false;
             for (auto const& spec : mFilterSpecs) {
-              if ((isSelected = DataRefUtils::match(*mCurrent, spec)) == true) {
+              if ((isSelected = DataRefUtils::match<EXTRA_HEADERS...>(*mCurrent, spec)) == true) {
                 break;
               }
             }

Framework/Utils/include/DPLUtils/DPLRawPageSequencer.h

@@ -191,7 +191,7 @@ class DPLRawPageSequencer
   }
 
  private:
-  InputRecordWalker mInput;
+  InputRecordWalker<> mInput;
 
   template <typename Predicate, typename Inserter>
   void forwardInternal(Predicate pred, Inserter inserter, const char* data, size_t size, const o2::header::DataHeader* dh)