platform/reactors/LogInputReactor.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2008 Atomic Labs, Inc.  (http://www.atomiclabs.com)
//
// Pion is free software: you can redistribute it and/or modify it under the
// terms of the GNU Affero General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// Pion is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
// FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for
// more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with Pion.  If not, see <http://www.gnu.org/licenses/>.
//

#ifndef __PION_LOGINPUTREACTOR_HEADER__
#define __PION_LOGINPUTREACTOR_HEADER__

#include <set>
#include <string>
#include <boost/asio.hpp>
#include <boost/regex.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/condition.hpp>
#include <boost/iostreams/filtering_stream.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionLogger.hpp>
#include <pion/PionException.hpp>
#include <pion/platform/Event.hpp>
#include <pion/platform/Codec.hpp>
#include <pion/platform/Reactor.hpp>


namespace pion {        // begin namespace pion
namespace plugins {     // begin namespace plugins


class LogInputReactor :
    public pion::platform::Reactor
{
public:

    class EmptyCodecException : public PionException {
    public:
        EmptyCodecException(const std::string& reactor_id)
            : PionException("LogInputReactor configuration is missing a required Codec parameter: ", reactor_id) {}
    };

    class EmptyFilenameException : public PionException {
    public:
        EmptyFilenameException(const std::string& reactor_id)
            : PionException("LogInputReactor configuration is missing a required Filename parameter: ", reactor_id) {}
    };
    
    class EmptyDirectoryException : public PionException {
    public:
        EmptyDirectoryException(const std::string& reactor_id)
            : PionException("LogInputReactor configuration is missing a required Directory parameter: ", reactor_id) {}
    };
    
    class DirectoryNotFoundException : public PionException {
    public:
        DirectoryNotFoundException(const std::string& dir)
            : PionException("LogInputReactor directory not found: ", dir) {}
    };
    
    class NotADirectoryException : public PionException {
    public:
        NotADirectoryException(const std::string& dir)
            : PionException("LogInputReactor Directory parameter is not a directory: ", dir) {}
    };
    
    class BadFrequencyException : public PionException {
    public:
        BadFrequencyException(const std::string& dir)
            : PionException("LogInputReactor frequency must be greater than zero: ", dir) {}
    };
    
    class OpenLogException : public PionException {
    public:
        OpenLogException(const std::string& log_filename)
            : PionException("Unable to open log file for reading: ", log_filename) {}
    };

    class EmptyLogException : public PionException {
    public:
        EmptyLogException(const std::string& log_filename)
            : PionException("Log file is empty: ", log_filename) {}
    };
    
    class ReadEventException : public PionException {
    public:
        ReadEventException(const std::string& log_filename)
            : PionException("Unable to read event from log file: ", log_filename) {}
    };

    
    LogInputReactor(void)
        : Reactor(TYPE_COLLECTION),
        m_logger(PION_GET_LOGGER("pion.LogInputReactor")),
        m_just_one(false), m_tail_f(false), m_frequency(DEFAULT_FREQUENCY), m_worker_is_active(false)
    {
        // publish the "FinishedLog" signal
        publish("FinishedLog");
    }
    
    virtual ~LogInputReactor() { stop(); }
    
    virtual void setConfig(const pion::platform::Vocabulary& v, const xmlNodePtr config_ptr);
    
    virtual void query(std::ostream& out, const QueryBranches& branches,
        const QueryParams& qp);
    
    virtual void start(void);
    
    virtual void stop(void);
    
    inline void setLogger(PionLogger log_ptr) { m_logger = log_ptr; }
    
    inline PionLogger getLogger(void) { return m_logger; }
    
        
private:

    typedef std::set<std::string>       LogFileCollection;

    typedef std::pair<boost::shared_ptr<boost::iostreams::filtering_istream>, 
                      boost::shared_ptr<boost::uint64_t> >          StreamData;

    typedef std::map<std::string, StreamData>                       StreamMap;

    void scheduleLogFileCheck(boost::uint32_t seconds);
    
    void checkForLogFiles(void);

    void recordLogFileAsDone(void);

    inline void scheduleReadFromLog(void) {
        getScheduler().getIOService().post(boost::bind(&LogInputReactor::readFromLog, this));
    }
    
    void readFromLog(void);
    
    void getLogFilesInLogDirectory(LogFileCollection& files);
    
    inline void deliverEventFromLog(const pion::platform::EventPtr& e, bool return_immediately = false) {
        incrementEventsIn();
        ConfigReadLock cfg_lock(*this);
        deliverEvent(e, return_immediately);
    }

    inline void finishWorkerThread(void) {
        boost::mutex::scoped_lock worker_lock(m_worker_mutex);
        PION_LOG_DEBUG(m_logger, "Log reader thread has finished: " << getId());
        m_worker_is_active = false;
        m_worker_stopped.notify_all();
    }

    
    static const boost::uint32_t        DEFAULT_FREQUENCY;
    
    static const std::string            CODEC_ELEMENT_NAME;

    static const std::string            DIRECTORY_ELEMENT_NAME;

    static const std::string            FILENAME_ELEMENT_NAME;

    static const std::string            JUST_ONE_ELEMENT_NAME;

    static const std::string            TAIL_F_ELEMENT_NAME;

    static const std::string            FREQUENCY_ELEMENT_NAME;

    static const std::string            CURRENT_LOG_ELEMENT_NAME;

    static const std::string            CONSUMED_LOG_ELEMENT_NAME;

    
    PionLogger                          m_logger;
    
    std::string                         m_codec_id;
    
    bool                                m_just_one;

    bool                                m_tail_f;

    boost::uint32_t                     m_frequency;

    std::string                         m_log_directory;

    boost::regex                        m_log_regex;
    
    LogFileCollection                   m_logs_consumed;
    
    std::string                         m_log_file;

    StreamData                          m_current_stream_data;

    StreamMap                           m_open_streams;

    boost::scoped_ptr<boost::asio::deadline_timer>  m_timer_ptr;

    std::string                         m_history_cache_filename;

    std::string                         m_current_log_file_cache_filename;

    std::map<std::string, boost::uint64_t>  m_num_events_read_previously;

    boost::mutex                        m_logs_consumed_mutex;

    boost::mutex                        m_worker_mutex;

    boost::condition                    m_worker_stopped;
    
    volatile bool                       m_worker_is_active;
};


}   // end namespace plugins
}   // end namespace pion

#endif

---