VtFeeder.cxx

Go to the documentation of this file.

/* Copyright (C) 2025 Jagiellonian University, Krakow
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Bartosz Sobol [committer],  Jan de Cuveland */
 
#include "VtFeeder.h"
 
#include <TimesliceAutoSource.hpp>
#include <TimesliceOutputArchive.hpp>
#include <TimeslicePublisher.hpp>
 
#include <chrono>
#include <thread>
 
#include <fmt/format.h>
 
VtFeeder::VtFeeder(const FeederOptions& options, volatile sig_atomic_t* signalStatus)
  : fOptions{options}
  , fSignalStatus{signalStatus}
  , fSource{}
  , fPublishers{}
  , fCount{0}
{
  fSource = std::make_unique<VtSource>(fOptions.fInput);
 
  for (const auto& output : fOptions.fOutputs) {
    UriComponents uri{output};
 
    if (uri.scheme == "tcp") {
      uint32_t hwm = 1;
      for (auto& [key, value] : uri.query_components) {
        if (key == "hwm") {
          hwm = stou(value);
        }
        else {
          throw std::runtime_error(
            fmt::format("[VtFeeder] Query parameter not implemented for scheme {}: {}", uri.scheme, key));
        }
      }
      const auto address = uri.scheme + "://" + uri.authority;
      fPublishers.push_back(std::unique_ptr<fles::TimesliceSink>(new fles::TimeslicePublisher(address, hwm)));
    }
    else if (uri.scheme == "file" || uri.scheme.empty()) {
      fles::ArchiveCompression compression = fles::ArchiveCompression::None;
      for (const auto& [key, value] : uri.query_components) {
        if (key == "c") {
          if (value == "zstd") {
            compression = fles::ArchiveCompression::Zstd;
          }
          else if (value != "none") {
            throw std::runtime_error(fmt::format("[VtFeeder] invalid compression type for scheme file: {}", value));
          }
        }
        else {
          throw std::runtime_error(
            fmt::format("[VtFeeder] Query parameter not implemented for scheme {}: {}", uri.scheme, key));
        }
      }
      const auto filePath = uri.authority + uri.path;
      fPublishers.push_back(
        std::unique_ptr<fles::TimesliceSink>(new fles::TimesliceOutputArchive(filePath, compression)));
    }
    else {
      throw std::runtime_error(fmt::format("[VtFeeder] Unsupported output scheme: {}", uri.scheme));
    }
  }
}
 
void VtFeeder::Run()
{
  uint64_t i = 0;
  std::unique_ptr<fles::StorableTimeslice> timeslice{};
  while (*fSignalStatus == 0 and (timeslice = fSource->get())) {
    if (i >= fOptions.fOffset && (i - fOptions.fOffset) % fOptions.fStride == 0) {
      ++i;
    }
    else {
      ++i;
      continue;
    }
 
    std::shared_ptr<const fles::StorableTimeslice> ts{};
    if (fOptions.fReleaseMode) {
      ts = std::make_shared<const fles::StorableTimeslice>(*timeslice);
      timeslice.reset();
    }
    else {
      ts = std::shared_ptr<const fles::StorableTimeslice>(std::move(timeslice));
    }
    for (auto& publisher : fPublishers) {
      publisher->put(ts);
    }
    if (++fCount == fOptions.fMaximumNumber) {
      break;
    }
    timeslice.reset();
  }
 
  L_(info) << fmt::format("[VtFeeder::Run] Total timeslices processed: {}", fCount);
}