activating handling of variable payload sequences

Author: matthiasrichter

Framework/Core/src/DataProcessingDevice.cxx

@@ -932,20 +932,28 @@ void DataProcessingDevice::handleData(DataProcessorContext& context, InputChanne
         LOGP(error, "Header stack does not contain DataProcessingHeader");
         continue;
       }
-      // We can set the type for the next splitPayloadParts
-      // because we are guaranteed they are all the same.
-      // If splitPayloadParts = 0, we assume that means there is only one (header, payload)
-      // pair.
-      size_t finalSplitPayloadIndex = pi + (dh->splitPayloadParts > 0 ? dh->splitPayloadParts : 1) * 2;
-      if (finalSplitPayloadIndex > parts.Size()) {
-        LOGP(error, "DataHeader::splitPayloadParts invalid");
-        results.back().type = InputType::Invalid;
-        continue;
-      }
-      results.back().type = InputType::Data;
-      results.back().size = 2;
-      for (; pi + 2 < finalSplitPayloadIndex; pi += 2) {
-        results.emplace_back(pi + 2, 2, InputType::Data);
+      if (dh->splitPayloadParts > 0 && dh->splitPayloadParts == dh->splitPayloadIndex) {
+        // this is indicating a sequence of payloads following the header
+        // FIXME: we will probably also set the DataHeader version
+        results.back().type = InputType::Data;
+        results.back().size = dh->splitPayloadParts;
+        pi += dh->splitPayloadParts - 1;
+      } else {
+        // We can set the type for the next splitPayloadParts
+        // because we are guaranteed they are all the same.
+        // If splitPayloadParts = 0, we assume that means there is only one (header, payload)
+        // pair.
+        size_t finalSplitPayloadIndex = pi + (dh->splitPayloadParts > 0 ? dh->splitPayloadParts : 1) * 2;
+        if (finalSplitPayloadIndex > parts.Size()) {
+          LOGP(error, "DataHeader::splitPayloadParts invalid");
+          results.back().type = InputType::Invalid;
+          continue;
+        }
+        results.back().type = InputType::Data;
+        results.back().size = 2;
+        for (; pi + 2 < finalSplitPayloadIndex; pi += 2) {
+          results.emplace_back(pi + 2, 2, InputType::Data);
+        }
       }
     }
     // FIXME: this is an intermediate check, remove it for supporting O2 messages with
@@ -970,12 +978,19 @@ void DataProcessingDevice::handleData(DataProcessorContext& context, InputChanne
       switch (input.type) {
         case InputType::Data: {
           auto headerIndex = input.position;
-          auto payloadIndex = input.position + 1;
-          assert(payloadIndex < parts.Size());
-          auto dh = o2::header::get<DataHeader*>(parts.At(headerIndex)->GetData());
+          auto nMessages = 0;
+          if (input.size > 1) {
+            // header and multiple payload sequence
+            nMessages = input.size + 1;
+          } else {
+            // multiple header-payload pairs
+            auto dh = o2::header::get<DataHeader*>(parts.At(headerIndex)->GetData());
+            nMessages = dh->splitPayloadParts > 0 ? dh->splitPayloadParts * 2 : 2;
+          }
           auto relayed = relayer.relay(parts.At(headerIndex)->GetData(),
                                        &parts.At(headerIndex),
-                                       dh->splitPayloadParts > 0 ? dh->splitPayloadParts * 2 : 2);
+                                       nMessages,
+                                       input.size);
           switch (relayed) {
             case DataRelayer::Backpressured:
               policy.backpressure(info);