Mirror/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2026-06-16 13:05:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

实现websocket客户端模板

Browse Source
This commit is contained in:
xiongziliang
2019-09-16 17:45:27 +08:00
parent 3095075749
commit 644905a374
4 changed files with 377 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files

463
src/Http/WebSocketClient.h
View File

@@ -1,205 +1,326 @@
#ifndef Http_WebSocketClient_h
#define Http_WebSocketClient_h
/*
* MIT License
*
* Copyright (c) 2016-2019 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_WebSocketClient_H
#define ZLMEDIAKIT_WebSocketClient_H
#include <stdio.h>
#include <string.h>
#include <functional>
#include <memory>
#include "Util/util.h"
#include "Util/mini.h"
#include "Network/TcpClient.h"
#include "Common/Parser.h"
#include "HttpRequestSplitter.h"
#include "HttpCookie.h"
#include "HttpChunkedSplitter.h"
#include "strCoding.h"
#include "Http/HttpClient.h"
#include "Http/WebSocketSplitter.h"
#include "Http/WebSocketSession.h"
#include <cstdlib>
#include <random>
#include "Common/config.h"
#include "Util/SHA1.h"
#include "Util/base64.h"
using namespace std;
#include "Util/SHA1.h"
#include "Network/TcpClient.h"
#include "HttpClientImp.h"
#include "WebSocketSplitter.h"
using namespace toolkit;
namespace mediakit {
namespace mediakit{
/**
* @brief 客户端的状态
*/
typedef enum WSClientStatus {
WSCONNECT,
HANDSHAKING, ///握手中
WORKING, ///工作中
} WSClientStatus;
template <typename ClientType,WebSocketHeader::Type DataType>
class HttpWsClient;
class WebSocketClient : public TcpClient , public HttpRequestSplitter, public WebSocketSplitter
{
template <typename ClientType,WebSocketHeader::Type DataType>
class ClientTypeImp : public ClientType {
public:
typedef StrCaseMap HttpHeader;
typedef std::shared_ptr<WebSocketClient> Ptr;
WebSocketClient() :_WSClientStatus(WSCONNECT) {}
virtual ~WebSocketClient() {}
template <typename SessionType>
void startConnect(const string &strUrl, uint16_t iPort, float fTimeOutSec = 3)
{
_ip = strUrl;
_port = iPort;
TcpClient::startConnect(strUrl, iPort, fTimeOutSec);
typedef function<int(const Buffer::Ptr &buf)> onBeforeSendCB;
/**
* 该类实现了TcpSession派生类发送数据的截取
* 目的是发送业务数据前进行websocket协议的打包
*/
class SessionImp : public SessionType {
public:
SessionImp(const Socket::Ptr &pSock) :SessionType(pSock) {}
~SessionImp() {}
/**
* 设置发送数据截取回调函数
* @param cb 截取回调函数
*/
void setOnBeforeSendCB(const onBeforeSendCB &cb) {
_beforeSendCB = cb;
}
protected:
/**
* 重载send函数截取数据
* @param buf 需要截取的数据
* @return 数据字节数
*/
int send(const Buffer::Ptr &buf) override {
if (_beforeSendCB) {
return _beforeSendCB(buf);
}
return SessionType::send(buf);
}
private:
onBeforeSendCB _beforeSendCB;
};
std::shared_ptr<SessionImp> temSession = std::make_shared<SessionImp>(_sock);
//此处截取数据并进行websocket协议打包
weak_ptr<WebSocketClient> weakSelf = dynamic_pointer_cast<WebSocketClient>(WebSocketClient::shared_from_this());
_sock->setOnErr([weakSelf](const SockException &ex) {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return;
}
strongSelf->onErr(ex);
});
temSession->setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf) {
auto strongSelf = weakSelf.lock();
if (strongSelf) {
WebSocketHeader header;
header._fin = true;
header._reserved = 0;
header._opcode = WebSocketHeader::BINARY;
header._mask_flag = false;
strongSelf->WebSocketSplitter::encode(header, (uint8_t *)buf->data(), buf->size());
}
return buf->size();
});
_session = temSession;
_session->onManager();
}
virtual int send(const string& buf);
virtual void clear();
const string &responseStatus() const;
const HttpHeader &responseHeader() const;
const Parser& response() const;
const string &getUrl() const;
typedef function<int(const Buffer::Ptr &buf)> onBeforeSendCB;
friend class HttpWsClient<ClientType,DataType>;
template<typename ...ArgsType>
ClientTypeImp(ArgsType &&...args): ClientType(std::forward<ArgsType>(args)...){}
~ClientTypeImp() override {};
protected:
virtual int64_t onResponseHeader(const string &status,const HttpHeader &headers);;
/**
* 发送前拦截并打包为websocket协议
* @param buf
* @return
*/
int send(const Buffer::Ptr &buf) override{
if(_beforeSendCB){
return _beforeSendCB(buf);
}
return ClientType::send(buf);
}
/**
* 设置发送数据截取回调函数
* @param cb 截取回调函数
*/
void setOnBeforeSendCB(const onBeforeSendCB &cb){
_beforeSendCB = cb;
}
private:
onBeforeSendCB _beforeSendCB;
};
virtual void onResponseBody(const char *buf,int64_t size,int64_t recvedSize,int64_t totalSize);;
template <typename ClientType,WebSocketHeader::Type DataType = WebSocketHeader::TEXT>
class HttpWsClient : public HttpClientImp , public WebSocketSplitter{
public:
typedef shared_ptr<HttpWsClient> Ptr;
HttpWsClient(ClientTypeImp<ClientType,DataType> &delegate) : _delegate(delegate){
_Sec_WebSocket_Key = encodeBase64(SHA1::encode_bin(makeRandStr(16, false)));
}
~HttpWsClient(){}
void startWsClient(const string &ws_url,float fTimeOutSec){
string http_url = ws_url;
replace(http_url,"ws://","http://");
replace(http_url,"wss://","https://");
setMethod("GET");
addHeader("Upgrade","websocket");
addHeader("Connection","Upgrade");
addHeader("Sec-WebSocket-Version","13");
addHeader("Sec-WebSocket-Key",_Sec_WebSocket_Key);
_onRecv = nullptr;
sendRequest(http_url,fTimeOutSec);
}
protected:
//HttpClientImp override
/**
* 收到http回复头
* @param status 状态码，譬如:200 OK
* @param headers http头
* @return 返回后续content的长度；-1:后续数据全是content；>=0:固定长度content
* 需要指出的是，在http头中带有Content-Length字段时，该返回值无效
*/
int64_t onResponseHeader(const string &status,const HttpHeader &headers) override {
if(status == "101"){
auto Sec_WebSocket_Accept = encodeBase64(SHA1::encode_bin(_Sec_WebSocket_Key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
if(Sec_WebSocket_Accept == const_cast<HttpHeader &>(headers)["Sec-WebSocket-Accept"]){
//success
onWebSocketException(SockException());
return 0;
}
shutdown(SockException(Err_shutdown,StrPrinter << "Sec-WebSocket-Accept mismatch"));
return 0;
}
shutdown(SockException(Err_shutdown,StrPrinter << "bad http status code:" << status));
return 0;
};
/**
* 接收http回复完毕,
*/
virtual void onResponseCompleted();
void onResponseCompleted() override {}
//TcpClient override
void onManager() override {
if(_onRecv){
//websocket连接成功了
_delegate.onManager();
} else{
//websocket连接中...
HttpClientImp::onManager();
}
}
//数据全部发送完毕后回调
void onFlush() override{
if(_onRecv){
//websocket连接成功了
_delegate.onFlush();
} else{
//websocket连接中...
HttpClientImp::onFlush();
}
}
/**
* http链接断开回调
* @param ex 断开原因
* tcp连接结果
* @param ex
*/
virtual void onDisconnect(const SockException &ex){}
void onConnect(const SockException &ex) override{
if(ex){
//tcp连接失败，直接返回失败
onWebSocketException(ex);
return;
}
//开始websocket握手
HttpClientImp::onConnect(ex);
}
//HttpRequestSplitter override
int64_t onRecvHeader(const char *data, uint64_t len) override;
/**
* tcp收到数据
* @param pBuf
*/
void onRecv(const Buffer::Ptr &pBuf) override{
if(_onRecv){
//完成websocket握手后，拦截websocket数据
_onRecv(pBuf);
}else{
//websocket握手数据
HttpClientImp::onRecv(pBuf);
}
}
void onRecvContent(const char *data, uint64_t len) override;
//tcp连接断开
void onErr(const SockException &ex) override{
//tcp断开或者shutdown导致的断开
onWebSocketException(ex);
}
protected:
virtual void onConnect(const SockException &ex) override;
//WebSocketSplitter override
/**
* 收到一个webSocket数据包包头，后续将继续触发onWebSocketDecodePlayload回调
* @param header 数据包头
*/
void onWebSocketDecodeHeader(const WebSocketHeader &header) override{
_payload.clear();
}
virtual void onRecv(const Buffer::Ptr &pBuf) override;
/**
* 收到webSocket数据包负载
* @param header 数据包包头
* @param ptr 负载数据指针
* @param len 负载数据长度
* @param recved 已接收数据长度(包含本次数据长度)，等于header._playload_len时则接受完毕
*/
void onWebSocketDecodePlayload(const WebSocketHeader &header, const uint8_t *ptr, uint64_t len, uint64_t recved) override{
_payload.append((char *)ptr,len);
}
virtual void onErr(const SockException &ex) override;
virtual void onFlush() override {};
/**
* 接收到完整的一个webSocket数据包后回调
* @param header 数据包包头
*/
void onWebSocketDecodeComplete(const WebSocketHeader &header_in) override{
WebSocketHeader& header = const_cast<WebSocketHeader&>(header_in);
auto flag = header._mask_flag;
//websocket客户端发送数据需要加密
header._mask_flag = true;
virtual void onManager() override;
switch (header._opcode){
case WebSocketHeader::CLOSE:{
//服务器主动关闭
WebSocketSplitter::encode(header,nullptr);
shutdown(SockException(Err_eof,"websocket server close the connection"));
}
break;
case WebSocketHeader::PING:{
//心跳包
header._opcode = WebSocketHeader::PONG;
WebSocketSplitter::encode(header,std::make_shared<BufferString>(std::move(_payload)));
}
break;
case WebSocketHeader::CONTINUATION:{
protected:
string generate_websocket_client_handshake(const char* ip, uint16_t port, const char * path, const char * key);
}
break;
case WebSocketHeader::TEXT:
case WebSocketHeader::BINARY:{
//接收完毕websocket数据包，触发onRecv事件
_delegate.onRecv(std::make_shared<BufferString>(std::move(_payload)));
}
break;
default:
break;
}
_payload.clear();
header._mask_flag = flag;
}
string get_random(size_t n);
/**
* websocket数据编码回调
* @param ptr 数据指针
* @param len 数据指针长度
*/
void onWebSocketEncodeData(const Buffer::Ptr &buffer) override{
HttpClientImp::send(buffer);
}
private:
void onWebSocketException(const SockException &ex){
if(!ex){
//websocket握手成功
//此处截取TcpClient派生类发送的数据并进行websocket协议打包
weak_ptr<HttpWsClient> weakSelf = dynamic_pointer_cast<HttpWsClient>(shared_from_this());
_delegate.setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf){
auto strongSelf = weakSelf.lock();
if(strongSelf){
WebSocketHeader header;
header._fin = true;
header._reserved = 0;
header._opcode = DataType;
//客户端需要加密
header._mask_flag = true;
strongSelf->WebSocketSplitter::encode(header,buf);
}
return buf->size();
});
//触发连接成功事件
_delegate._sock = HttpClientImp::_sock;
_delegate.onConnect(ex);
//拦截websocket数据接收
_onRecv = [this](const Buffer::Ptr &pBuf){
WebSocketSplitter::decode((uint8_t*)pBuf->data(),pBuf->size());
};
return;
}
void onWebSocketDecodeHeader(const WebSocketHeader &packet) override;
//websocket握手失败或者tcp连接失败或者中途断开
if(_onRecv){
//握手成功之后的中途断开
_onRecv = nullptr;
_delegate.onErr(ex);
return;
}
void onWebSocketDecodePlayload(const WebSocketHeader &packet, const uint8_t *ptr, uint64_t len, uint64_t recved) override;
void onWebSocketDecodeComplete(const WebSocketHeader &header_in) override;
virtual void onWebSocketEncodeData(const uint8_t *ptr, uint64_t len);
//websocket握手失败或者tcp连接失败
_delegate.onConnect(ex);
}
private:
void onResponseCompleted_l();
protected:
bool _isHttps;
private:
string _ip;
int _port;
string _url;
string _method;
string _path;
//recv
int64_t _recvedBodySize;
int64_t _totalBodySize;
Parser _parser;
string _lastHost;
Ticker _aliveTicker;
float _fTimeOutSec = 0;
std::shared_ptr<HttpChunkedSplitter> _chunkedSplitter;
std::string _key; ///客户端的key
WSClientStatus _WSClientStatus; ///客户端状态
bool _firstPacket = true;
string _remian_data;
std::shared_ptr<TcpSession> _session;
string _Sec_WebSocket_Key;
function<void(const Buffer::Ptr &pBuf)> _onRecv;
ClientTypeImp<ClientType,DataType> &_delegate;
string _payload;
};
} /* namespace mediakit */
template <typename ClientType,WebSocketHeader::Type DataType = WebSocketHeader::TEXT,bool userWSS = false >
class WebSocketClient : public ClientTypeImp<ClientType,DataType>{
public:
typedef std::shared_ptr<WebSocketClient> Ptr;
#endif /* Http_HttpClient_h */
template<typename ...ArgsType>
WebSocketClient(ArgsType &&...args) : ClientTypeImp<ClientType,DataType>(std::forward<ArgsType>(args)...){
_wsClient.reset(new HttpWsClient<ClientType,DataType>(*this));
}
~WebSocketClient() override {}
void startConnect(const string &strUrl, uint16_t iPort, float fTimeOutSec = 3) override {
string ws_url;
if(userWSS){
ws_url = StrPrinter << "wss://" + strUrl << ":" << iPort << "/" ;
}else{
ws_url = StrPrinter << "ws://" + strUrl << ":" << iPort << "/" ;
}
_wsClient->startWsClient(ws_url,fTimeOutSec);
}
private:
typename HttpWsClient<ClientType,DataType>::Ptr _wsClient;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_WebSocketClient_H