<commit_before>// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-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/>.
//

#include <sstream>
#include <boost/bind.hpp>
#include <pion/PionId.hpp>
#include <pion/net/HTTPResponse.hpp>
#include <pion/net/HTTPResponseWriter.hpp>
#include "pion/platform/Event.hpp"
#include "PlatformConfig.hpp"
#include "MonitorService.hpp"

using namespace pion::net;
using namespace pion::server;
using namespace pion::platform;


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

const std::string dtd = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>";

// MonitorHandler member functions
	
MonitorHandler::MonitorHandler(pion::platform::ReactionEngine &reaction_engine,
						 const std::string& reactor_id)
	: m_reaction_engine(reaction_engine),
	m_logger(PION_GET_LOGGER("pion.server.MonitorHandler")),
	m_connection_id(PionId().to_string()),
	m_reactor_id(reactor_id)
{}

bool MonitorHandler::sendResponse(void)
{
	// build some XML response content for the request
	std::stringstream ss;
	m_reaction_engine.writeConnectionsXML(ss, getConnectionId());
	return true;
}
	
// MonitorWriter member functions

MonitorWriter::MonitorWriter(pion::platform::ReactionEngine &reaction_engine, platform::VocabularyPtr& vptr,
					   const std::string& reactor_id, unsigned size, bool scroll)
	: MonitorHandler(reaction_engine, reactor_id),
	m_event_buffer(size), m_size(size), m_scroll(scroll), m_vocab_ptr(vptr), m_truncate(100), m_stopped(false),
	m_reaction_engine(reaction_engine)
{}
	
MonitorWriter::~MonitorWriter()
{
	PION_LOG_INFO(m_logger, "Closing output feed to " << getConnectionInfo()
				  << " (" << getConnectionId() << ')');
	stop();
}
	
void MonitorWriter::writeEvent(EventPtr& e)
{
	PION_LOG_DEBUG(m_logger, "Sending event to " << getConnectionInfo()
				   << " (" << getConnectionId() << ')');
	// lock the mutex to ensure that only one Event is sent at a time
	boost::mutex::scoped_lock send_lock(m_mutex);
	if (e.get() == NULL) {
		// Reactor is being removed -> close the connection
		// note that the ReactionEngine will remove the connection for us
		stop();
	} else {
		try {
			// Add latest event to end of circular buffer
			m_event_buffer.push_back(e);

			// If we're not scrolling, and buffer is full, then disconnect from the feed
			if (!m_scroll && m_event_buffer.full())
				stop();
		} catch (std::exception& ex) {
			// stop sending Events if we encounter an exception
			PION_LOG_WARN(m_logger, "Error sending event to " << getConnectionInfo()
						  << " (" << getConnectionId() << "):" << ex.what());
			stop();
		}
	}
}
	
void MonitorWriter::start(void)
{
	// lock the mutex to ensure that the HTTP response is sent first
	boost::mutex::scoped_lock send_lock(m_mutex);

	// tell the ReactionEngine to start sending us Events
	Reactor::EventHandler event_handler(boost::bind(&MonitorWriter::writeEvent,
													shared_from_this(), _1));
	m_reaction_engine.addTempConnectionOut(getReactorId(), getConnectionId(),
										   getConnectionInfo(),
										   event_handler);

	// send the HTTP response after the connection is successfully created
	if (! sendResponse())
		return;
	
	PION_LOG_INFO(m_logger, "Opened new output feed to " << getConnectionInfo()
				  << " (" << getConnectionId() << ')');
}

void MonitorWriter::SerializeXML(pion::platform::Vocabulary::TermRef tref,
	const pion::platform::Event::ParameterValue& value, std::ostream& xml, TermCol& cols) const
{
	if (tref > m_vocab_ptr->size())		// sanity check
		tref = Vocabulary::UNDEFINED_TERM_REF;
	const Vocabulary::Term& t((*m_vocab_ptr)[tref]);	// term corresponding with Event parameter
	// Have we seen this tref (column) yet?
	TermCol::iterator i = cols.find(tref);
	if (i == cols.end()) {
		cols[tref] = cols.size();
		i = cols.find(tref);
	}
	// Don't serialize the non-serializable
	if (t.term_type == Vocabulary::TYPE_NULL || t.term_type == Vocabulary::TYPE_OBJECT)
		xml << "<Term id=\"C" << i->second << "\"\\>";
	else {
		std::string tmp;		// tmp storage for values
		xml << "<Term id=\"C" << i->second << "\">"
			<< ConfigManager::xml_encode(Event::write(tmp, value, t).substr(0, m_truncate))
			<< "</Term>";
	}
}

std::string MonitorWriter::getStatus(void)
{
	// Map for termref -> Cnn index, we'll use it for building the guide
	TermCol col_map;
<commit_msg>MonitorService Changing the encoding of Object Type, into a pseudo-column, with the object URN term_id as the value<commit_after>// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-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/>.
//

#include <sstream>
#include <boost/bind.hpp>
#include <pion/PionId.hpp>
#include <pion/net/HTTPResponse.hpp>
#include <pion/net/HTTPResponseWriter.hpp>
#include "pion/platform/Event.hpp"
#include "PlatformConfig.hpp"
#include "MonitorService.hpp"

using namespace pion::net;
using namespace pion::server;
using namespace pion::platform;


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

const std::string dtd = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>";

// MonitorHandler member functions
	
MonitorHandler::MonitorHandler(pion::platform::ReactionEngine &reaction_engine,
						 const std::string& reactor_id)
	: m_reaction_engine(reaction_engine),
	m_logger(PION_GET_LOGGER("pion.server.MonitorHandler")),
	m_connection_id(PionId().to_string()),
	m_reactor_id(reactor_id)
{}

bool MonitorHandler::sendResponse(void)
{
	// build some XML response content for the request
	std::stringstream ss;
	m_reaction_engine.writeConnectionsXML(ss, getConnectionId());
	return true;
}
	
// MonitorWriter member functions

MonitorWriter::MonitorWriter(pion::platform::ReactionEngine &reaction_engine, platform::VocabularyPtr& vptr,
					   const std::string& reactor_id, unsigned size, bool scroll)
	: MonitorHandler(reaction_engine, reactor_id),
	m_event_buffer(size), m_size(size), m_scroll(scroll), m_vocab_ptr(vptr), m_truncate(100), m_stopped(false),
	m_reaction_engine(reaction_engine)
{}
	
MonitorWriter::~MonitorWriter()
{
	PION_LOG_INFO(m_logger, "Closing output feed to " << getConnectionInfo()
				  << " (" << getConnectionId() << ')');
	stop();
}
	
void MonitorWriter::writeEvent(EventPtr& e)
{
	PION_LOG_DEBUG(m_logger, "Sending event to " << getConnectionInfo()
				   << " (" << getConnectionId() << ')');
	// lock the mutex to ensure that only one Event is sent at a time
	boost::mutex::scoped_lock send_lock(m_mutex);
	if (e.get() == NULL) {
		// Reactor is being removed -> close the connection
		// note that the ReactionEngine will remove the connection for us
		stop();
	} else {
		try {
			// Add latest event to end of circular buffer
			m_event_buffer.push_back(e);

			// If we're not scrolling, and buffer is full, then disconnect from the feed
			if (!m_scroll && m_event_buffer.full())
				stop();
		} catch (std::exception& ex) {
			// stop sending Events if we encounter an exception
			PION_LOG_WARN(m_logger, "Error sending event to " << getConnectionInfo()
						  << " (" << getConnectionId() << "):" << ex.what());
			stop();
		}
	}
}
	
void MonitorWriter::start(void)
{
	// lock the mutex to ensure that the HTTP response is sent first
	boost::mutex::scoped_lock send_lock(m_mutex);

	// tell the ReactionEngine to start sending us Events
	Reactor::EventHandler event_handler(boost::bind(&MonitorWriter::writeEvent,
													shared_from_this(), _1));
	m_reaction_engine.addTempConnectionOut(getReactorId(), getConnectionId(),
										   getConnectionInfo(),
										   event_handler);

	// send the HTTP response after the connection is successfully created
	if (! sendResponse())
		return;
	
	PION_LOG_INFO(m_logger, "Opened new output feed to " << getConnectionInfo()
				  << " (" << getConnectionId() << ')');
}

void MonitorWriter::SerializeXML(pion::platform::Vocabulary::TermRef tref,
	const pion::platform::Event::ParameterValue& value, std::ostream& xml, TermCol& cols) const
{
	if (tref > m_vocab_ptr->size())		// sanity check
		tref = Vocabulary::UNDEFINED_TERM_REF;
	const Vocabulary::Term& t((*m_vocab_ptr)[tref]);	// term corresponding with Event parameter
	// Have we seen this tref (column) yet?
	if (t.term_type == Vocabulary::TYPE_OBJECT)
		tref = Vocabulary::UNDEFINED_TERM_REF;			// Mask OBJECT to undef
	TermCol::iterator i = cols.find(tref);
	if (i == cols.end()) {
		cols[tref] = cols.size();
		i = cols.find(tref);
	}
	// Don't serialize the non-serializable
	if (t.term_type == Vocabulary::TYPE_NULL)
		xml << "<Term id=\"C" << i->second << "\"\\>";
	if (t.term_type == Vocabulary::TYPE_OBJECT)
		xml << "<Term id=\"C" << i->second << "\">" << t.term_id << "</Term>";
	else {
		std::string tmp;		// tmp storage for values
		xml << "<Term id=\"C" << i->second << "\">"
			<< ConfigManager::xml_encode(Event::write(tmp, value, t).substr(0, m_truncate))
			<< "</Term>";
	}
}

std::string MonitorWriter::getStatus(void)
{
	// Map for termref -> Cnn index, we'll use it for building the guide
	TermCol col_map;

{'fraction_non_alphanumeric': 0.15918443696221474, 'fraction_numerical': 0.0031799476243920687, 'mean_word_length': 3.8275395033860047, 'pattern_counts': {'":': 0, '<': 135, '<?xml version=': 2, '>': 38, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 11}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5919970', 'n_tokens_mistral': 3537, 'n_tokens_neox': 3244, 'n_words': 941}