BufferedFileReader.hpp

Go to the documentation of this file.

#ifndef DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEREADER_HPP_
#define DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEREADER_HPP_
 
#include "datahandlinglibs/DataHandlingIssues.hpp"
#include "datahandlinglibs/ReadoutLogging.hpp"
 
#include "logging/Logging.hpp"
 
#include <boost/align/aligned_allocator.hpp>
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/filter/lzma.hpp>
#include <boost/iostreams/filter/zlib.hpp>
#include <boost/iostreams/filter/zstd.hpp>
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/stream_buffer.hpp>
 
#include <fcntl.h>
#include <fstream>
#include <iostream>
#include <limits>
#include <string>
#include <unistd.h>
 
using dunedaq::datahandlinglibs::logging::TLVL_WORK_STEPS;
 
namespace dunedaq {
namespace datahandlinglibs {
template<class ReadoutType, size_t Alignment = 4096>
class BufferedFileReader
{
  using io_source_t = boost::iostreams::file_descriptor_source;
  using aligned_allocator_t = boost::alignment::aligned_allocator<io_source_t::char_type, Alignment>;
  using filtering_istream_t =
    boost::iostreams::filtering_stream<boost::iostreams::input, char, std::char_traits<char>, aligned_allocator_t>;
 
public:
  BufferedFileReader(std::string filename, size_t buffer_size, std::string compression_algorithm = "None")
  {
    open(filename, buffer_size, compression_algorithm);
  }

  BufferedFileReader() {}
 
  BufferedFileReader(const BufferedFileReader&) = delete;            
  BufferedFileReader& operator=(const BufferedFileReader&) = delete; 
  BufferedFileReader(BufferedFileReader&&) = delete;                 
  BufferedFileReader& operator=(BufferedFileReader&&) = delete;      

  void open(std::string filename, size_t buffer_size, std::string compression_algorithm = "None")
  {
    m_filename = filename;
    m_buffer_size = buffer_size;
    m_compression_algorithm = compression_algorithm;
 
    int fd = ::open(m_filename.c_str(), O_RDONLY);
    if (fd == -1) {
      throw BufferedReaderWriterCannotOpenFile(ERS_HERE, m_filename);
    }
 
    io_source_t io_source(fd, boost::iostreams::file_descriptor_flags::close_handle);
    if (m_compression_algorithm == "zstd") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using zstd compression" << std::endl;
      m_input_stream.push(boost::iostreams::zstd_decompressor());
    } else if (m_compression_algorithm == "lzma") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using lzma compression" << std::endl;
      m_input_stream.push(boost::iostreams::lzma_decompressor());
    } else if (m_compression_algorithm == "zlib") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Using zlib compression" << std::endl;
      m_input_stream.push(boost::iostreams::zlib_decompressor());
    } else if (m_compression_algorithm == "None") {
      TLOG_DEBUG(TLVL_WORK_STEPS) << "Running without compression" << std::endl;
    } else {
      throw BufferedReaderWriterConfigurationError(ERS_HERE,
                                                   "Non-recognized compression algorithm: " + m_compression_algorithm);
    }
 
    m_input_stream.push(io_source, m_buffer_size);
    m_is_open = true;
  }

  bool is_open() const { return m_is_open; }

  bool read(ReadoutType& element)
  {
    if (!m_is_open)
      return false;
    m_input_stream.read(reinterpret_cast<char*>(&element), sizeof(element)); // NOLINT
    return (m_input_stream.gcount() == sizeof(element));
  }

  void close()
  {
    m_input_stream.reset();
    m_is_open = false;
  }
 
private:
  // Config parameters
  std::string m_filename;
  size_t m_buffer_size;
  std::string m_compression_algorithm;
 
  // Internals
  filtering_istream_t m_input_stream;
  bool m_is_open = false;
};
 
} // namespace datahandlinglibs
} // namespace dunedaq
 
#endif // DATAHANDLINGLIBS_INCLUDE_DATAHANDLINGLIBS_UTILS_BUFFEREDFILEREADER_HPP_