/*
* Copyright (C) 2013 Emeric Poupon
*
* This file is part of LMS.
*
* LMS is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* LMS 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with LMS. If not, see .
*/
#include
#include
#include
#include
#include "logger/Logger.hpp"
#include "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