InputRecordWalker.h

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
#ifndef FRAMEWORK_INPUTRECORDWALKER_H
#define FRAMEWORK_INPUTRECORDWALKER_H
 
 
#include "Framework/InputRecord.h"
#include "Framework/DataRefUtils.h"
 
namespace o2::framework
{
 
template <DataHeaderLike... EXTRA_HEADERS>
class InputRecordWalker
{
 public:
  InputRecordWalker() = delete;
  InputRecordWalker(InputRecord& record, std::vector<InputSpec> filterSpecs = {}) : mRecord(record), mFilterSpecs(filterSpecs) {}
 
  template <typename T>
  class Iterator
  {
   public:
    using self_type = Iterator;
    using iterator_category = std::forward_iterator_tag;
    using value_type = T;
    using reference = T&;
    using pointer = T*;
    using difference_type = std::ptrdiff_t;
    // the iterator over the input routes
    using input_iterator = decltype(std::declval<InputRecord>().begin());
 
    // the range over parts in one slot — must be stored as a member so the
    // part_iterator (which holds a pointer into it) does not dangle
    using part_range = InputRecord::PartRange;
    using part_iterator = InputSpan::Iterator<part_range, const DataRef>;
 
    Iterator() = delete;
 
    Iterator(input_iterator it, input_iterator end, std::vector<InputSpec> const& filterSpecs)
      : mInputIterator(it), mEnd(end), mCurrentRange(it.parts()), mCurrent(mCurrentRange.begin()), mFilterSpecs(filterSpecs)
    {
      next(true);
    }
 
    // prefix increment
    self_type& operator++()
    {
      next();
      return *this;
    }
    // postfix increment
    self_type operator++(int /*unused*/)
    {
      self_type copy(*this);
      operator++();
      return copy;
    }
    // return reference
    reference operator*()
    {
      return *mCurrent;
    }
    // comparison
    bool operator==(const self_type& other) const
    {
      return mInputIterator == other.mInputIterator && mCurrent == other.mCurrent;
    }
 
    bool operator!=(const self_type& rh) const
    {
      return not operator==(rh);
    }
 
    fair::mq::Message* getPayload() const
    {
      return mCurrentRange.getPayloadAtIndices(mCurrent.indices());
    }
 
   private:
    bool next(bool isInitialPart = false)
    {
      if (!isInitialPart) {
        ++mCurrent;
      }
      while (mInputIterator != mEnd) {
        for (; mCurrent != mCurrentRange.end(); ++mCurrent) {
          // check filter rules
          if (mFilterSpecs.size() > 0) {
            bool isSelected = false;
            for (auto const& spec : mFilterSpecs) {
              if ((isSelected = DataRefUtils::match<EXTRA_HEADERS...>(*mCurrent, spec)) == true) {
                break;
              }
            }
            if (!isSelected) {
              continue;
            }
          }
          return true;
        }
        ++mInputIterator;
        mCurrentRange = mInputIterator.parts();
        mCurrent = mCurrentRange.begin();
      } // end loop over record
      return false;
    }
 
    input_iterator mInputIterator;
    input_iterator mEnd;
    part_range mCurrentRange; // declared before mCurrent — initialized first
    part_iterator mCurrent;
    std::vector<InputSpec> const& mFilterSpecs;
  };
 
  using const_iterator = Iterator<DataRef const>;
 
  const_iterator begin() const
  {
    return const_iterator(mRecord.begin(), mRecord.end(), mFilterSpecs);
  }
 
  const_iterator end() const
  {
    return const_iterator(mRecord.end(), mRecord.end(), mFilterSpecs);
  }
 
 private:
  InputRecord& mRecord;
  std::vector<InputSpec> mFilterSpecs;
};
 
} // namespace o2::framework
 
#endif // FRAMEWORK_INPUTRECORDWALKER_H