refactoring to match new interfaces

src/interfaces/protocol.hpp

@@ -28,10 +28,12 @@
 #ifndef WEBSOCKET_INTERFACE_FRAME_PARSER_HPP
 #define WEBSOCKET_INTERFACE_FRAME_PARSER_HPP
 
+#include <boost/shared_ptr.hpp>
+
 namespace websocketpp {
 namespace protocol {
 
-class protocol {
+class processor {
 	// validate client handshake
 	// validate server handshake
 	
@@ -39,8 +41,9 @@ class protocol {
 	// to start a websocket session. If so begin constructing a response, if not throw a handshake 
 	// exception.
 	// validate handshake request
+	virtual void validate_handshake(const http::parser::request& headers) const = 0;
 	
-	
+	virtual void handshake_response(const http::parser::request& headers,http::parser::response& headers) = 0;
 	
 	// Given a list of HTTP headers determin if the values are a reasonable 
 	// response to our handshake request. If so 
@@ -54,6 +57,7 @@ class protocol {
 	// some sort of message type? for onping onpong?
 };
 
+typedef boost::shared_ptr<processor> processor_ptr;
 
 }
 }

src/interfaces/session.hpp

@@ -28,21 +28,217 @@
 #ifndef WEBSOCKET_INTERFACE_SESSION_HPP
 #define WEBSOCKET_INTERFACE_SESSION_HPP
 
-namespace websocketpp {
+#include <boost/shared_ptr.hpp>
+
+#include <string>
+#include <vector>
+
+#include "websocket_constants.hpp"
+#include "network_utilities.hpp"
+
+namespace websocketpp {	
 namespace session {
 
+namespace state {
+	enum value {
+		CONNECTING = 0,
+		OPEN = 1,
+		CLOSING = 2,
+		CLOSED = 3
+	};
+}
 
-// 
-class interface {
+ /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
+  *                             Server Session API                          *
+  * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+class server {
 public:
 	// Valid always
-	//virtual ??? get_state() const = 0;
+	virtual session::state::value get_state() const = 0;
+	virtual int get_version() const = 0;
 	
-	// Valid for CONNECTING connections
+	virtual std::string get_origin() const = 0;
+	virtual std::string get_request_header(const std::string& key) const = 0;
+	virtual const ws_uri& get_uri() const = 0;
+	virtual bool get_secure() const = 0;
 	
+	// Valid for CONNECTING state
+	virtual void add_response_header(const std::string& key, const std::string& value) = 0;
+	virtual void replace_response_header(const std::string& key, const std::string& value) = 0;
+	virtual const std::vector<std::string>& get_subprotocols() const = 0;
+	virtual const std::vector<std::string>& get_extensions() const = 0;
+	virtual void select_subprotocol(const std::string& value) = 0;
+	virtual void select_extension(const std::string& value) = 0;
+	
+	// Valid for OPEN state
+	virtual void send(const std::string& msg) = 0;
+	virtual void send(const std::vector<unsigned char>& data) = 0;
+	virtual void close(close::status::value code, const std::string& reason) = 0;
+	virtual void ping(const std::string& payload) = 0;
+	virtual void pong(const std::string& payload) = 0;
+	
+	// Valid for CLOSED state
+	virtual close::status::value get_local_close_code() const = 0;
+	virtual std::string get_local_close_reason() const = 0;
+	virtual close::status::value get_remote_close_code() const = 0;
+	virtual std::string get_remote_close_reason() const = 0;
+	virtual bool failed_by_me() const = 0;
+	virtual bool dropped_by_me() const = 0;
+	virtual bool closed_by_me() const = 0;
+};
+typedef boost::shared_ptr<server> server_ptr;
+
+ /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
+  *                             Server Handler API                          *
+  * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+class server_handler {
+	// validate will be called after a websocket handshake has been received and
+	// before it is accepted. It provides a handler the ability to refuse a 
+	// connection based on application specific logic (ex: restrict domains or
+	// negotiate subprotocols). To reject the connection throw a handshake_error
+	//
+	// handshake_error parameters:
+	//  log_message - error message to send to server log
+	//  http_error_code - numeric HTTP error code to return to the client
+	//  http_error_msg - (optional) string HTTP error code to return to the
+	//    client (useful for returning non-standard error codes)
+	virtual void validate(server_ptr session) = 0;
+	
+	// on_open is called after the websocket session has been successfully 
+	// established and is in the OPEN state. The session is now avaliable to 
+	// send messages and will begin reading frames and calling the on_message/
+	// on_close/on_error callbacks. A client may reject the connection by 
+	// closing the session at this point.
+	virtual void on_open(server_ptr session) = 0;
+	
+	// on_close is called whenever an open session is closed for any reason. 
+	// This can be due to either endpoint requesting a connection close or an 
+	// error occuring. Information about why the session was closed can be 
+	// extracted from the session itself. 
+	// 
+	// on_close will be the last time a session calls its handler. If your 
+	// application will need information from `session` after this function you
+	// should either save the session_ptr somewhere or copy the data out.
+	virtual void on_close(server_ptr session) = 0;
+	
+	// on_message (binary version) will be called when a binary message is 
+	// recieved. Message data is passed as a vector of bytes (unsigned char).
+	// data will not be avaliable after this callback ends so the handler must 
+	// either completely process the message or copy it somewhere else for 
+	// processing later.
+	virtual void on_message(server_ptr session,
+	                        const std::vector<unsigned char> &data) = 0;
+	
+	// on_message (text version). Identical to on_message except the data 
+	// parameter is a string interpreted as UTF-8. WebSocket++ guarantees that
+	// this string is valid UTF-8.
+	virtual void on_message(server_ptr session,const std::string &msg) = 0;
+	
+	// on_fail is called whenever a session is terminated or failed before it 
+	// was successfully established. This happens if there is an error during 
+	// the handshake process or if the server refused the connection.
+	// 
+	// on_fail will be the last time a session calls its handler. If your 
+	// application will need information from `session` after this function you
+	// should either save the session_ptr somewhere or copy the data out.
+	virtual void on_fail(server_ptr session) {};
 };
 
+typedef boost::shared_ptr<server_handler> server_handler_ptr;
 
+ /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
+  *                             Client Session API                          *
+  * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+	
+class client {
+public:
+	client(const std::string& uri) = 0;
+	
+	// Valid always
+	virtual session::state::value get_state() const = 0;
+	virtual int get_version() const = 0;
+	
+	virtual std::string get_origin() const = 0;
+	virtual const ws_uri& get_uri() const = 0;
+	virtual bool get_secure() const = 0;
+	
+	// Valid for CONNECTING state
+	virtual void set_origin(const std::string& origin) = 0;
+	virtual void add_request_header(const std::string& key, const std::string& value) = 0;
+	virtual void replace_request_header(const std::string& key, const std::string& value) = 0;
+	virtual void request_subprotocol(const std::string& value) = 0;
+	virtual void request_extension(const std::string& value) = 0;
+	
+	// Valid for OPEN state
+	virtual std::string get_response_header(const std::string& key) const = 0;
+	virtual std::string get_subprotocol() const;
+	virtual const std::vector<std::string>& get_extensions() const = 0;
+	
+	virtual void send(const std::string& msg) = 0;
+	virtual void send(const std::vector<unsigned char>& data) = 0;
+	virtual void close(close::status::value code, const std::string& reason) = 0;
+	virtual void ping(const std::string& payload) = 0;
+	virtual void pong(const std::string& payload) = 0;
+	
+	// Valid for CLOSED state
+	virtual close::status::value get_local_close_code() const = 0;
+	virtual std::string get_local_close_reason() const = 0;
+	virtual close::status::value get_remote_close_code() const = 0;
+	virtual std::string get_remote_close_reason() const = 0;
+	virtual bool failed_by_me() const = 0;
+	virtual bool dropped_by_me() const = 0;
+	virtual bool closed_by_me() const = 0;
+};
+
+typedef boost::shared_ptr<client> client_ptr;
+
+ /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
+  *                             Client Handler API                          *
+  * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
+	
+class client_handler {
+	// on_open is called after the websocket session has been successfully 
+	// established and is in the OPEN state. The session is now avaliable to 
+	// send messages and will begin reading frames and calling the on_message/
+	// on_close/on_error callbacks. A client may reject the connection by 
+	// closing the session at this point.
+	virtual void on_open(client_ptr session) = 0;
+	
+	// on_close is called whenever an open session is closed for any reason. 
+	// This can be due to either endpoint requesting a connection close or an 
+	// error occuring. Information about why the session was closed can be 
+	// extracted from the session itself. 
+	// 
+	// on_close will be the last time a session calls its handler. If your 
+	// application will need information from `session` after this function you
+	// should either save the session_ptr somewhere or copy the data out.
+	virtual void on_close(client_ptr session) = 0;
+	
+	// on_message (binary version) will be called when a binary message is 
+	// recieved. Message data is passed as a vector of bytes (unsigned char).
+	// data will not be avaliable after this callback ends so the handler must 
+	// either completely process the message or copy it somewhere else for 
+	// processing later.
+	virtual void on_message(client_ptr session,
+							const std::vector<unsigned char> &data) = 0;
+	
+	// on_message (text version). Identical to on_message except the data 
+	// parameter is a string interpreted as UTF-8. WebSocket++ guarantees that
+	// this string is valid UTF-8.
+	virtual void on_message(client_ptr session,const std::string &msg) = 0;
+	
+	// on_fail is called whenever a session is terminated or failed before it 
+	// was successfully established. This happens if there is an error during 
+	// the handshake process or if the server refused the connection.
+	// 
+	// on_fail will be the last time a session calls its handler. If your 
+	// application will need information from `session` after this function you
+	// should either save the session_ptr somewhere or copy the data out.
+	virtual void on_fail(client_ptr session) {};
+};
+
+typedef boost::shared_ptr<client_handler> client_handler_ptr;
+	
 }
 }
 #endif // WEBSOCKET_INTERFACE_SESSION_HPP

src/websocket_server.hpp

@@ -37,6 +37,9 @@ namespace po = boost::program_options;
 #include <set>
 
 #include "websocketpp.hpp"
+
+#include "interfaces/session.hpp"
+
 #include "websocket_session.hpp"
 #include "websocket_connection_handler.hpp"
 
@@ -48,30 +51,273 @@ namespace po = boost::program_options;
 #include "logger/logger.hpp"
 
 using boost::asio ::ip::tcp;
+using websocketpp::session::server_handler_ptr;
 
 namespace websocketpp {
+namespace server {
+
+template <typename server_policy>
+class session : public websocketpp::session::server, boost::enable_shared_from_this< session<server_policy> >  {
+public:
+	typedef server_policy server_type;
+	typedef session<server_policy> session_type;
+	
+	typedef boost::shared_ptr<server_type> server_ptr;
+	
+	session(server_ptr s,
+			boost::asio::io_service& io_service,
+			server_handler_ptr handler)
+	: m_server(s),
+	  m_io_service(io_service),
+	  m_socket(io_service),
+	  m_timer(io_service,boost::posix_time::seconds(0)),
+	  m_buf(/* TODO: needs a max here */),
+	  m_handler(handler) {}
+	
+	tcp::socket& get_socket() {
+		return m_socket;
+	}
+	
+	void read_request() {
+		// start reading HTTP header and attempt to determine if the incoming 
+		// connection is a websocket connection. If it is determine the version
+		// and generate a session processor for that version. If it is not a 
+		// websocket connection either drop or pass to the default HTTP pass 
+		// through handler.
+		
+		m_timer.expires_from_now(boost::posix_time::seconds(5 /* TODO */));
+		
+		m_timer.async_wait(
+		    boost::bind(
+		        &session_type::fail_on_expire,
+		        session_type::shared_from_this(),
+		        boost::asio::placeholders::error
+		    )
+		);
+		
+		boost::asio::async_read_until(
+		    m_socket,
+		    m_buf,
+		    "\r\n\r\n",
+		    boost::bind(
+		        &session_type::handle_read_request,
+		        session_type::shared_from_this(),
+		        boost::asio::placeholders::error,
+		        boost::asio::placeholders::bytes_transferred
+		    )
+		);
+	}
+	
+	void handle_read_request(const boost::system::error_code& e,
+							 std::size_t bytes_transferred) {
+		if (e) {
+			log_error("Error reading HTTP request",e);
+			drop_tcp();
+			return;
+		}
+		
+		try {
+			std::istream request(&m_buf);
+			
+			// TODO: use a more generic consume api where we just call read_some
+			// and have the handshake consume and validate as we go.
+			// 
+			// For now, because it simplifies things we will use the parse_header
+			// member function which requires the complete header to be passed in
+			// initially. ASIO can guarantee us this.
+			//
+			// 
+			//m_remote_handshake.consume(response_stream);
+			if (!m_request.parse_complete(request)) {
+				// not a valid HTTP request/response
+				throw handshake_error("Recieved invalid HTTP Request",http::status_code::BAD_REQUEST);
+			}
+			
+			// Log the raw handshake.
+			m_server->alog().at(log::alevel::DEBUG_HANDSHAKE) << m_request.raw() << log::endl;
+			
+			// Determine what sort of connection this is:
+			int m_version = -1;
+			
+			if (boost::ifind_first(m_request.header("Upgrade","websocket"))) {
+				if (handshake.header("Sec-WebSocket-Version") == "") {
+					m_version = 0;
+				} else {
+					m_version = atoi(h.c_str());
+					if (m_version == 0) {
+						throw(handshake_error("Unable to determine connection version",http::status_code::BAD_REQUEST));
+					}
+				}
+			}
+			
+			if (m_version == -1) {
+				// Probably a plain HTTP request
+				// TODO: forward to an http handler?
+			} else {
+				// websocket connection
+				// create a processor based on version.
+				if (m_version == 0) {
+					// create hybi 00 processor
+					
+					// grab hybi00 token first
+					char foo[9];
+					foo[8] = 0;
+					
+					request.get(foo,9);
+					
+					if (request.gcount() != 8) {
+						throw handshake_error("Missing Key3",http::status_code::BAD_REQUEST);
+					}
+					m_request.set_header("Sec-WebSocket-Key3",std::string(foo));
+					
+					m_processor = protocol::processor_ptr(new protocol::hybi_00_processor());
+				} else if (m_version == 7 || m_version == 8 || m_version == 13) {
+					// create hybi 17 processor
+					m_processor = protocol::processor_ptr(new protocol::hybi_17_processor());
+				} else {
+					// TODO: respond with unknown version message per spec
+				}
+				
+				// ask new protocol whether this set of headers is valid
+				m_processor->validate_handshake(m_request);
+				
+				// ask local application to confirm that it wants to accept
+				m_handler->validate(boost::static_pointer_cast<websocketpp::session::server>(session_type::shared_from_this()));
+				
+				m_response.set_status(http::status_code::SWITCHING_PROTOCOLS);
+			}
+			
+		} catch (const handshake_error& e) {
+			m_server->alog().at(log::alevel::DEBUG_HANDSHAKE) << e.what() << log::endl;
+			
+			m_server->elog().at(log::elevel::ERROR) 
+			<< "Caught handshake exception: " << e.what() << log::endl;
+			
+			m_response.set_status(e.m_http_error_code,e.m_http_error_msg);
+		}
+		
+		write_response();
+	}
+	
+	// write the response to the client's request.
+	void write_response() {
+		std::string response;
+		
+		m_response.set_version("HTTP/1.1");
+		
+		if (m_response.status_code() == http::status_code::SWITCHING_PROTOCOLS) {
+			// websocket response
+			m_processor->handshake_response(m_request,m_response);
+			
+			if (m_subprotocol != "") {
+				m_response.replace_header("Sec-WebSocket-Protocol",m_subprotocol);
+			}
+			
+			// TODO: return negotiated extensions
+		} else {
+			// HTTP response
+		}
+		
+		m_response.replace_header("Server","WebSocket++/2011-10-31");
+		
+		std::string raw = m_response.raw();
+		
+		if (m_version == 0) {
+			raw += digest;
+		}
+		
+		m_server->alog().at(log::alevel::DEBUG_HANDSHAKE) << raw << log::endl;
+		
+		// start async write to handle_write_handshake
+		boost::asio::async_write(
+		    m_socket,
+		    boost::asio::buffer(raw),
+		    boost::bind(
+		        &session_type::handle_write_response,
+		        session_type::shared_from_this(),
+		        boost::asio::placeholders::error
+		    )
+		);
+	}
+	
+	void handle_write_response(const boost::system::error_code& error) {
+		if (error) {
+			log_error("Error writing handshake response",error);
+			drop_tcp();
+			return;
+		}
+		
+		log_open_result();
+		
+		if (m_response.status_code() != http::status_code::SWITCHING_PROTOCOLS) {
+			m_server->elog().at(log::elevel::ERROR) 
+			<< "Handshake ended with HTTP error: " 
+			<< m_response.status_code() << " " << m_response.status_msg() 
+			<< log::endl;
+			
+			drop_tcp();
+			// TODO: tell client that connection failed?
+			// use on_fail?
+			return;
+		}
+		
+		m_state = state::OPEN;
+		
+		// stop the handshake timer
+		m_timer.cancel();
+		
+		m_handler->on_open(boost::static_pointer_cast<websocketpp::session::server>(session_type::shared_from_this()));
+		
+		// TODO: start read message loop.
+	}
+	
+	void fail_on_expire(const boost::system::error_code& error) {
+		if (error) {
+			if (error != boost::asio::error::operation_aborted) {
+				m_server->elog().at(log::elevel::DEVEL) 
+				<< "fail_on_expire timer ended in unknown error" << log::endl;
+				//drop_tcp(true);
+			}
+			return;
+		}
+		m_server->elog().at(log::elevel::DEVEL) 
+		<< "fail_on_expire timer expired" << log::endl;
+		drop_tcp(true);
+	}
+	
+private:
+	server_ptr					m_server;
+	
+	boost::asio::io_service&	m_io_service;
+	tcp::socket					m_socket;
+	boost::asio::deadline_timer	m_timer;
+	boost::asio::streambuf		m_buf;
+	
+	server_handler_ptr			m_handler;
+	protocol::processor_ptr		m_processor;
+				
+	http::parser::request		m_request;
+	http::parser::response		m_response;
+};
 
 
 // TODO: potential policies:
 // - http parser
-template <typename rng_policy = blank_rng, template <class> class logger_type = log::logger>
-class server : public boost::enable_shared_from_this< server<rng_policy> > {
+template <template <class> class logger_type = log::logger>
+class server : public boost::enable_shared_from_this< server<logger_type> > {
 public:
-	typedef rng_policy rng_t;
-	
-	typedef server<rng_policy> endpoint_type;
-	typedef session<endpoint_type> session_type;
-	typedef connection_handler<session_type> connection_handler_type;
+	typedef server<logger_type> endpoint_type;
+	typedef websocketpp::server::session<endpoint_type> session_type;
 	
 	typedef boost::shared_ptr<endpoint_type> ptr;
+	//typedef websocketpp::session::server_ptr session_ptr;
 	typedef boost::shared_ptr<session_type> session_ptr;
-	typedef boost::shared_ptr<connection_handler_type> connection_handler_ptr;
 	
-	server<rng_policy>(boost::asio::io_service& io_service, 
-		   const tcp::endpoint& endpoint) 
-	: m_max_message_size(DEFAULT_MAX_MESSAGE_SIZE),
-	  m_io_service(io_service), 
-	  m_acceptor(io_service, endpoint), 
+	server<logger_type>(uint16_t port, server_handler_ptr handler) 
+	: m_endpoint(tcp::v6(),port),
+	  m_acceptor(m_io_service,m_endpoint),
+	  m_handler(handler),
+	  m_max_message_size(DEFAULT_MAX_MESSAGE_SIZE),
 	  m_desc("websocketpp::server") 
 	{
 		m_desc.add_options()
@@ -81,29 +327,20 @@ public:
 		;
 	}
 	
-	void set_default_connection_handler(connection_handler_ptr c) {
-		m_def_con_handler = c;
-	}
-	
 	// creates a new session object and connects the next websocket
 	// connection to it.
 	void start_accept() {
-		if (m_def_con_handler == connection_handler_ptr()) {
-			throw server_error("start_accept called before a connection handler was set");
-		}
-		
 		// TODO: sanity check whether the session buffer size bound could be reduced
 		session_ptr new_session(
 			new session_type(
 				endpoint_type::shared_from_this(),
 				m_io_service,
-				m_def_con_handler,
-				m_max_message_size*2
+				m_handler
 			)
 		);
 		
 		m_acceptor.async_accept(
-			new_session->socket(),
+			new_session->get_socket(),
 			boost::bind(
 				&endpoint_type::handle_accept,
 				endpoint_type::shared_from_this(),
@@ -113,11 +350,6 @@ public:
 		);
 	}
 	
-	template <template <class> class T>
-	connection_handler_ptr make_handler() {
-		return boost::shared_ptr< T<session_type> >(new T<session_type>());
-	}
-	
 	// INTERFACE FOR LOCAL APPLICATIONS
 	void set_max_message_size(uint64_t val) {
 		if (val > frame::limits::PAYLOAD_SIZE_JUMBO) {
@@ -271,6 +503,8 @@ private:
 	{
 		if (!error) {
 			session->read_request();
+			
+			// TODO: add session to local session vector
 		} else {
 			std::stringstream err;
 			err << "Error accepting socket connection: " << error;
@@ -283,22 +517,24 @@ private:
 	}
 	
 private:
+	boost::asio::io_service	m_io_service;
+	tcp::endpoint			m_endpoint;
+	tcp::acceptor			m_acceptor;
+	
+	server_handler_ptr		m_handler;
+	
 	logger_type<log::alevel::value>	m_alog;
 	logger_type<log::elevel::value> m_elog;
 	
 	std::vector<session_ptr>	m_sessions;
 	
-	uint64_t					m_max_message_size;
-	boost::asio::io_service&	m_io_service;
-	tcp::acceptor				m_acceptor;
-	connection_handler_ptr		m_def_con_handler;
-	
-	rng_policy					m_rng;
+	uint64_t					m_max_message_size;	
 	
 	po::options_description		m_desc;
 	po::variables_map			m_vm;
 };
 
+}
 }
 
 #endif // WEBSOCKET_SERVER_HPP