WIP, Remote Client/Server...

remote/server/Connection.cpp

@@ -1,7 +1,11 @@
 
 #include <utility>
 #include <vector>
+
 #include <boost/bind.hpp>
+#include <boost/foreach.hpp>
+
+#include "messages/messages.pb.h"
 
 #include "RequestHandler.hpp"
 
@@ -18,6 +22,7 @@ Connection::Connection(boost::asio::ip::tcp::socket socket,
 _connectionManager(manager),
 _requestHandler(handler)
 {
+	std::cout << "Server::Connection::Connection, Creating connection" << std::endl;
 }
 
 boost::asio::ip::tcp::socket&
@@ -29,29 +34,59 @@ Connection::socket()
 void
 Connection::start()
 {
-  _socket.async_read_some(boost::asio::buffer(_headerBuffer),
-      boost::bind(&Connection::handleRead, shared_from_this(),
-        boost::asio::placeholders::error,
-        boost::asio::placeholders::bytes_transferred));
+	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()
 {
-  _socket.close();
+	std::cout << "Server::Connection::stop, Stopping connection" << std::endl;
+	_socket.close();
 }
 
 void
-Connection::handleRead(const boost::system::error_code& error, std::size_t bytes_transferred)
+Connection::handleReadHeader(const boost::system::error_code& error, std::size_t bytes_transferred)
 {
 	if (!error)
 	{
-/*			boost::asio::async_write(_socket, reply_.to_buffers(),
-					boost::bind(&Connection::handleWrite, shared_from_this(),
-						boost::asio::placeholders::error));*/
+		if (bytes_transferred != Remote::Header::size)
+		{
+			std::cerr << "bytes_transferred (" << bytes_transferred << ") != Remote::Header::size!" << std::endl;
+			_connectionManager.stop(shared_from_this());
+			return;
+		}
 
+		_inputStreamBuf.commit(bytes_transferred);
+
+		std::istream is(&_inputStreamBuf);
+
+		Remote::Header header;
+		if (!header.from_istream(is))
+		{
+			std::cerr << "Cannot read header from buffer!" << std::endl;
+			_connectionManager.stop(shared_from_this());
+			return;
+		}
+
+		std::cout << "Header received. Size = " << header.getSize() << std::endl;
+
+		// Now read the real message
+		boost::asio::streambuf::mutable_buffers_type bufs = _inputStreamBuf.prepare(header.getSize());
+
+		boost::asio::async_read(_socket,
+				bufs,
+				boost::asio::transfer_exactly(header.getSize()),
+				boost::bind(&Connection::handleReadMsg, shared_from_this(),
+					boost::asio::placeholders::error,
+					boost::asio::placeholders::bytes_transferred));
 
-		start();
 	}
 	else if (error != boost::asio::error::operation_aborted)
 	{
@@ -60,21 +95,90 @@ Connection::handleRead(const boost::system::error_code& error, std::size_t bytes
 	}
 }
 
-void Connection::handleWrite(const boost::system::error_code& error)
+void
+Connection::handleReadMsg(const boost::system::error_code& error, std::size_t bytes_transferred)
 {
 	if (!error)
 	{
-		// Initiate graceful Connection closure.
-		boost::system::error_code ignored_ec;
-		_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
-	}
+		_inputStreamBuf.commit(bytes_transferred);
 
-	if (error != boost::asio::error::operation_aborted)
+		std::istream is(&_inputStreamBuf);
+		std::ostream os(&_outputStreamBuf);
+
+		std::vector<Remote::ServerMessage> responses;
+		Remote::ClientMessage request;
+
+		if (!request.ParseFromIstream(&is))
+		{
+			std::cerr << "Cannot parse request!" << std::endl;
+			_connectionManager.stop(shared_from_this());
+			return;
+		}
+
+		if (!_requestHandler.process(request, responses))
+		{
+			std::cerr << "Cannot process request!" << std::endl;
+			_connectionManager.stop(shared_from_this());
+			return;
+		}
+
+		BOOST_FOREACH(const Remote::ServerMessage& response, responses)
+		{
+			boost::system::error_code	ec;
+
+			if (!response.SerializeToOstream(&os))
+			{
+				std::cerr << "Cannot serialize to ostream!" << std::endl;
+				_connectionManager.stop(shared_from_this());
+				return;
+			}
+
+			std::array<unsigned char, Remote::Header::size> headerBuffer;
+			{
+				Remote::Header header;
+				header.setSize(_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)
+			{
+				std::cerr << "cannot write header: " << error.message() << std::endl;
+				_connectionManager.stop(shared_from_this());
+			}
+
+			// Now send serialized payload
+			n = boost::asio::write(_socket,
+					_outputStreamBuf.data(),
+					boost::asio::transfer_exactly(_outputStreamBuf.size()),
+					ec);
+
+			_outputStreamBuf.consume(n);
+			assert(n == _outputStreamBuf.size());
+
+			if (ec)
+			{
+				std::cerr << "cannot write msg: " << error.message() << std::endl;
+				_connectionManager.stop(shared_from_this());
+			}
+		}
+		start();
+
+		// Initiate graceful Connection closure.
+		//		boost::system::error_code ignored_ec;
+		//		_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
+
+	}
+	else if (error != boost::asio::error::operation_aborted)
 	{
-		std::cerr << "Connection::handleWrite: " << error.message() << std::endl;
+		std::cerr << "Connection::handleRead: " << error.message() << std::endl;
 		_connectionManager.stop(shared_from_this());
 	}
 }
 
+
 } // namespace Server
 } // namespace Remote