StreamWriter.h

Go to the documentation of this file.

 
#ifndef __ROGUE_UTILITIES_FILEIO_STREAM_WRITER_H__
#define __ROGUE_UTILITIES_FILEIO_STREAM_WRITER_H__
#include "rogue/Directives.h"
 
#include <stdint.h>
 
#include <chrono>
#include <condition_variable>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
 
#include "rogue/EnableSharedFromThis.h"
#include "rogue/Logging.h"
#include "rogue/interfaces/stream/Frame.h"
 
namespace rogue {
namespace utilities {
namespace fileio {
 
class StreamWriterChannel;
 
class StreamWriter : public rogue::EnableSharedFromThis<rogue::utilities::fileio::StreamWriter> {
    friend class StreamWriterChannel;
 
  protected:
    // Log
    std::shared_ptr<rogue::Logging> log_;
 
    // Active output file descriptor.
    int32_t fd_;
 
    // Base output file path.
    std::string baseName_;
 
    // True when writer is currently open for output.
    bool isOpen_;
 
    // Current file index for rollover mode.
    uint32_t fdIdx_;
 
    // Auto-rollover size limit in bytes. Zero disables rollover.
    uint64_t sizeLimit_;
 
    // Bytes written to current file.
    uint64_t currSize_;
 
    // Total bytes written across all files since open().
    uint64_t totSize_;
 
    // Write-buffer size in bytes. Zero disables buffering.
    uint32_t buffSize_;
 
    // Write buffer storage.
    uint8_t* buffer_;
 
    // Number of bytes currently staged in write buffer.
    uint32_t currBuffer_;
 
    // Drop frames with non-zero error field when true.
    bool dropErrors_;
 
    // Lock for writer state and counters.
    std::mutex mtx_;
 
    // Number of frames written since open().
    uint32_t frameCount_;
 
    // Byte total in sliding one-second bandwidth window.
    uint64_t bandwidthBytes_;
 
    // Time-tagged write samples for bandwidth calculation.
    std::deque<std::pair<std::chrono::steady_clock::time_point, uint32_t>> bandwidthHistory_;
 
    // Writes bytes with optional staging buffer.
    void intWrite(void* data, uint32_t size);
 
    // Applies max-size rollover checks before write.
    void checkSize(uint32_t size);
 
    // Flushes staged bytes to file descriptor.
    void flush();
 
    // Adds a write sample to bandwidth statistics.
    void recordBandwidth(uint32_t size);
 
    // Removes samples older than one second.
    void pruneBandwidth(std::chrono::steady_clock::time_point now);
 
    // Writes payload-only format when true.
    bool raw_;
 
    // Condition variable for frame-count waiters.
    std::condition_variable cond_;
 
    std::map<uint32_t, std::shared_ptr<rogue::utilities::fileio::StreamWriterChannel>> channelMap_;
 
    // Writes one frame to file. Called by StreamWriterChannel.
    virtual void writeFile(uint8_t channel, std::shared_ptr<rogue::interfaces::stream::Frame> frame);
 
  public:
    static std::shared_ptr<rogue::utilities::fileio::StreamWriter> create();
 
    static void setup_python();
 
    StreamWriter();
 
    virtual ~StreamWriter();
 
    void open(std::string file);
 
    void close();
 
    bool isOpen();
 
    void setRaw(bool raw);
 
    bool getRaw();
 
    void setBufferSize(uint32_t size);
 
    void setMaxSize(uint64_t size);
 
    void setDropErrors(bool drop);
 
    std::shared_ptr<rogue::utilities::fileio::StreamWriterChannel> getChannel(uint8_t channel);
 
    uint64_t getTotalSize();
 
    uint64_t getCurrentSize();
 
    double getBandwidth();
 
    uint32_t getFrameCount();
 
    bool waitFrameCount(uint32_t count, uint64_t timeout);
};
 
// Convenience
typedef std::shared_ptr<rogue::utilities::fileio::StreamWriter> StreamWriterPtr;
}  // namespace fileio
}  // namespace utilities
}  // namespace rogue
#endif