#include #include #include #include #include "logger/Logger.hpp" #include "messages/messages.pb.h" #include "RequestHandler.hpp" #include "ConnectionManager.hpp" #include "Connection.hpp" namespace Remote { namespace Server { Connection::Connection(boost::asio::io_service& ioService, boost::asio::ssl::context& context, ConnectionManager& manager, const boost::filesystem::path& dbPath) : _closing(false), _socket(ioService, context), _connectionManager(manager), _requestHandler(dbPath) { LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Server::Connection::Connection, Creating connection"; } void Connection::start() { LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Starting connection..."; _socket.async_handshake(boost::asio::ssl::stream_base::server, boost::bind(&Connection::handleHandshake, this, boost::asio::placeholders::error)); } void Connection::handleHandshake(const boost::system::error_code& error) { if (!error) { LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Handshake successfully performed... Now reading messages"; readMsg(); } else if (error != boost::asio::error::operation_aborted) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Connection::handleHandshake: " << error.message(); _connectionManager.stop(shared_from_this()); } else LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Handshake error: " << error.message(); } void Connection::readMsg() { // Read a header first boost::asio::streambuf::mutable_buffers_type bufs = _inputStreamBuf.prepare(Remote::Header::size); boost::asio::async_read(_socket, bufs, boost::asio::transfer_exactly(Remote::Header::size), boost::bind(&Connection::handleReadHeader, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } void Connection::stop() { if (!_closing) { boost::system::error_code ec; _closing = true; LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Server::Connection::stop, Stopping connection " << this; _socket.shutdown(ec); if (ec) LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Error while shutting down connection " << this << ": " << ec.message(); LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Server::Connection::stop, connection stopped " << this; } else LMS_LOG(MOD_REMOTE, SEV_DEBUG) << "Stop: close already in progress..."; } void Connection::handleReadHeader(const boost::system::error_code& error, std::size_t bytes_transferred) { if (!error) { if (bytes_transferred != Remote::Header::size) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "bytes_transferred (" << bytes_transferred << ") != Remote::Header::size!"; _connectionManager.stop(shared_from_this()); return; } _inputStreamBuf.commit(bytes_transferred); std::istream is(&_inputStreamBuf); Remote::Header header; if (!header.from_istream(is)) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Cannot read header from buffer!"; _connectionManager.stop(shared_from_this()); return; } // Now read the real message payload boost::asio::streambuf::mutable_buffers_type bufs = _inputStreamBuf.prepare(header.getDataSize()); boost::asio::async_read(_socket, bufs, boost::asio::transfer_exactly(header.getDataSize()), boost::bind(&Connection::handleReadMsg, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } else if (error != boost::asio::error::operation_aborted) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Connection::handleReadHeader: " << error.message(); _connectionManager.stop(shared_from_this()); } } void Connection::handleReadMsg(const boost::system::error_code& error, std::size_t bytes_transferred) { if (!error) { _inputStreamBuf.commit(bytes_transferred); std::istream is(&_inputStreamBuf); std::ostream os(&_outputStreamBuf); Remote::ServerMessage response; Remote::ClientMessage request; if (!request.ParseFromIstream(&is)) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Parse request failed!"; _connectionManager.stop(shared_from_this()); return; } if (!_requestHandler.process(request, response)) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Process request failed!"; _connectionManager.stop(shared_from_this()); return; } { boost::system::error_code ec; if (!response.SerializeToOstream(&os)) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Cannot serialize to ostream!"; _connectionManager.stop(shared_from_this()); return; } if (_outputStreamBuf.size() >= Remote::Header::max_data_size) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "output message is too big! " << _outputStreamBuf.size() << " > " << Remote::Header::max_data_size; _connectionManager.stop(shared_from_this()); return; } std::array headerBuffer; { Remote::Header header; header.setDataSize(_outputStreamBuf.size()); header.to_buffer(headerBuffer); } std::size_t n = boost::asio::write(_socket, boost::asio::buffer(headerBuffer), boost::asio::transfer_exactly(Remote::Header::size), ec); if (ec) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "cannot write header: " << error.message(); _connectionManager.stop(shared_from_this()); } else { assert(n == Remote::Header::size); } // Now send serialized payload n = boost::asio::write(_socket, _outputStreamBuf.data(), boost::asio::transfer_exactly(_outputStreamBuf.size()), ec); if (ec) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "cannot write msg: " << error.message(); _connectionManager.stop(shared_from_this()); } else { assert(n == _outputStreamBuf.size()); _outputStreamBuf.consume(n); } } // All good here, read another message readMsg(); } else if (error != boost::asio::error::operation_aborted) { LMS_LOG(MOD_REMOTE, SEV_ERROR) << "Connection::handleRead: " << error.message(); _connectionManager.stop(shared_from_this()); } } } // namespace Server } // namespace Remote