soa/logger/logger.h

/* logger.h                                                        -*- C++ -*-
   Jeremy Barnes, 20 May 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

*/

#pragma once

#include "soa/service/zmq_named_pub_sub.h"
#include "soa/service/zmq_utils.h"
#include "soa/service/socket_per_thread.h"
#include <sstream>
#include "jml/utils/filter_streams.h"
#include <boost/thread/thread.hpp>
#include "jml/utils/smart_ptr_utils.h"
#include "jml/utils/vector_utils.h"
#include "jml/arch/atomic_ops.h"
#include "ace/Synch.h"
#include <boost/function.hpp>
#include <boost/regex.hpp>
#include <boost/shared_ptr.hpp>
#include "soa/jsoncpp/json.h"


namespace Datacratic {


/*****************************************************************************/
/* LOG OUTPUT                                                                */
/*****************************************************************************/

struct LogOutput {
    LogOutput()
    {
    }

    virtual ~LogOutput()
    {
    }

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

    virtual void close() = 0;

    virtual Json::Value stats() const;

    virtual void clearStats();
};


/*****************************************************************************/
/* LOGGER                                                                    */
/*****************************************************************************/

struct Logger {

    Logger();

    Logger(zmq::context_t & contextRef);

    Logger(std::shared_ptr<zmq::context_t> & context);

    ~Logger();

    void init();

    void subscribe(const std::string & uri,
                   const std::vector<std::string> & channels,
                   const std::string & identity = "");

    void logTo(const std::string & uri,
               const boost::regex & allowChannels = boost::regex(),
               const boost::regex & denyChannels = boost::regex(),
               double logProbability = 1.0);

    void addOutput(std::shared_ptr<LogOutput> output,
                   const boost::regex & allowChannels = boost::regex(),
                   const boost::regex & denyChannels = boost::regex(),
                   double logProbability = 1.0);

    void addCallback(boost::function<void (std::string, std::string)> callback,
                     const boost::regex & allowChannels = boost::regex(),
                     const boost::regex & denyChannels = boost::regex(),
                     double logProbability = 1.0);

    void clearOutputs();

    template<typename... Args>
    void operator () (const std::string & channel, Args... args)
    {
        logMessage(channel, args...);
    }

    template<typename... Args>
    void logMessage(const std::string & channel, Args... args)
    {
        if (!outputs) return;
        ML::atomic_add(messagesSent, 1);
        messages.push({ channel, Date::now().print(5), args... });
    }

    template<typename... Args>
    void logMessageNoTimestamp(const std::string & channel, Args... args)
    {
        if (!outputs) return;
        ML::atomic_add(messagesSent, 1);
        messages.push({ channel, args... });
    }

    void logMessageNoTimestamp(const std::vector<std::string> & message)
    {
        if (!outputs) return;

        if (message.empty())
            throw ML::Exception("can't log empty message");

        ML::atomic_add(messagesSent, 1);
        messages.push(message);
    }

    template<typename GetEl>
    void logMessage(const std::string & channel,
                    int numElements,
                    GetEl getElement)
    {
        if (!outputs) return;

        std::vector<std::string> message;
        message.push_back(channel);
        message.push_back(Date::now().print(5));

        for (unsigned i = 0;  i < numElements;  ++i) {
            message.push_back(getElement(i));
        }

        messages.push(message);
    }

    void start(std::function<void ()> onStop = 0);

    void waitUntilFinished();

    void shutdown();
    
    std::map<std::string, size_t> getStats();
    void resetStats();

    void replay(const std::string & filename,
                ssize_t maxEvents = -1);

    void replayDirect(const std::string & filename,
                      ssize_t maxEvents = -1) const;
    
    uint64_t numMessagesSent() const { return messagesSent; }
    uint64_t numMessagesDone() const { return messagesDone; }

    void handleListenerMessage(std::vector<std::string> const & message);
    void handleRawListenerMessage(std::vector<std::string> const & message);
    void handleMessage(std::vector<zmq::message_t> && message);

    MessageLoop messageLoop;

private:
    std::shared_ptr<zmq::context_t> context;

    std::map<std::string, size_t> stats;

    TypedMessageSink<std::vector<std::string>> messages;

#if 0

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

    void runLogThread(ACE_Semaphore & sem);
#endif

    struct Output;
    struct Outputs;

    Outputs * outputs;

    std::vector<std::shared_ptr<zmq::socket_t> > subscriptions;

    bool doShutdown;

    mutable uint64_t messagesSent;

    uint64_t messagesDone;

};


} // namespace Datacratic