DataSamplingPolicy.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 ALICEO2_DATASAMPLINGPOLICY_H
#define ALICEO2_DATASAMPLINGPOLICY_H
 
 
#include "Headers/DataHeader.h"
#include "Framework/InputSpec.h"
#include "Framework/Output.h"
#include "Framework/OutputSpec.h"
#include "DataSampling/DataSamplingCondition.h"
#include "Framework/DataSpecUtils.h"
 
#include <boost/property_tree/ptree_fwd.hpp>
 
namespace o2::utilities
{
 
class DataSamplingPolicy
{
 private:
  using PathVectorBase = std::vector<std::pair<framework::InputSpec, framework::OutputSpec>>;
  class PathMap : public PathVectorBase
  {
   public:
    PathMap() = default;
    ~PathMap() = default;
    const PathVectorBase::const_iterator find(const framework::ConcreteDataMatcher& input) const
    {
      return std::find_if(begin(), end(), [input](const auto& el) {
        return framework::DataSpecUtils::match(el.first, input);
      });
    }
  };
 
 public:
  static DataSamplingPolicy fromConfiguration(const boost::property_tree::ptree&);
 
  DataSamplingPolicy(std::string name);
  ~DataSamplingPolicy() = default;
  DataSamplingPolicy(DataSamplingPolicy&&) = default;
  DataSamplingPolicy& operator=(DataSamplingPolicy&&) = default;
 
  void registerPath(const framework::InputSpec&, const framework::OutputSpec&);
  //  void registerPolicy(framework::InputSpec&&, framework::OutputSpec&&);
  void registerCondition(std::unique_ptr<DataSamplingCondition>&&);
  void setFairMQOutputChannel(std::string);
 
  const framework::OutputSpec* match(const framework::ConcreteDataMatcher& input) const;
  bool decide(const o2::framework::DataRef&);
  framework::Output prepareOutput(const framework::ConcreteDataMatcher& input, framework::Lifetime lifetime = framework::Lifetime::Timeframe) const;
 
  const std::string& getName() const;
  const PathMap& getPathMap() const;
  // optional fairmq channel to send stuff outside of DPL
  const std::string& getFairMQOutputChannel() const;
  std::string getFairMQOutputChannelName() const;
  uint32_t getTotalAcceptedMessages() const;
  uint32_t getTotalEvaluatedMessages() const;
  bool isActive() const;
 
  static header::DataOrigin createPolicyDataOrigin();
  static header::DataDescription createPolicyDataDescription(std::string policyName, size_t id);
 
 private:
  bool mActive = true;
  std::string mName;
  PathMap mPaths;
  std::vector<std::unique_ptr<DataSamplingCondition>> mConditions;
  std::string mFairMQOutputChannel;
 
  // stats
  uint32_t mTotalAcceptedMessages = 0;
  uint32_t mTotalEvaluatedMessages = 0;
};
 
} // namespace o2::utilities
 
#endif // ALICEO2_DATASAMPLINGPOLICY_H