add simple http server. fix bug in tcp::socket::close preventing sock…

…ets from being reused

Jamfile

@@ -52,6 +52,7 @@ SOURCES =
 	queue
 	acceptor
 	default_config
+	http_server
 	;
 
 lib simulator

include/simulator/http_server.hpp

@@ -0,0 +1,70 @@
+/*
+
+Copyright (c) 2015, Arvid Norberg
+All rights reserved.
+
+    This program 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.
+    This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+#include "simulator/simulator.hpp"
+
+namespace sim
+{
+
+	std::string send_response(int code, char const* status_message
+		, int len = 0, char const** extra_header = NULL);
+
+// This is a very simple http server that only supports a single concurrent
+// connection
+struct http_server
+{
+	http_server(asio::io_service& ios, int listen_port);
+
+	void stop();
+
+	typedef std::function<std::string (std::string, std::string)> handler_t;
+	void register_handler(std::string path, handler_t const& h);
+
+private:
+
+	void on_accept(boost::system::error_code const& ec);
+	void read();
+	void on_read(boost::system::error_code const& ec, size_t bytes_transferred);
+	void on_write(boost::system::error_code const& ec, size_t bytes_transferred);
+	void close_connection();
+
+	asio::io_service& m_ios;
+
+	asio::ip::tcp::acceptor m_listen_socket;
+
+	asio::ip::tcp::socket m_connection;
+	asio::ip::tcp::endpoint m_ep;
+
+	std::unordered_map<std::string, handler_t> m_handlers;
+
+	// read buffer, we receive bytes into this buffer for the connection
+	std::string m_recv_buffer;
+
+	// the number of bytes of m_recv_buffer that we've actually read data into.
+	// The remaining is uninitialized, possibly being read into in an async call
+	int m_bytes_used;
+
+	std::string m_send_buffer;
+
+	// set to true when shutting down
+	bool m_close;
+};
+
+}
+
+

include/simulator/simulator.hpp

@@ -108,7 +108,6 @@ namespace sim
 		std::shared_ptr<sink> last() const
 		{ return hops.back(); }
 
-
 	private:
 		std::vector<std::shared_ptr<sink>> hops;
 	};
@@ -172,6 +171,7 @@ namespace sim
 	using boost::asio::mutable_buffer;
 	using boost::asio::const_buffers_1;
 	using boost::asio::mutable_buffers_1;
+	using boost::asio::buffer;
 
 	struct io_service;
 
@@ -840,6 +840,7 @@ namespace sim
 			boost::system::error_code cancel(boost::system::error_code& ec);
 			void cancel();
 
+			void listen(int qs = -1);
 			void listen(int qs, boost::system::error_code& ec);
 
 			void async_accept(ip::tcp::socket& peer

src/acceptor.cpp

@@ -40,8 +40,17 @@ namespace ip {
 		close(ec);
 	}
 
+	void tcp::acceptor::listen(int qs)
+	{
+		boost::system::error_code ec;
+		listen(qs, ec);
+		if (ec) throw boost::system::system_error(ec);
+	}
+
 	void tcp::acceptor::listen(int qs, boost::system::error_code& ec)
 	{
+		if (qs == -1) qs = 20;
+
 		if (!m_open)
 		{
 			ec = error::bad_descriptor;
@@ -145,19 +154,30 @@ namespace ip {
 
 	void tcp::acceptor::incoming_packet(aux::packet p)
 	{
-		fprintf(stderr, "acceptor incoming packet (p: %" PRIu64 ")\n"
-			, p.seq_nr);
-		if (p.type != aux::packet::syn)
+		switch (p.type)
 		{
-			assert(false);
-			// TODO: not sure why this would happen. It would be nice to respond
-			// with a reset
-			// ignore for now
-			return;
+			case aux::packet::syn:
+				m_incoming_queue.push_back(p.channel);
+				check_accept_queue();
+				return;
+			case aux::packet::error:
+				assert(false); // something is not wired up correctly
+				if (m_accept_handler)
+				{
+					m_io_service.post(std::bind(m_accept_handler
+						, boost::system::error_code(error::operation_aborted)));
+					m_accept_handler = 0;
+					m_accept_into = NULL;
+					m_remote_endpoint = NULL;
+				}
+				return;
+			default:
+				assert(false);
+				// TODO: not sure why this would happen. It would be nice to respond
+				// with a reset
+				// ignore for now
+				return;
 		}
-
-		m_incoming_queue.push_back(p.channel);
-		check_accept_queue();
 	}
 
 	void tcp::acceptor::check_accept_queue()
@@ -205,6 +225,7 @@ namespace ip {
 		if (m_remote_endpoint) *m_remote_endpoint = c->ep[0];
 
 		boost::system::error_code ec;
+		// if the acceptor socket is closed. Any potential socket in the queue
 		m_accept_into->internal_connect(m_bound_to, c, ec);
 
 		// notify the other end

src/http_server.cpp

@@ -0,0 +1,202 @@
+/*
+
+Copyright (c) 2015, Arvid Norberg
+All rights reserved.
+
+    This program 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.
+    This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+#include "simulator/simulator.hpp"
+#include "simulator/http_server.hpp"
+
+#include <functional>
+
+using namespace sim::asio;
+using namespace sim::asio::ip;
+using namespace std::placeholders;
+
+using boost::system::error_code;
+
+namespace sim
+{
+	std::string send_response(int code, char const* status_message
+		, int len, char const** extra_header)
+	{
+		char msg[600];
+		int pkt_len = snprintf(msg, sizeof(msg), "HTTP/1.1 %d %s\r\n"
+			"content-length: %d\r\n"
+			"%s"
+			"%s"
+			"%s"
+			"%s"
+			"\r\n"
+			, code, status_message, len
+			, extra_header ? extra_header[0] : ""
+			, extra_header ? extra_header[1] : ""
+			, extra_header ? extra_header[2] : ""
+			, extra_header ? extra_header[3] : "");
+		return std::string(msg, pkt_len);
+	}
+
+	http_server::http_server(io_service& ios, int listen_port)
+		: m_ios(ios)
+		, m_listen_socket(ios)
+		, m_connection(ios)
+		, m_bytes_used(0)
+		, m_close(false)
+	{
+		m_listen_socket.open(tcp::v4());
+
+		m_listen_socket.bind(tcp::endpoint(address_v4(), listen_port));
+		m_listen_socket.listen();
+
+		m_listen_socket.async_accept(m_connection, m_ep
+			, std::bind(&http_server::on_accept, this, _1));
+	}
+
+	void http_server::on_accept(error_code const& ec)
+	{
+		if (ec)
+		{
+			printf("http_server::on_accept: (%d) %s\n"
+				, ec.value(), ec.message().c_str());
+			close_connection();
+			return;
+		}
+
+		printf("http_server accepted connection from: %s : %d\n",
+			m_ep.address().to_string().c_str(), m_ep.port());
+
+		read();
+	}
+
+	void http_server::register_handler(std::string path, handler_t const& h)
+	{
+		m_handlers[path] = h;
+	}
+
+	void http_server::read()
+	{
+		if (m_bytes_used >= m_recv_buffer.size() / 2)
+		{
+			m_recv_buffer.resize((std::max)(500, m_bytes_used * 2));
+		}
+		assert(m_recv_buffer.size() > m_bytes_used);
+		m_connection.async_read_some(asio::mutable_buffers_1(&m_recv_buffer[m_bytes_used]
+				, m_recv_buffer.size() - m_bytes_used)
+			, std::bind(&http_server::on_read, this, _1, _2));
+	}
+
+	void http_server::on_read(error_code const& ec, size_t bytes_transferred)
+	{
+		if (ec)
+		{
+			printf("http_server::on_read: (%d) %s\n"
+				, ec.value(), ec.message().c_str());
+			close_connection();
+			return;
+		}
+
+		m_bytes_used += bytes_transferred;
+
+		char const* end_of_request = strstr(m_recv_buffer.c_str(), "\r\n\r\n");
+		if (end_of_request == NULL)
+		{
+			read();
+			return;
+		}
+		const int req_len = end_of_request - m_recv_buffer.c_str() + 4;
+
+		char const* space = strstr(m_recv_buffer.c_str(), " ");
+		if (space == NULL)
+		{
+			printf("http_server: failed to parse request:\n%s\n"
+				, m_recv_buffer.c_str());
+			close_connection();
+			return;
+		}
+
+		char const* space2 = strstr(space + 1, " ");
+		if (space2 == NULL)
+		{
+			printf("http_server: failed to parse request:\n%s\n"
+				, m_recv_buffer.c_str());
+			close_connection();
+			return;
+		}
+		std::string method(m_recv_buffer.c_str(), space);
+		std::string req(space+1, space2);
+		std::string path(req.substr(0, req.find_first_of('?')));
+		printf("http_server: incoming request: %s %s [%s]\n"
+			, method.c_str(), path.c_str(), req.c_str());
+
+		auto it = m_handlers.find(path);
+		if (it == m_handlers.end())
+		{
+			// no handler found, 404
+			m_send_buffer = send_response(404, "Not Found");
+		}
+		else
+		{
+			m_send_buffer = it->second(method, req);
+		}
+
+		m_recv_buffer.erase(m_recv_buffer.begin(), m_recv_buffer.begin() + req_len);
+
+		async_write(m_connection, asio::const_buffers_1(m_send_buffer.data()
+			, m_send_buffer.size()), std::bind(&http_server::on_write
+			, this, _1, _2));
+	}
+
+	void http_server::on_write(error_code const& ec, size_t bytes_transferred)
+	{
+		if (ec)
+		{
+			printf("http_server::on_write: (%d) %s\n"
+				, ec.value(), ec.message().c_str());
+			close_connection();
+			return;
+		}
+
+		// try to read another request out of the buffer
+		m_ios.post(std::bind(&http_server::on_read, this, error_code(), 0));
+	}
+
+	void http_server::stop()
+	{
+		m_close = true;
+		m_listen_socket.close();
+	}
+
+	void http_server::close_connection()
+	{
+		m_recv_buffer.clear();
+		m_bytes_used = 0;
+
+		error_code err;
+		m_connection.close(err);
+		if (err)
+		{
+			printf("http_server::close: failed to close connection (%d) %s\n"
+				, err.value(), err.message().c_str());
+			return;
+		}
+
+		if (m_close) return;
+
+		// now we can accept another connection
+		m_listen_socket.async_accept(m_connection, m_ep
+			, std::bind(&http_server::on_accept, this, _1));
+	}
+}
+