web socket RFC6455 connection --asio C++11

#ifndef __APP_WEBSOCKET_CONNECTION_H__
#define __APP_WEBSOCKET_CONNECTION_H__
#include <asio.hpp>
#include "tcp_connection.hpp"

class websocket_connection : public tcp_connection
{
public:
	websocket_connection( const std::tr1::shared_ptr<asio::ip::tcp::socket> & s );
	std::tr1::shared_ptr<asio::ip::tcp::socket> get_socket() ;
	//握手应答后调用该方法
	void start_read();
	~websocket_connection();
	//buf为payload数据，write方法内部自动加上web socket frame的头部.
	void write( const std::tr1::shared_ptr<buffer> & buf);
	//服务器接受到连接时，调用该方法执行握手过程
	void start_handshake();
	//设置接收到消息的回调方法
	void set_message_handler(const OnMessage & on_message);
protected:
	//读取web socket frame 的最小长度值
	void read_min_head();
	//读取剩余的数据、在一个比较大的包中，可能会出现多次调用该方法才读到一个完整的包
	void read_leave_data( int leave_len );
private:
	shared_ptr<buffer> read_buf;
	OnMessage on_message_;

};

#endif

websocket_connection::websocket_connection( const std::tr1::shared_ptr<asio::ip::tcp::socket> & s ):tcp_connection(s){
		read_buf = get_buffer(8192);
		read_buf->size(8192);
		}
	std::tr1::shared_ptr<asio::ip::tcp::socket> websocket_connection:: get_socket() { return socket_;}

	void websocket_connection::  start_read(){
		//从包头读取
		read_min_head();
	}
	websocket_connection::~websocket_connection(){
		LOG(ERROR)<<"~websocket_connection";
	}
	void websocket_connection::write( const std::tr1::shared_ptr<buffer> & buf){

		websocket_frame frame;
		frame.fin = true;
		frame.mask = true;//使用mask、某些浏览器版本较低不能使用
		frame.opcode = 0x02;//发送的是二进制数据
		buf->offset(4);
		frame.payload = buf;
		frame.package_size();
		if( frame.payload_len == 127 ){
			LOG(ERROR)<<"ignore big message.."<<buf->capacity();
			return;
		}
		shared_ptr<buffer> frame_buf = frame.package();

		tcp_connection::write(frame_buf);

	}
void websocket_connection::read_min_head(){
		auto buf = asio::buffer( read_buf->data(),websocket_frame::fix_min_len);

		auto self(shared_from_this());

		auto read_handler = [self,this](std::error_code ec, std::size_t length){
			if( ec ){
				LOG(ERROR)<<"read_min_head error:"<<ec.message();
				if( on_error_ ){
					on_error_(ec.message());
				}
			}else{
				read_buf->size(length);
				websocket_frame frame;
				int leave_len = frame.unpakcage(read_buf);
				//读取需要解析一个frame所需要的剩余数据
				read_leave_data(leave_len);
			}
		};
		//asio 异步读取指定长度的数据,读取成功后回调read_handler方法.
		asio::async_read(*socket_, buf,read_handler);
	}
	void websocket_connection::read_leave_data( int leave_len ){

		if( read_buf->length() + leave_len > read_buf->capacity() ){
			if( on_error_ ){
				on_error_("data to big...");
			}
			return;
		}
		auto buf = asio::buffer( read_buf->data()+read_buf->length(),leave_len);
		auto self( shared_from_this());

		auto read_handler =[self,this](std::error_code ec, std::size_t length){
			if( ec ){
				if( on_error_ ){
					on_error_(ec.message());
				}
			}else{
				read_buf->size(read_buf->length() + length);
				websocket_frame frame;
				int leave_len = frame.unpakcage(read_buf);
				if( leave_len == 0 ){
					//读取到一个完整的freame了...
					if( on_message_ ){
						on_message_(frame.payload);
					}

					read_buf->size(0);
					start_read();
				}else{
					//如果不够、则继续读取剩余的字节.

					read_leave_data(leave_len);
				}
			}
		};
		asio::async_read(*socket_, buf,read_handler);
	}

void websocket_connection::start_handshake(){
	shared_ptr<buffer> buf = get_buffer(512);
	auto  asio_buf = asio::buffer( buf->data(), 512);
	auto self(shared_from_this());

	socket_->async_read_some( asio_buf,
		[self,this,buf]( const asio::error_code& error,  std::size_t bytes_transferred ){
			    if( error ){
					if( on_error_ ){
						on_error_(error.message());
					}
					recycle_buffer(buf);
					return;
			    }
				std::string data((const char*)buf->data(),bytes_transferred);
				recycle_buffer(buf);

				std::string key="Sec-WebSocket-Key:";
				auto pos = data.find(key);
				auto posEnd = data.find("\r\n",pos);
				auto value = data.substr(pos + key.length(),posEnd - (pos + key.length()));
				std::string sha1Src = trim(value);
				sha1Src += std::string("258EAFA5-E914-47DA-95CA-C5AB0DC85B11");

				unsigned char sha1out[20];
				sha1((const unsigned char *)sha1Src.c_str(),sha1Src.length(),sha1out);
				std::vector<unsigned char> data64;
				for( auto c: sha1out) data64.push_back(c);

				std::ostringstream os_rsp;
				os_rsp<<"HTTP/1.1 101 Switching Protocols\r\n"
    				  <<"Upgrade: websocket\r\n"
				      <<"Connection: Upgrade\r\n"
				      <<"Sec-WebSocket-Accept: "<<base64Encode(data64)<<"=\r\n"
					  <<"\r\n";
				std::string rsp = os_rsp.str();

				socket_->async_send(asio::buffer(rsp),[self,this](const asio::error_code& ec, std::size_t bytes_transferred){

				});
				start_read();//握手应答后，启动对web socke frame的读
			});
}

void websocket_connection::set_message_handler(const OnMessage & on_message){
	on_message_ = on_message;
}

使用；；

void init_service_admin_websocket( asio::io_service & io){

	static std::once_flag init_flag;
	std::call_once( init_flag,[&io]{
		auto new_connected_handler = []( const std::tr1::shared_ptr<asio::ip::tcp::socket> & s)
		{//新连接到来时的回调方法
			shared_ptr<websocket_connection> websocket( new websocket_connection(s));
			//设置错误发生时的回调方法
			auto on_error =[websocket](const std::string & error ){//主要清理资源
				LOG(ERROR)<<error<<" "<<websocket->name();
				sessions::clear_session(websocket);
				websocket.reset();
			};
			websocket->set_error_handler(on_error);

			std::tr1::weak_ptr<websocket_connection> wpsocket(websocket);
			//设置接收到消息时的回调方法，websocket frame的payload
			auto on_message = [wpsocket]( const std::tr1::shared_ptr<buffer> & buf)
			{
				//消息分发

			};

			websocket->set_message_handler(on_message);

			websocket->start_handshake();//启动握手

		};
		unsigned int port = 80;
	 	tcp_server * srv (new tcp_server(io, port , new_connected_handler));
	 	srv->start_accept();
		});
}

web socket RFC6455 connection --asio C++11,布布扣,bubuko.com