Communication works without error messages (7cd4071a) · Commits · IA2 / metadata_exporter

src/DBManager.cpp

+7 −0

Original line number	Diff line number	Diff line
		@@ -103,6 +103,13 @@ DBManager::InformationList DBManager::retrieveInformation(std::string schema,

		std::copy(rows.begin(), rows.end(), std::back_inserter(informationList));

		if(informationList.empty())
		{
		std::stringstream errorStream;
		errorStream << schema << "." << table << " not exists";
		throw soci::soci_error(errorStream.str());
		}

		return informationList;
		}

src/PlainSession.cpp

+24 −11

Original line number	Diff line number	Diff line
		#include <PlainSession.h>

		#include <boost/bind.hpp>

		#include <netinet/in.h>
		#include <boost/lexical_cast.hpp>

		namespace MetadataExporter_ns
		{
		@@ -26,6 +25,9 @@ PlainSession::~PlainSession()
		{
		DEBUG_STREAM << "PlainSession::~PlainSession()" << endl;

		INFO_STREAM << "PlainSession::~PlainSession() Disconnection from "
		<< remoteEndpoint << endl;

		boost::system::error_code errorCode;

		m_plainSocket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, errorCode);
		@@ -63,7 +65,11 @@ void PlainSession::start()
		{
		DEBUG_STREAM << "PlainSession::start()" << endl;

		INFO_STREAM << m_plainSocket.remote_endpoint() << " CONNECTED" << endl;
		remoteEndpoint = boost::lexical_cast<std::string>(
		m_plainSocket.remote_endpoint());

		INFO_STREAM << "PlainSession::start() Connection from "
		<< remoteEndpoint << endl;

		startReadRequestHeader();
		}
		@@ -89,14 +95,16 @@ void PlainSession::startReadRequestBody(boost::uint32_t bodySize)
		{
		DEBUG_STREAM << "PlainSession::startReadRequestBody()" << endl;

		INFO_STREAM << m_plainSocket.remote_endpoint()
		<< " >>>> " << bodySize << " BYTE" << endl;

		m_readBuff.resize(HEADER_SIZE + bodySize);

		boost::asio::mutable_buffers_1 mutableBuffer =
		boost::asio::buffer(&m_readBuff[HEADER_SIZE], bodySize);

		#ifdef VERBOSE_DEBUG
		INFO_STREAM << "PlainSession::startReadRequestBody() "
		<< remoteEndpoint << " >>>> " << bodySize << " BYTE" << endl;
		#endif

		boost::asio::async_read(m_plainSocket, mutableBuffer,
		m_strand.wrap(boost::bind(&PlainSession::handleReadRequestBody,
		shared_from_this(), boost::asio::placeholders::error)));
		@@ -119,9 +127,6 @@ void PlainSession::startWriteResponse()

		boost::uint32_t bodySize = response_sp->ByteSize();

		INFO_STREAM << m_plainSocket.remote_endpoint()
		<< " <<<< " << bodySize << " BYTE" << endl;

		std::vector<boost::uint8_t> writeBuff;
		writeBuff.resize(HEADER_SIZE + bodySize);

		@@ -129,17 +134,25 @@ void PlainSession::startWriteResponse()

		response_sp->SerializeToArray(&writeBuff[HEADER_SIZE], bodySize);

		#ifdef VERBOSE_DEBUG
		INFO_STREAM << "PlainSession::startWriteResponse() "
		<< remoteEndpoint << " <<<< " << bodySize << " byte" << endl;
		#endif

		boost::asio::async_write(m_plainSocket, boost::asio::buffer(writeBuff),
		m_strand.wrap(boost::bind(&PlainSession::handleWriteResponse,
		shared_from_this(), boost::asio::placeholders::error)));
		}
		catch(std::runtime_error& ec)
		{
		ERROR_STREAM << "SSLSession::startWriteResponse() " << ec.what() << endl;
		ERROR_STREAM << "SSLSession::startWriteResponse() "
		<< ec.what() << " from " << remoteEndpoint << endl;
		}
		catch(...)
		{
		ERROR_STREAM << "SSLSession::startWriteResponse() unknown error" << endl;
		ERROR_STREAM << "SSLSession::startWriteResponse() unknown error from "
		<< remoteEndpoint << endl;

		}
		}

src/ProtocolManager.cpp

+23 −23

Original line number	Diff line number	Diff line
		@@ -150,23 +150,22 @@ ResponseSP ProtocolManager::prepareValidation(RequestSP request_sp)
		const std::string& schema = validationReq.schema();
		const std::string& table = validationReq.table();

		try
		{
		DBManager::InformationList informationList =
		m_dBManager_sp->retrieveInformation(schema, table);

		if(validationReq.columns_size() == (int)informationList.size())
		{
		if(validationReq.columns_size() != (int)informationList.size())
		throw std::runtime_error("Columns number does not match");

		const google::protobuf::RepeatedPtrField
		< Request::Validation::Column >& columns = validationReq.columns();

		google::protobuf::RepeatedPtrField
		< Request::Validation::Column >::const_iterator it;

		try
		{
		for(it=columns.begin(); it!=columns.end(); ++it)
		{
		validateColumn(*it, informationList);
		}

		m_isValidated = true;

		@@ -180,12 +179,6 @@ ResponseSP ProtocolManager::prepareValidation(RequestSP request_sp)
		}
		}
		else
		{
		validationRes->set_state(Response::Validation::REJECTED);
		validationRes->set_status("Columns number does not match");
		}
		}
		else
		{
		WARN_STREAM << "ProtocolManager::prepareValidation() "
		<< "Already validated" << endl;
		@@ -292,6 +285,13 @@ void ProtocolManager::validateColumn(const Request::Validation::Column& column,
		<< "server " << isNullable << " client " << column.nullable();
		throw std::runtime_error(errorStream.str());
		}

		#ifdef VERBOSE_DEBUG
		INFO_STREAM << "SERVER: " << columnName << " \| " << columnType
		<< " \| " << isNullable << endl;
		INFO_STREAM << "SERVER: " << column.name() << " \| "
		<< column.type() << " \| " << column.nullable() << endl;
		#endif
		}
		}

src/SSLSession.cpp

+23 −11

Original line number	Diff line number	Diff line
		#include <SSLSession.h>

		#include <boost/bind.hpp>
		#include <boost/lexical_cast.hpp>

		namespace MetadataExporter_ns
		{
		@@ -24,6 +26,9 @@ SSLSession::~SSLSession()
		{
		DEBUG_STREAM << "SSLSession::~SSLSession()" << endl;

		INFO_STREAM << "SSLSession::~SSLSession() Disconnection from "
		<< remoteEndpoint << endl;

		boost::system::error_code errorCode;

		m_sslSocket.shutdown(errorCode);
		@@ -65,8 +70,10 @@ void SSLSession::start()
		{
		DEBUG_STREAM << "SSLSession::start()" << endl;

		INFO_STREAM << m_sslSocket.lowest_layer().remote_endpoint()
		<< " CONNECTED" << endl;
		remoteEndpoint = boost::lexical_cast<std::string>(
		m_sslSocket.lowest_layer().remote_endpoint());

		INFO_STREAM << "SSLSession::start() Connection from " << remoteEndpoint << endl;

		startHandShake();
		}
		@@ -96,9 +103,8 @@ void SSLSession::handleHandShake(const boost::system::error_code& errorCode)
		}
		else
		{
		WARN_STREAM << "SSLSession::handleHandShake() error "
		<< errorCode.message() << " from "
		<< m_sslSocket.lowest_layer().remote_endpoint() << endl;
		ERROR_STREAM << "SSLSession::handleHandShake() error "
		<< errorCode.message() << " from " << remoteEndpoint << endl;
		}
		}

		@@ -128,8 +134,10 @@ void SSLSession::startReadRequestBody(boost::uint32_t bodySize)
		boost::asio::mutable_buffers_1 mutableBuffer =
		boost::asio::buffer(&m_readBuff[HEADER_SIZE], bodySize);

		INFO_STREAM << m_sslSocket.lowest_layer().remote_endpoint()
		<< " >>>> " << bodySize << " BYTE" << endl;
		#ifdef VERBOSE_DEBUG
		INFO_STREAM << "SSLSession::startReadRequestBody() "
		<< remoteEndpoint << " >>>> " << bodySize << " byte" << endl;
		#endif

		boost::asio::async_read(m_sslSocket, mutableBuffer,
		m_strand.wrap(boost::bind(&SSLSession::handleReadRequestBody,
		@@ -161,8 +169,10 @@ void SSLSession::startWriteResponse()

		response_sp->SerializeToArray(&writeBuff[HEADER_SIZE], bodySize);

		INFO_STREAM << m_sslSocket.lowest_layer().remote_endpoint()
		<< " <<<< " << bodySize << " BYTE" << endl;
		#ifdef VERBOSE_DEBUG
		INFO_STREAM << "SSLSession::startWriteResponse() "
		<< remoteEndpoint << " <<<< " << bodySize << " byte" << endl;
		#endif

		boost::asio::async_write(m_sslSocket, boost::asio::buffer(writeBuff),
		m_strand.wrap(boost::bind(&SSLSession::handleWriteResponse,
		@@ -170,11 +180,13 @@ void SSLSession::startWriteResponse()
		}
		catch(std::runtime_error& ec)
		{
		ERROR_STREAM << "SSLSession::startWriteResponse() " << ec.what() << endl;
		ERROR_STREAM << "SSLSession::startWriteResponse() "
		<< ec.what() << " from " << remoteEndpoint << endl;
		}
		catch(...)
		{
		ERROR_STREAM << "SSLSession::startWriteResponse() unknown error" << endl;
		ERROR_STREAM << "SSLSession::startWriteResponse() unknown error from "
		<< remoteEndpoint << endl;
		}
		}

src/Server.cpp

+11 −2

Original line number	Diff line number	Diff line
		@@ -69,7 +69,13 @@ void Server::start() throw(std::runtime_error)
		std::string localHost = m_configuration_sp->getLocalHost();
		unsigned int localPort = m_configuration_sp->getLocalPort();

		INFO_STREAM << "LISTENING ON " << localHost << ":" << localPort << endl;
		std::stringstream infoStream;
		infoStream << "Listening on " << localHost << ":" << localPort << endl;

		INFO_STREAM << "Server::start()" << infoStream.str() << endl;

		writeState(Tango::ON);
		writeStatus(infoStream.str());

		boost::asio::ip::tcp::resolver::query query(localHost,
		boost::lexical_cast<std::string>(localPort));
		@@ -101,6 +107,9 @@ void Server::stop() throw(std::runtime_error)
		{
		DEBUG_STREAM << "Server::stop()" << endl;

		writeState(Tango::OFF);
		writeStatus("Disconnected");

		boost::system::error_code errorCode;
		m_acceptor_sp->close(errorCode);

		@@ -181,7 +190,7 @@ void Server::handleAccept(Session::SP session_sp,
		}
		else
		{
		WARN_STREAM << "Server::handleAccept() " << ec.message() << endl;
		ERROR_STREAM << "Server::handleAccept() " << ec.message() << endl;
		}

		startAccept();