ZLMediaKit/src/Http/WebSocketSession.h

243 lines
8.1 KiB
C++
Raw Normal View History

2018-12-20 10:31:31 +08:00
/*
2020-04-04 20:30:09 +08:00
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
2018-12-20 10:31:31 +08:00
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
2018-12-20 10:31:31 +08:00
*
2020-04-04 20:30:09 +08:00
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
2018-12-20 10:31:31 +08:00
*/
#ifndef ZLMEDIAKIT_WEBSOCKETSESSION_H
#define ZLMEDIAKIT_WEBSOCKETSESSION_H
#include "HttpSession.h"
2019-11-30 11:44:05 +08:00
#include "Network/TcpServer.h"
2018-12-20 10:31:31 +08:00
/**
*
*/
class SendInterceptor{
public:
2021-01-19 16:05:38 +08:00
typedef function<ssize_t (const Buffer::Ptr &buf)> onBeforeSendCB;
SendInterceptor() = default;
virtual ~SendInterceptor() = default;
virtual void setOnBeforeSendCB(const onBeforeSendCB &cb) = 0;
};
/**
* TcpSession派生类发送数据的截取
* websocket协议的打包
*/
template <typename TcpSessionType>
class TcpSessionTypeImp : public TcpSessionType, public SendInterceptor{
public:
typedef std::shared_ptr<TcpSessionTypeImp> Ptr;
TcpSessionTypeImp(const Parser &header, const HttpSession &parent, const Socket::Ptr &pSock) :
TcpSessionType(pSock), _identifier(parent.getIdentifier()) {}
~TcpSessionTypeImp() {}
/**
*
* @param cb
*/
void setOnBeforeSendCB(const onBeforeSendCB &cb) override {
_beforeSendCB = cb;
}
protected:
/**
* send函数截取数据
* @param buf
* @return
*/
2021-01-19 16:05:38 +08:00
ssize_t send(Buffer::Ptr buf) override {
if (_beforeSendCB) {
return _beforeSendCB(buf);
}
return TcpSessionType::send(std::move(buf));
}
string getIdentifier() const override {
return _identifier;
}
private:
string _identifier;
onBeforeSendCB _beforeSendCB;
};
template <typename TcpSessionType>
class TcpSessionCreator {
public:
//返回的TcpSession必须派生于SendInterceptor可以返回null
TcpSession::Ptr operator()(const Parser &header, const HttpSession &parent, const Socket::Ptr &pSock){
return std::make_shared<TcpSessionTypeImp<TcpSessionType> >(header,parent,pSock);
}
};
2018-12-20 10:31:31 +08:00
/**
* WebSocket协议
* WebSock协议下的具体业务协议WebSocket协议的Rtmp协议等
*/
template<typename Creator, typename HttpSessionType = HttpSession, WebSocketHeader::Type DataType = WebSocketHeader::TEXT>
class WebSocketSessionBase : public HttpSessionType {
2018-12-20 10:31:31 +08:00
public:
WebSocketSessionBase(const Socket::Ptr &pSock) : HttpSessionType(pSock){}
virtual ~WebSocketSessionBase(){}
2018-12-20 10:31:31 +08:00
//收到eof或其他导致脱离TcpServer事件的回调
void onError(const SockException &err) override{
HttpSessionType::onError(err);
if(_session){
_session->onError(err);
}
}
//每隔一段时间触发,用来做超时管理
void onManager() override{
if(_session){
_session->onManager();
}else{
HttpSessionType::onManager();
}
}
void attachServer(const TcpServer &server) override{
HttpSessionType::attachServer(server);
_weak_server = const_cast<TcpServer &>(server).shared_from_this();
2018-12-20 10:31:31 +08:00
}
2018-12-20 10:31:31 +08:00
protected:
2019-09-16 17:42:52 +08:00
/**
* websocket客户端连接上事件
* @param header http头
* @return true代表允许websocket连接
*/
bool onWebSocketConnect(const Parser &header) override{
//创建websocket session类
_session = _creator(header, *this,HttpSessionType::getSock());
if(!_session){
//此url不允许创建websocket连接
return false;
}
auto strongServer = _weak_server.lock();
2019-09-16 17:42:52 +08:00
if(strongServer){
_session->attachServer(*strongServer);
}
//此处截取数据并进行websocket协议打包
weak_ptr<WebSocketSessionBase> weakSelf = dynamic_pointer_cast<WebSocketSessionBase>(HttpSessionType::shared_from_this());
dynamic_pointer_cast<SendInterceptor>(_session)->setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf) {
2019-09-16 17:42:52 +08:00
auto strongSelf = weakSelf.lock();
if (strongSelf) {
2019-09-16 17:42:52 +08:00
WebSocketHeader header;
header._fin = true;
header._reserved = 0;
header._opcode = DataType;
header._mask_flag = false;
strongSelf->WebSocketSplitter::encode(header, buf);
2019-09-16 17:42:52 +08:00
}
return buf->size();
});
//允许websocket客户端
return true;
}
2018-12-20 10:31:31 +08:00
/**
* webSocket数据包
*/
void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{
//新包,原来的包残余数据清空掉
_payload_section.clear();
2018-12-20 10:31:31 +08:00
}
/**
* websocket数据包负载
*/
void onWebSocketDecodePayload(const WebSocketHeader &packet,const uint8_t *ptr,size_t len,size_t recved) override {
_payload_section.append((char *)ptr,len);
2018-12-20 10:31:31 +08:00
}
/**
* webSocket数据包后回调
* @param header
*/
void onWebSocketDecodeComplete(const WebSocketHeader &header_in) override {
WebSocketHeader& header = const_cast<WebSocketHeader&>(header_in);
auto flag = header._mask_flag;
header._mask_flag = false;
switch (header._opcode){
case WebSocketHeader::CLOSE:{
HttpSessionType::encode(header,nullptr);
HttpSessionType::shutdown(SockException(Err_shutdown, "recv close request from client"));
2018-12-20 10:31:31 +08:00
break;
}
2018-12-20 10:31:31 +08:00
case WebSocketHeader::PING:{
2019-09-16 17:42:52 +08:00
header._opcode = WebSocketHeader::PONG;
HttpSessionType::encode(header,std::make_shared<BufferString>(_payload_section));
2018-12-20 10:31:31 +08:00
break;
}
case WebSocketHeader::CONTINUATION:
2018-12-20 10:31:31 +08:00
case WebSocketHeader::TEXT:
case WebSocketHeader::BINARY:{
if (!header._fin) {
//还有后续分片数据, 我们先缓存数据,所有分片收集完成才一次性输出
_payload_cache.append(std::move(_payload_section));
if (_payload_cache.size() < MAX_WS_PACKET) {
//还有内存容量缓存分片数据
break;
}
//分片缓存太大,需要清空
}
//最后一个包
if (_payload_cache.empty()) {
//这个包是唯一个分片
_session->onRecv(std::make_shared<WebSocketBuffer>(header._opcode, header._fin, std::move(_payload_section)));
break;
}
//这个包由多个分片组成
_payload_cache.append(std::move(_payload_section));
_session->onRecv(std::make_shared<WebSocketBuffer>(header._opcode, header._fin, std::move(_payload_cache)));
_payload_cache.clear();
2018-12-20 10:31:31 +08:00
break;
}
default: break;
2018-12-20 10:31:31 +08:00
}
_payload_section.clear();
2018-12-20 10:31:31 +08:00
header._mask_flag = flag;
}
/**
* websocket协议打包后回调
2018-12-20 10:31:31 +08:00
*/
void onWebSocketEncodeData(Buffer::Ptr buffer) override{
HttpSessionType::send(std::move(buffer));
2018-12-20 10:31:31 +08:00
}
2018-12-20 10:31:31 +08:00
private:
string _payload_cache;
string _payload_section;
weak_ptr<TcpServer> _weak_server;
TcpSession::Ptr _session;
Creator _creator;
2018-12-20 10:31:31 +08:00
};
template<typename TcpSessionType,typename HttpSessionType = HttpSession,WebSocketHeader::Type DataType = WebSocketHeader::TEXT>
class WebSocketSession : public WebSocketSessionBase<TcpSessionCreator<TcpSessionType>,HttpSessionType,DataType>{
public:
WebSocketSession(const Socket::Ptr &pSock) : WebSocketSessionBase<TcpSessionCreator<TcpSessionType>,HttpSessionType,DataType>(pSock){}
virtual ~WebSocketSession(){}
};
2018-12-20 10:31:31 +08:00
#endif //ZLMEDIAKIT_WEBSOCKETSESSION_H