soa/logger/compressing_output.h

/* compressing_output.h                                            -*- C++ -*-
   Jeremy Barnes, 20 September 2012
   Copyright (c) 2012 Datacratic Inc.  All rights reserved.

*/

#ifndef __logger__compressing_output_h__
#define __logger__compressing_output_h__

#include "logger.h"
#include "compressor.h"
#include "jml/utils/ring_buffer.h"
#include "jml/arch/timers.h"


namespace Datacratic {


enum FileFlushLevel {
    FLUSH_NONE,       
    FLUSH_TO_OS,      
    FLUSH_TO_DISK     
};


/*****************************************************************************/
/* WORKER THREAD OUTPUT                                                      */
/*****************************************************************************/

struct WorkerThreadOutput : public LogOutput {

    WorkerThreadOutput(size_t ringBufferSize = 65536);

    virtual ~WorkerThreadOutput();

    virtual void logMessage(const std::string & channel,
                            const std::string & message);

    virtual Json::Value stats() const;

    virtual void clearStats();

protected:
    void startWorkerThread();
    void stopWorkerThread();

    void pushOperation(const std::function<void ()> & op);

    enum MessageType {
        MT_LOG,     
        MT_END,     
        MT_OP,      
        MT_SHUTDOWN
    };

    struct Message {
        MessageType type;
        std::string channel;
        std::string contents;
        std::function<void ()> op;
    };

    ML::RingBufferSRMW<Message> ringBuffer;

    virtual void implementLogMessage(const std::string & channel,
                                     const std::string & message) = 0;

    boost::scoped_ptr<boost::thread> logThread;

    void runLogThread();

    ML::Duty_Cycle_Timer duty;

    volatile int shutdown_;
    volatile int up_;
};


/*****************************************************************************/
/* COMPRESSING OUTPUT                                                        */
/*****************************************************************************/

struct CompressingOutput : public WorkerThreadOutput {

    CompressingOutput(size_t ringBufferSize = 65536,
                      Compressor::FlushLevel compressorFlushLevel
                          = Compressor::FLUSH_AVAILABLE);

    virtual ~CompressingOutput();

    struct Sink {
        virtual ~Sink()
        {
        }

        virtual void close() = 0;
        
        virtual size_t write(const char * data, size_t size) = 0;

        virtual size_t flush(FileFlushLevel flushLevel) = 0;
    };

    void open(std::shared_ptr<Sink> sink,
              const std::string & compression,
              int compressionLevel);

    void closeCompressor();

    boost::function<void (std::string, std::size_t)> onFileWrite;

protected:
    Compressor::FlushLevel compressorFlushLevel;
    std::shared_ptr<Sink> sink;
    std::shared_ptr<Compressor> compressor;
    std::function<size_t (const char *, size_t)> onData;

    // Overrides

    virtual void implementLogMessage(const std::string & channel,
                                     const std::string & message);
};


} // namespace Datacratic


#endif /* __logger__compressing_output_h__ */