platform/services/MonitorService.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2010 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_MONITORSERVICE_HEADER__
#define __PION_MONITORSERVICE_HEADER__

#include <string>
#include <iosfwd>
#include <boost/date_time/gregorian/gregorian.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/circular_buffer.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/unordered_map.hpp>
#include <pion/net/HTTPTypes.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionLogger.hpp>
#include <pion/net/TCPStream.hpp>
#include <pion/platform/Codec.hpp>
#include <pion/platform/Reactor.hpp>
#include <pion/platform/ReactionEngine.hpp>
#include "PlatformService.hpp"


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


class MonitorWriter
    : public boost::enable_shared_from_this<MonitorWriter>
{
    typedef std::set<pion::platform::Vocabulary::TermRef>   TermRefSet;

    typedef boost::circular_buffer<pion::platform::EventPtr> EventBuffer;

private:
    pion::platform::ReactionEngine &    m_reaction_engine;

    PionLogger                          m_logger;   

    const std::string                   m_connection_id;

    const std::string                   m_reactor_id;

    mutable boost::mutex                m_mutex;

    EventBuffer                         m_event_buffer;

    unsigned                            m_size;

    bool                                m_scroll;

    platform::VocabularyPtr             m_vocab_ptr;

    unsigned                            m_truncate;

    volatile bool                       m_stopped;

    bool                                m_hide_all;

    TermRefSet                          m_show_terms;

    TermRefSet                          m_suppressed_terms;

    TermRefSet                          m_terms_seen;

    TermRefSet                          m_events_seen;

    TermRefSet                          m_filtered_events;

    boost::uint32_t                     m_event_counter;
    boost::uint32_t                     m_change_counter;

    boost::posix_time::ptime            m_age;

public:

    typedef boost::unordered_map<pion::platform::Vocabulary::TermRef, unsigned> TermCol;
    
    ~MonitorWriter()
    {
//      PION_LOG_INFO(m_logger, "Closing output feed to " << getConnectionId());
        stop();
//      PION_LOG_INFO(m_logger, "Closing output feed #2 to " << getConnectionId());
        boost::mutex::scoped_lock send_lock(m_mutex);
        m_event_buffer.clear();
//      PION_LOG_INFO(m_logger, "Closing output feed #3 to " << getConnectionId());
    }

    MonitorWriter(pion::platform::ReactionEngine &reaction_engine, platform::VocabularyPtr& vptr,
               const std::string& reactor_id, unsigned size, bool scroll, PionLogger logger)
        : m_reaction_engine(reaction_engine), m_logger(logger), m_connection_id(PionId().to_string()), m_reactor_id(reactor_id),
        m_event_buffer(size), m_size(size), m_scroll(scroll), m_vocab_ptr(vptr), m_truncate(100), m_stopped(false),
        m_hide_all(false), m_event_counter(0), m_change_counter(0)
    { }
    
    void writeEvent(pion::platform::EventPtr& e);

    void setQP(const pion::net::HTTPTypes::QueryParams& qp);
    
    void start(const pion::net::HTTPTypes::QueryParams& qp);

    void stop(bool Stop = true, bool Flush = false) {
        if (m_stopped == false && Stop == true)
            m_reaction_engine.removeTempConnection(getConnectionId());
        m_stopped = true;
        if (Flush) {
            m_event_buffer.clear();
            m_age = boost::date_time::not_a_date_time;
        }
    }

    void SerializeXML(pion::platform::Vocabulary::TermRef tref,
        const pion::platform::Event::ParameterValue& value,
        std::ostream& xml, TermCol& cols);

    inline const std::string& getConnectionId(void) const { return m_connection_id; }

    inline const std::string& getReactorId(void) const { return m_reactor_id; }

    std::string getStatus(const pion::net::HTTPTypes::QueryParams& qp);

    boost::posix_time::ptime getAge(void) const { return m_age; }
    void setAge(void) { m_age = boost::posix_time::second_clock::local_time(); }
};

typedef boost::shared_ptr<MonitorWriter>    MonitorWriterPtr;



class MonitorService
    : public pion::server::PlatformService
{
    PionLogger                          m_logger;   

    std::vector<MonitorWriterPtr>       m_writers;

    volatile bool                       m_running;

public:
    
    MonitorService(void)
        : PlatformService("pion.MonitorService"),
        m_logger(PION_GET_LOGGER("pion.MonitorService")),
        m_writers(WRITERS),                                 // a default of ten simultaneous monitors allowed
        m_running(true)
    { }
    
    virtual ~MonitorService()
    {
        if (m_running) {
            m_running = false;
//          PION_LOG_INFO(m_logger, "MonitorService: starting shut down");
            for (unsigned i = 0; i < m_writers.size(); i++)
                if (m_writers[i] != NULL)
                    m_writers[i]->stop();
            m_writers.clear();
//          PION_LOG_INFO(m_logger, "MonitorService: Done");
        }
    }
    
    virtual void operator()(pion::net::HTTPRequestPtr& request,
                            pion::net::TCPConnectionPtr& tcp_conn);

    std::string getPermissionType(void) const { return MONITOR_SERVICE_PERMISSION_TYPE; }

private:

    static const std::string            MONITOR_SERVICE_PERMISSION_TYPE;
    static const unsigned               WRITERS;
};

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

#endif

---