Port Forwarder built with boost::asio

aioforwarder.cpp

#include <memory>
#include <functional>
#include <queue>
#include <iostream>
#include <boost/asio.hpp>

using boost::system::error_code;
using boost::asio::ip::tcp;
using boost::asio::mutable_buffer;

#define BIND_HANDLER(handler) ([this, self=shared_from_this()](error_code ec){handler(ec);})
#define BIND_HANDLER_S(handler) ([this, self=shared_from_this()](error_code ec, std::size_t length){handler(ec, length);})

class session : public std::enable_shared_from_this<session>
{
public:
	explicit session(tcp::socket socket) :
		dst_socket(socket.get_io_service()),
		src_socket(std::move(socket))
	{
		std::cout << "{ create object 0x" << this << std::endl;
	}

	~session()
	{
		std::cout << "} delete object 0x" << this << std::endl;
	}

	void start(tcp::endpoint remote)
	{
		dst_socket.async_connect(remote, BIND_HANDLER(dst_connect_handler));
	}

private:
	void dst_connect_handler(error_code ec)
	{
		if (ec)
		{
			src_socket.close();
			dst_socket.close();
			return;
		}
		src_socket.async_read_some(boost::asio::buffer(src_read_buffer, read_buffer_length), BIND_HANDLER_S(src_read_some_handler));
		dst_socket.async_read_some(boost::asio::buffer(dst_read_buffer, read_buffer_length), BIND_HANDLER_S(dst_read_some_handler));
	}

	void src_read_some_handler(error_code ec, std::size_t length)
	{
		if (ec)
		{
			src_socket.close();
			dst_socket.close();
			return;
		}
		enqueue_dst_write(length);
		src_socket.async_read_some(boost::asio::buffer(src_read_buffer, read_buffer_length), BIND_HANDLER_S(src_read_some_handler));
	}
	
	void dst_read_some_handler(error_code ec, std::size_t length)
	{
		if (ec)
		{
			src_socket.close();
			dst_socket.close();
			return;
		}
		enqueue_src_write(length);
		dst_socket.async_read_some(boost::asio::buffer(dst_read_buffer, read_buffer_length), BIND_HANDLER_S(dst_read_some_handler));
	}

	void src_write_handler(error_code ec, std::size_t)
	{
		write_buffer_type buffer = src_write_queue.front();
		src_write_queue.pop();
		std::cout << "] deallocating buffer 0x" << buffer.first << "[" << buffer.second << "]" << std::endl;
		boost::asio::asio_handler_deallocate(buffer.first, buffer.second);
		if (ec)
		{
			src_socket.close();
			dst_socket.close();
			return;
		}
	}

	void dst_write_handler(error_code ec, std::size_t)
	{
		write_buffer_type buffer = dst_write_queue.front();
		dst_write_queue.pop();
		std::cout << "] deallocating buffer 0x" << buffer.first << "[" << buffer.second << "]" << std::endl;
		boost::asio::asio_handler_deallocate(buffer.first, buffer.second);
		if (ec)
		{
			src_socket.close();
			dst_socket.close();
			return;
		}
	}

	void enqueue_src_write(std::size_t length)
	{
		void *buffer_data = boost::asio::asio_handler_allocate(length, &BIND_HANDLER_S(src_write_handler));
		std::cout << "[ allocated buffer 0x" << buffer_data << "[" << length << "]" << std::endl;
		//std::copy(dst_read_buffer, dst_read_buffer + length, reinterpret_cast<char*>(buffer_data));
		std::memcpy(buffer_data, dst_read_buffer, length);
		write_buffer_type buffer(buffer_data, length);
		src_write_queue.push(buffer);
		boost::asio::async_write(src_socket, boost::asio::buffer(buffer_data, length), BIND_HANDLER_S(src_write_handler));
	}

	void enqueue_dst_write(std::size_t length)
	{
		void *buffer_data = boost::asio::asio_handler_allocate(length, &BIND_HANDLER_S(dst_write_handler));
		std::cout << "[ allocated buffer 0x" << buffer_data << "[" << length << "]" << std::endl;
		//std::copy(src_read_buffer, src_read_buffer + length, reinterpret_cast<char*>(buffer_data));
		std::memcpy(buffer_data, src_read_buffer, length);
		write_buffer_type buffer(buffer_data, length);
		dst_write_queue.push(buffer);
		boost::asio::async_write(dst_socket, boost::asio::buffer(buffer_data, length), BIND_HANDLER_S(dst_write_handler));
	}

	tcp::socket src_socket, dst_socket;

	static const std::size_t read_buffer_length = 1500;
	char src_read_buffer[read_buffer_length], dst_read_buffer[read_buffer_length];
	std::size_t write_buffer_length = read_buffer_length;
	typedef std::pair<void*, size_t> write_buffer_type;
	std::queue<write_buffer_type> src_write_queue, dst_write_queue;
};

class server
{
public:
	server(boost::asio::io_service &io_service, short port) :
		acceptor_(io_service, tcp::endpoint(tcp::v4(), port)),
		socket_(io_service)
	{
		do_accept();
	}

private:
	void do_accept()
	{
		acceptor_.async_accept(socket_, [this](boost::system::error_code ec)
		{
			if (!ec)
			{
				auto s = std::make_shared<session>(std::move(socket_));
				s->start(tcp::endpoint(boost::asio::ip::address_v4::from_string("202.202.43.42"), 80));
			}
			do_accept();
		});
	}

	tcp::acceptor acceptor_;
	tcp::socket socket_;
};

int main()
{
	boost::asio::io_service io_service;
	boost::asio::ip::tcp::socket socket(io_service);
	//socket.connect()
	server s(io_service, 2333);
	io_service.run();
	return 0;
}