ExternalFairMQDeviceProxy.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_RAWDEVICESOURCE_H
#define FRAMEWORK_RAWDEVICESOURCE_H
 
#include "Framework/DataProcessorSpec.h"
#include "Framework/OutputSpec.h"
#include <fairmq/FwdDecls.h>
#include <vector>
#include <functional>
 
namespace o2::framework
{
 
using ChannelRetriever = std::function<std::string const&(OutputSpec const&, DataProcessingHeader::StartTime)>;
using InjectorFunction = std::function<bool(TimingInfo&, ServiceRegistryRef const& services, fair::mq::Parts& inputs, ChannelRetriever, size_t newTimesliceId, bool& stop)>;
using ChannelSelector = std::function<std::string(InputSpec const& input, const std::unordered_map<std::string, std::vector<fair::mq::Channel>>& channels)>;
 
struct InputChannelSpec;
struct OutputChannelSpec;
 
std::string formatExternalChannelConfiguration(InputChannelSpec const&);
 
std::string formatExternalChannelConfiguration(OutputChannelSpec const&);
 
void sendOnChannel(fair::mq::Device& device, o2::header::Stack&& headerStack, fair::mq::MessagePtr&& payloadMessage, OutputSpec const& spec, ChannelRetriever& channelRetriever);
 
void sendOnChannel(fair::mq::Device& device, fair::mq::Parts& messages, std::string const& channel, size_t timeSlice);
 
void appendForSending(fair::mq::Device& device, o2::header::Stack&& headerStack, size_t timeSliceID, fair::mq::MessagePtr&& payloadMessage, OutputSpec const& spec, fair::mq::Parts& messageCache, ChannelRetriever& channelRetriever);
 
InjectorFunction incrementalConverter(OutputSpec const& spec, o2::header::SerializationMethod method, uint64_t startTime, uint64_t step);
 
InjectorFunction o2DataModelAdaptor(OutputSpec const& spec, uint64_t startTime, uint64_t step);
 
struct DPLModelAdapterConfig {
  bool throwOnUnmatchedInputs = true;
  bool paranoid = false;
  bool blindForward = false;
};
 
InjectorFunction dplModelAdaptor(std::vector<OutputSpec> const& specs = {{header::gDataOriginAny, header::gDataDescriptionAny}},
                                 DPLModelAdapterConfig config = DPLModelAdapterConfig{});
 
inline InjectorFunction dplModelAdaptor(std::vector<OutputSpec> const& specs, bool throwOnUnmatchedInputs)
{
  return dplModelAdaptor(specs, DPLModelAdapterConfig{throwOnUnmatchedInputs});
}
 
static auto gDefaultConverter = incrementalConverter(OutputSpec{"TST", "TEST", 0}, header::gSerializationMethodROOT, 0, 1);
 
std::string defaultOutputProxyChannelSelector(InputSpec const& input, const std::unordered_map<std::string, std::vector<fair::mq::Channel>>& channels);
 
DataProcessorSpec specifyExternalFairMQDeviceProxy(char const* label,
                                                   std::vector<OutputSpec> const& outputs,
                                                   const char* defaultChannelConfig,
                                                   InjectorFunction converter,
                                                   uint64_t minSHM = 0,
                                                   bool sendTFcounter = false,
                                                   bool doInjectMissingData = false,
                                                   unsigned int doPrintSizes = 0);
 
DataProcessorSpec specifyFairMQDeviceOutputProxy(char const* label,
                                                 Inputs const& inputSpecs,
                                                 const char* defaultChannelConfig);
DataProcessorSpec specifyFairMQDeviceMultiOutputProxy(char const* label,
                                                      Inputs const& inputSpecs,
                                                      const char* defaultChannelConfig,
                                                      ChannelSelector channelSelector = defaultOutputProxyChannelSelector);
 
} // namespace o2::framework
 
#endif // FRAMEWORK_RAWDEVICESOURCE_H