redis_ipc_stream_writer.hpp

// redis_ipc_stream_writer.hpp
// Created by Greg on 1/30/25.
// Copyright (C) 2025 Presage Security, Inc.
//
// SPDX-License-Identifier: LGPL-3.0-or-later
 
#pragma once
 
#include <atomic>
#include <memory>
#include <mutex>
#include <string>
#include <vector>
 
#include "../ipc_stream_writer.hpp"
#include "redis_ipc_configuration.hpp"
#include "redis_client.hpp"
 
namespace presage::smartspectra::redis_ipc {

template<container::settings::IntegrationMode TIntegrationMode>
class RedisIpcStreamWriter : public ipc::IpcStreamWriter<TIntegrationMode> {
public:
    using Base = ipc::IpcStreamWriter<TIntegrationMode>;
    using typename Base::BackgroundContainer;
 
    explicit RedisIpcStreamWriter(const RedisIpcConfig& configuration);
    virtual ~RedisIpcStreamWriter();
 
    // Implement base interface
    absl::Status WriteStatus(const presage::physiology::StatusValue& status) override;
    absl::Status WriteCoreMetrics(const presage::physiology::MetricsBuffer& metrics, int64_t timestamp_us) override;
    absl::Status WriteEdgeMetrics(const presage::physiology::Metrics& metrics, int64_t timestamp_us) override;
    absl::Status WriteFrame(const cv::Mat& frame, int64_t timestamp_us) override;
    void Close() override;
    bool IsOperational() const override;
    absl::Status Flush() override;

    absl::Status PublishRecordingState(bool recording);
 
protected:
    absl::Status Initialize();

    std::string BuildKey(const std::string& suffix) const;

    std::string BuildChannel(const std::string& suffix) const;
 
private:
 
 
 
    RedisIpcConfig configuration_;
    RedisClient redis_client_;
 
    // Thread safety
    mutable std::mutex write_mutex_;
    std::atomic<bool> is_operational_{false};
 
    // Error tracking with rate limiting
    std::atomic<int> publish_error_count_{0};
    static constexpr int kMaxErrorLogs = 5;
};
 
} // namespace presage::smartspectra::redis_ipc