Merge pull request #7 from zlmediakit/master

update
This commit is contained in:
baiyfcu 2019-09-17 12:09:24 +08:00 committed by GitHub
commit f31773ee19
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 642 additions and 580 deletions

@ -1 +1 @@
Subproject commit 0b406073125080ab8edd13ee7c14e573e54baa35 Subproject commit 91246bb01475c7336040a4b7ec35d0584887f365

View File

@ -42,7 +42,7 @@
- HTTP server,suppor directory meun、RESTful http api. - HTTP server,suppor directory meun、RESTful http api.
- HTTP client,downloader,uploader,and http api requester. - HTTP client,downloader,uploader,and http api requester.
- Cookie supported. - Cookie supported.
- WebSocket Server. - WebSocket Server and Client.
- File access authentication. - File access authentication.
- Others - Others
@ -95,7 +95,7 @@
| RTSP[S] Play Server | Y | | RTSP[S] Play Server | Y |
| RTSP[S] Push Server | Y | | RTSP[S] Push Server | Y |
| RTMP | Y | | RTMP | Y |
| HTTP[S]/WebSocket | Y | | HTTP[S]/WebSocket[S] | Y |
- Client supported: - Client supported:
@ -106,6 +106,7 @@
| RTMP Player | Y | | RTMP Player | Y |
| RTMP Pusher | Y | | RTMP Pusher | Y |
| HTTP[S] | Y | | HTTP[S] | Y |
| WebSocket[S] | Y |

View File

@ -51,7 +51,7 @@
- 完整HTTP API服务器可以作为web后台开发框架。 - 完整HTTP API服务器可以作为web后台开发框架。
- 支持跨域访问。 - 支持跨域访问。
- 支持http客户端、服务器cookie - 支持http客户端、服务器cookie
- 支持WebSocket服务器 - 支持WebSocket服务器和客户端
- 支持http文件访问鉴权 - 支持http文件访问鉴权
- 其他 - 其他
@ -110,7 +110,7 @@
| RTSP[S] Play Server | Y | | RTSP[S] Play Server | Y |
| RTSP[S] Push Server | Y | | RTSP[S] Push Server | Y |
| RTMP | Y | | RTMP | Y |
| HTTP[S]/WebSocket | Y | | HTTP[S]/WebSocket[S] | Y |
- 支持的客户端类型 - 支持的客户端类型
@ -121,6 +121,7 @@
| RTMP Player | Y | | RTMP Player | Y |
| RTMP Pusher | Y | | RTMP Pusher | Y |
| HTTP[S] | Y | | HTTP[S] | Y |
| WebSocket[S] | Y |
## 后续任务 ## 后续任务
- 完善支持H265 - 完善支持H265

View File

@ -270,13 +270,13 @@ int main(int argc,char *argv[]) {
shellSrv->start<ShellSession>(shellPort); shellSrv->start<ShellSession>(shellPort);
rtspSrv->start<RtspSession>(rtspPort);//默认554 rtspSrv->start<RtspSession>(rtspPort);//默认554
rtmpSrv->start<RtmpSession>(rtmpPort);//默认1935 rtmpSrv->start<RtmpSession>(rtmpPort);//默认1935
//http服务器,支持websocket //http服务器
httpSrv->start<EchoWebSocketSession>(httpPort);//默认80 httpSrv->start<HttpSession>(httpPort);//默认80
//如果支持ssl还可以开启https服务器 //如果支持ssl还可以开启https服务器
TcpServer::Ptr httpsSrv(new TcpServer()); TcpServer::Ptr httpsSrv(new TcpServer());
//https服务器,支持websocket //https服务器,支持websocket
httpsSrv->start<SSLEchoWebSocketSession>(httpsPort);//默认443 httpsSrv->start<HttpsSession>(httpsPort);//默认443
//支持ssl加密的rtsp服务器可用于诸如亚马逊echo show这样的设备访问 //支持ssl加密的rtsp服务器可用于诸如亚马逊echo show这样的设备访问
TcpServer::Ptr rtspSSLSrv(new TcpServer()); TcpServer::Ptr rtspSSLSrv(new TcpServer());

View File

@ -195,15 +195,14 @@ void HttpSession::onManager() {
} }
} }
bool HttpSession::checkWebSocket(){
inline bool HttpSession::checkWebSocket(){
auto Sec_WebSocket_Key = _parser["Sec-WebSocket-Key"]; auto Sec_WebSocket_Key = _parser["Sec-WebSocket-Key"];
if(Sec_WebSocket_Key.empty()){ if(Sec_WebSocket_Key.empty()){
return false; return false;
} }
auto Sec_WebSocket_Accept = encodeBase64(SHA1::encode_bin(Sec_WebSocket_Key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11")); auto Sec_WebSocket_Accept = encodeBase64(SHA1::encode_bin(Sec_WebSocket_Key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"));
KeyValue headerOut; KeyValue headerOut = makeHttpHeader();
headerOut["Upgrade"] = "websocket"; headerOut["Upgrade"] = "websocket";
headerOut["Connection"] = "Upgrade"; headerOut["Connection"] = "Upgrade";
headerOut["Sec-WebSocket-Accept"] = Sec_WebSocket_Accept; headerOut["Sec-WebSocket-Accept"] = Sec_WebSocket_Accept;
@ -223,12 +222,17 @@ inline bool HttpSession::checkWebSocket(){
} }
//如果checkLiveFlvStream返回false,则代表不是websocket-flv而是普通的websocket连接 //如果checkLiveFlvStream返回false,则代表不是websocket-flv而是普通的websocket连接
if(!onWebSocketConnect(_parser)){
sendResponse("501 Not Implemented",headerOut,"");
shutdown(SockException(Err_shutdown,"WebSocket server not implemented"));
return true;
}
sendResponse("101 Switching Protocols",headerOut,""); sendResponse("101 Switching Protocols",headerOut,"");
return true; return true;
} }
//http-flv 链接格式:http://vhost-url:port/app/streamid.flv?key1=value1&key2=value2 //http-flv 链接格式:http://vhost-url:port/app/streamid.flv?key1=value1&key2=value2
//如果url(除去?以及后面的参数)后缀是.flv,那么表明该url是一个http-flv直播。 //如果url(除去?以及后面的参数)后缀是.flv,那么表明该url是一个http-flv直播。
inline bool HttpSession::checkLiveFlvStream(const function<void()> &cb){ bool HttpSession::checkLiveFlvStream(const function<void()> &cb){
auto pos = strrchr(_parser.Url().data(),'.'); auto pos = strrchr(_parser.Url().data(),'.');
if(!pos){ if(!pos){
//未找到".flv"后缀 //未找到".flv"后缀
@ -316,9 +320,9 @@ inline bool HttpSession::checkLiveFlvStream(const function<void()> &cb){
return true; return true;
} }
inline bool makeMeun(const string &httpPath,const string &strFullPath, string &strRet) ; bool makeMeun(const string &httpPath,const string &strFullPath, string &strRet) ;
inline static string findIndexFile(const string &dir){ static string findIndexFile(const string &dir){
DIR *pDir; DIR *pDir;
dirent *pDirent; dirent *pDirent;
if ((pDir = opendir(dir.data())) == NULL) { if ((pDir = opendir(dir.data())) == NULL) {
@ -336,7 +340,7 @@ inline static string findIndexFile(const string &dir){
return ""; return "";
} }
inline string HttpSession::getClientUid(){ string HttpSession::getClientUid(){
//如果http客户端不支持cookie那么我们可以通过url参数来追踪用户 //如果http客户端不支持cookie那么我们可以通过url参数来追踪用户
//如果url参数也没有那么只能通过ip+端口号来追踪用户 //如果url参数也没有那么只能通过ip+端口号来追踪用户
//追踪用户的目的是为了减少http短链接情况的重复鉴权验证通过缓存记录鉴权结果提高性能 //追踪用户的目的是为了减少http短链接情况的重复鉴权验证通过缓存记录鉴权结果提高性能
@ -349,13 +353,13 @@ inline string HttpSession::getClientUid(){
//字符串是否以xx结尾 //字符串是否以xx结尾
static inline bool end_of(const string &str, const string &substr){ static bool end_of(const string &str, const string &substr){
auto pos = str.rfind(substr); auto pos = str.rfind(substr);
return pos != string::npos && pos == str.size() - substr.size(); return pos != string::npos && pos == str.size() - substr.size();
}; };
//拦截hls的播放请求 //拦截hls的播放请求
static inline bool checkHls(BroadcastHttpAccessArgs){ static bool checkHls(BroadcastHttpAccessArgs){
if(!end_of(args._streamid,("/hls.m3u8"))) { if(!end_of(args._streamid,("/hls.m3u8"))) {
//不是hls //不是hls
return false; return false;
@ -371,7 +375,7 @@ static inline bool checkHls(BroadcastHttpAccessArgs){
return NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,args_copy,mediaAuthInvoker,sender); return NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,args_copy,mediaAuthInvoker,sender);
} }
inline void HttpSession::canAccessPath(const string &path_in,bool is_dir,const function<void(const string &errMsg,const HttpServerCookie::Ptr &cookie)> &callback_in){ void HttpSession::canAccessPath(const string &path_in,bool is_dir,const function<void(const string &errMsg,const HttpServerCookie::Ptr &cookie)> &callback_in){
auto path = path_in; auto path = path_in;
replace(const_cast<string &>(path),"//","/"); replace(const_cast<string &>(path),"//","/");
@ -472,13 +476,12 @@ inline void HttpSession::canAccessPath(const string &path_in,bool is_dir,const f
} }
inline void HttpSession::Handle_Req_GET(int64_t &content_len) { void HttpSession::Handle_Req_GET(int64_t &content_len) {
//先看看是否为WebSocket请求 //先看看是否为WebSocket请求
if(checkWebSocket()){ if(checkWebSocket()){
content_len = -1; content_len = -1;
auto parserCopy = _parser; _contentCallBack = [this](const char *data,uint64_t len){
_contentCallBack = [this,parserCopy](const char *data,uint64_t len){ WebSocketSplitter::decode((uint8_t *)data,len);
onRecvWebSocketData(parserCopy,data,len);
//_contentCallBack是可持续的后面还要处理后续数据 //_contentCallBack是可持续的后面还要处理后续数据
return true; return true;
}; };
@ -666,7 +669,7 @@ inline void HttpSession::Handle_Req_GET(int64_t &content_len) {
}); });
} }
inline bool makeMeun(const string &httpPath,const string &strFullPath, string &strRet) { bool makeMeun(const string &httpPath,const string &strFullPath, string &strRet) {
string strPathPrefix(strFullPath); string strPathPrefix(strFullPath);
string last_dir_name; string last_dir_name;
if(strPathPrefix.back() == '/'){ if(strPathPrefix.back() == '/'){
@ -764,7 +767,8 @@ inline bool makeMeun(const string &httpPath,const string &strFullPath, string &s
ss.str().swap(strRet); ss.str().swap(strRet);
return true; return true;
} }
inline void HttpSession::sendResponse(const char* pcStatus, const KeyValue& header, const string& strContent) {
void HttpSession::sendResponse(const char* pcStatus, const KeyValue& header, const string& strContent) {
_StrPrinter printer; _StrPrinter printer;
printer << "HTTP/1.1 " << pcStatus << "\r\n"; printer << "HTTP/1.1 " << pcStatus << "\r\n";
for (auto &pr : header) { for (auto &pr : header) {
@ -775,7 +779,8 @@ inline void HttpSession::sendResponse(const char* pcStatus, const KeyValue& head
send(strSend); send(strSend);
_ticker.resetTime(); _ticker.resetTime();
} }
inline HttpSession::KeyValue HttpSession::makeHttpHeader(bool bClose, int64_t iContentSize,const char* pcContentType) {
HttpSession::KeyValue HttpSession::makeHttpHeader(bool bClose, int64_t iContentSize,const char* pcContentType) {
KeyValue headerOut; KeyValue headerOut;
GET_CONFIG(string,charSet,Http::kCharSet); GET_CONFIG(string,charSet,Http::kCharSet);
GET_CONFIG(uint32_t,keepAliveSec,Http::kKeepAliveSecond); GET_CONFIG(uint32_t,keepAliveSec,Http::kKeepAliveSecond);
@ -814,14 +819,14 @@ string HttpSession::urlDecode(const string &str){
return ret; return ret;
} }
inline void HttpSession::urlDecode(Parser &parser){ void HttpSession::urlDecode(Parser &parser){
parser.setUrl(urlDecode(parser.Url())); parser.setUrl(urlDecode(parser.Url()));
for(auto &pr : _parser.getUrlArgs()){ for(auto &pr : _parser.getUrlArgs()){
const_cast<string &>(pr.second) = urlDecode(pr.second); const_cast<string &>(pr.second) = urlDecode(pr.second);
} }
} }
inline bool HttpSession::emitHttpEvent(bool doInvoke){ bool HttpSession::emitHttpEvent(bool doInvoke){
///////////////////是否断开本链接/////////////////////// ///////////////////是否断开本链接///////////////////////
GET_CONFIG(uint32_t,reqCnt,Http::kMaxReqCount); GET_CONFIG(uint32_t,reqCnt,Http::kMaxReqCount);
@ -857,7 +862,8 @@ inline bool HttpSession::emitHttpEvent(bool doInvoke){
} }
return consumed; return consumed;
} }
inline void HttpSession::Handle_Req_POST(int64_t &content_len) {
void HttpSession::Handle_Req_POST(int64_t &content_len) {
GET_CONFIG(uint64_t,maxReqSize,Http::kMaxReqSize); GET_CONFIG(uint64_t,maxReqSize,Http::kMaxReqSize);
GET_CONFIG(int,maxReqCnt,Http::kMaxReqCount); GET_CONFIG(int,maxReqCnt,Http::kMaxReqCount);
@ -944,7 +950,8 @@ void HttpSession::responseDelay(bool bClose,
} }
sendResponse(codeOut.data(), headerOut, contentOut); sendResponse(codeOut.data(), headerOut, contentOut);
} }
inline void HttpSession::sendNotFound(bool bClose) {
void HttpSession::sendNotFound(bool bClose) {
GET_CONFIG(string,notFound,Http::kNotFound); GET_CONFIG(string,notFound,Http::kNotFound);
sendResponse("404 Not Found", makeHttpHeader(bClose, notFound.size()), notFound); sendResponse("404 Not Found", makeHttpHeader(bClose, notFound.size()), notFound);
} }

View File

@ -72,8 +72,8 @@ protected:
void onWrite(const Buffer::Ptr &data) override ; void onWrite(const Buffer::Ptr &data) override ;
void onDetach() override; void onDetach() override;
std::shared_ptr<FlvMuxer> getSharedPtr() override; std::shared_ptr<FlvMuxer> getSharedPtr() override;
//HttpRequestSplitter override
//HttpRequestSplitter override
int64_t onRecvHeader(const char *data,uint64_t len) override; int64_t onRecvHeader(const char *data,uint64_t len) override;
void onRecvContent(const char *data,uint64_t len) override; void onRecvContent(const char *data,uint64_t len) override;
@ -94,29 +94,32 @@ protected:
shutdown(SockException(Err_shutdown,"http post content is too huge,default closed")); shutdown(SockException(Err_shutdown,"http post content is too huge,default closed"));
} }
void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{ /**
shutdown(SockException(Err_shutdown,"websocket connection default closed")); * websocket客户端连接上事件
}; * @param header http头
* @return true代表允许websocket连接
void onRecvWebSocketData(const Parser &header,const char *data,uint64_t len){ */
WebSocketSplitter::decode((uint8_t *)data,len); virtual bool onWebSocketConnect(const Parser &header){
WarnL << "http server do not support websocket default";
return false;
} }
//WebSocketSplitter override
/** /**
* websocket协议打包后回调 * websocket协议打包后回调
* @param buffer * @param buffer websocket协议数据
*/ */
void onWebSocketEncodeData(const Buffer::Ptr &buffer) override; void onWebSocketEncodeData(const Buffer::Ptr &buffer) override;
private: private:
inline void Handle_Req_GET(int64_t &content_len); void Handle_Req_GET(int64_t &content_len);
inline void Handle_Req_POST(int64_t &content_len); void Handle_Req_POST(int64_t &content_len);
inline bool checkLiveFlvStream(const function<void()> &cb = nullptr); bool checkLiveFlvStream(const function<void()> &cb = nullptr);
inline bool checkWebSocket(); bool checkWebSocket();
inline bool emitHttpEvent(bool doInvoke); bool emitHttpEvent(bool doInvoke);
inline void urlDecode(Parser &parser); void urlDecode(Parser &parser);
inline void sendNotFound(bool bClose); void sendNotFound(bool bClose);
inline void sendResponse(const char *pcStatus,const KeyValue &header,const string &strContent); void sendResponse(const char *pcStatus,const KeyValue &header,const string &strContent);
inline KeyValue makeHttpHeader(bool bClose=false,int64_t iContentSize=-1,const char *pcContentType="text/html"); KeyValue makeHttpHeader(bool bClose=false,int64_t iContentSize=-1,const char *pcContentType="text/html");
void responseDelay(bool bClose, void responseDelay(bool bClose,
const string &codeOut, const string &codeOut,
const KeyValue &headerOut, const KeyValue &headerOut,
@ -134,14 +137,14 @@ private:
* @param is_dir path是否为目录 * @param is_dir path是否为目录
* @param callback * @param callback
*/ */
inline void canAccessPath(const string &path,bool is_dir,const function<void(const string &errMsg,const HttpServerCookie::Ptr &cookie)> &callback); void canAccessPath(const string &path,bool is_dir,const function<void(const string &errMsg,const HttpServerCookie::Ptr &cookie)> &callback);
/** /**
* id * id
* url参数返回参数ip+ * url参数返回参数ip+
* @return * @return
*/ */
inline string getClientUid(); string getClientUid();
//设置socket标志 //设置socket标志
void setSocketFlags(); void setSocketFlags();

View File

@ -1,288 +0,0 @@
#include "WebSocketClient.h"
int mediakit::WebSocketClient::send(const string& buf)
{
if (_sock)
{
if (_WSClientStatus == WORKING)
{
_session->send(buf);
return 0;
}
else
{
return -1;
}
}
}
void mediakit::WebSocketClient::clear()
{
_method.clear();
_path.clear();
_parser.Clear();
_recvedBodySize = 0;
_totalBodySize = 0;
_aliveTicker.resetTime();
_chunkedSplitter.reset();
HttpRequestSplitter::reset();
}
const std::string & mediakit::WebSocketClient::responseStatus() const
{
return _parser.Url();
}
const mediakit::WebSocketClient::HttpHeader & mediakit::WebSocketClient::responseHeader() const
{
return _parser.getValues();
}
const mediakit::Parser& mediakit::WebSocketClient::response() const
{
return _parser;
}
const std::string & mediakit::WebSocketClient::getUrl() const
{
return _url;
}
int64_t mediakit::WebSocketClient::onResponseHeader(const string &status, const HttpHeader &headers)
{
DebugL << status;
//无Content-Length字段时默认后面全是content
return -1;
}
void mediakit::WebSocketClient::onResponseBody(const char *buf, int64_t size, int64_t recvedSize, int64_t totalSize)
{
DebugL << size << " " << recvedSize << " " << totalSize;
}
void mediakit::WebSocketClient::onResponseCompleted()
{
DebugL;
}
int64_t mediakit::WebSocketClient::onRecvHeader(const char *data, uint64_t len)
{
_parser.Parse(data);
if (_parser.Url() == "101")
{
switch (_WSClientStatus)
{
case HANDSHAKING:
{
StrCaseMap& valueMap = _parser.getValues();
auto key = valueMap.find("Sec-WebSocket-Accept");
if (key != valueMap.end() && key->second.length() > 0) {
onConnect(SockException());
}
break;
}
}
return -1;
}
else
{
shutdown(SockException(Err_shutdown, _parser.Url().c_str()));
return 0;
}
return -1;
}
void mediakit::WebSocketClient::onRecvContent(const char *data, uint64_t len)
{
if (_chunkedSplitter) {
_chunkedSplitter->input(data, len);
return;
}
auto recvedBodySize = _recvedBodySize + len;
if (_totalBodySize < 0) {
//不限长度的content,最大支持INT64_MAX个字节
onResponseBody(data, len, recvedBodySize, INT64_MAX);
_recvedBodySize = recvedBodySize;
return;
}
//固定长度的content
if (recvedBodySize < _totalBodySize) {
//content还未接收完毕
onResponseBody(data, len, recvedBodySize, _totalBodySize);
_recvedBodySize = recvedBodySize;
return;
}
//content接收完毕
onResponseBody(data, _totalBodySize - _recvedBodySize, _totalBodySize, _totalBodySize);
bool biggerThanExpected = recvedBodySize > _totalBodySize;
onResponseCompleted_l();
if (biggerThanExpected) {
//声明的content数据比真实的小那么我们只截取前面部分的并断开链接
shutdown(SockException(Err_shutdown, "http response content size bigger than expected"));
}
}
void mediakit::WebSocketClient::onConnect(const SockException &ex)
{
_aliveTicker.resetTime();
if (ex) {
onDisconnect(ex);
return;
}
//先假设http客户端只会接收一点点数据只接受http头节省内存
_sock->setReadBuffer(std::make_shared<BufferRaw>(1 * 1024));
_totalBodySize = 0;
_recvedBodySize = 0;
HttpRequestSplitter::reset();
_chunkedSplitter.reset();
if (_WSClientStatus == WSCONNECT)
{
//Websocket握手
string random = get_random(16);
auto Sec_WebSocket_Key = encodeBase64(SHA1::encode_bin(random));
_key = Sec_WebSocket_Key;
string p = generate_websocket_client_handshake(_ip.c_str(), _port, _url.c_str(), _key.c_str());
TcpClient::send(p);
_WSClientStatus = HANDSHAKING;
}
else if (_WSClientStatus == HANDSHAKING)
{
_WSClientStatus = WORKING;
}
onFlush();
}
void mediakit::WebSocketClient::onRecv(const Buffer::Ptr &pBuf)
{
_aliveTicker.resetTime();
if (_WSClientStatus == HANDSHAKING || _WSClientStatus == WSCONNECT)
HttpRequestSplitter::input(pBuf->data(), pBuf->size());
else if (_WSClientStatus == WORKING)
{
WebSocketSplitter::decode((uint8_t *)pBuf->data(), pBuf->size());
}
}
void mediakit::WebSocketClient::onErr(const SockException &ex)
{
_session->onError(ex);
onDisconnect(ex);
}
void mediakit::WebSocketClient::onManager()
{
if (_WSClientStatus != WORKING)
{
if (_fTimeOutSec > 0 && _aliveTicker.elapsedTime() > _fTimeOutSec * 1000) {
//超时
shutdown(SockException(Err_timeout, "ws server respone timeout"));
}
}
else
_session->onManager();
}
std::string mediakit::WebSocketClient::generate_websocket_client_handshake(const char* ip, uint16_t port, const char * path, const char * key)
{
/**
* @brief 65535 - 14 - 1
*/
#define DATA_FRAME_MAX_LEN 65520
#define HANDSHAKE_SIZE 1024
char buf[HANDSHAKE_SIZE] = { 0 };
snprintf(buf, HANDSHAKE_SIZE,
"GET %s HTTP/1.1\r\n"
"Host: %s:%d\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Key: %s\r\n"
"Sec-WebSocket-Version: 13\r\n"
"\r\n",
path, ip, port, key);
string temBuf(buf);
return temBuf;
}
std::string mediakit::WebSocketClient::get_random(size_t n)
{
random_device rd;
_StrPrinter printer;
for (int i = 0; i < n; i++)
{
unsigned int rnd = rd();
printer << rnd % 9;
}
return string(printer);
}
void mediakit::WebSocketClient::onWebSocketDecodeHeader(const WebSocketHeader &packet)
{
//新包,原来的包残余数据清空掉
_remian_data.clear();
if (_firstPacket) {
//这是个WebSocket会话而不是普通的Http会话
_firstPacket = false;
//此处截取数据并进行websocket协议打包
}
}
void mediakit::WebSocketClient::onWebSocketDecodePlayload(const WebSocketHeader &packet, const uint8_t *ptr, uint64_t len, uint64_t recved)
{
_remian_data.append((char *)ptr, len);
}
void mediakit::WebSocketClient::onWebSocketDecodeComplete(const WebSocketHeader &header_in)
{
WebSocketHeader& header = const_cast<WebSocketHeader&>(header_in);
auto flag = header._mask_flag;
header._mask_flag = false;
switch (header._opcode) {
case WebSocketHeader::CLOSE: {
shutdown(SockException(Err_timeout, "session timeouted"));
}
break;
case WebSocketHeader::PING: {
const_cast<WebSocketHeader&>(header)._opcode = WebSocketHeader::PONG;
WebSocketSplitter::encode(header, (uint8_t *)_remian_data.data(), _remian_data.size());
}
break;
case WebSocketHeader::CONTINUATION: {
}
break;
case WebSocketHeader::TEXT:
case WebSocketHeader::BINARY: {
BufferString::Ptr buffer = std::make_shared<BufferString>(_remian_data);
_session->onRecv(buffer);
}
break;
default:
break;
}
_remian_data.clear();
header._mask_flag = flag;
}
void mediakit::WebSocketClient::onWebSocketEncodeData(const uint8_t *ptr, uint64_t len)
{
TcpClient::send(string((char*)ptr, len));
}
void mediakit::WebSocketClient::onResponseCompleted_l()
{
_totalBodySize = 0;
_recvedBodySize = 0;
onResponseCompleted();
}

View File

@ -1,205 +1,371 @@
#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/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" #include "Util/base64.h"
#include "Util/SHA1.h"
#include "Network/TcpClient.h"
#include "HttpClientImp.h"
using namespace std; #include "WebSocketSplitter.h"
using namespace toolkit; using namespace toolkit;
namespace mediakit { namespace mediakit{
template <typename ClientType,WebSocketHeader::Type DataType>
class HttpWsClient;
/** /**
* @brief * ,TcpClient数据发送前的拦截
*/ * @tparam ClientType TcpClient派生类
typedef enum WSClientStatus { * @tparam DataType ,
WSCONNECT,
HANDSHAKING, ///握手中
WORKING, ///工作中
} WSClientStatus;
class WebSocketClient : public TcpClient , public HttpRequestSplitter, public WebSocketSplitter
{
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 { template <typename ClientType,WebSocketHeader::Type DataType>
public: class ClientTypeImp : public ClientType {
SessionImp(const Socket::Ptr &pSock) :SessionType(pSock) {} public:
typedef function<int(const Buffer::Ptr &buf)> onBeforeSendCB;
~SessionImp() {} friend class HttpWsClient<ClientType,DataType>;
template<typename ...ArgsType>
ClientTypeImp(ArgsType &&...args): ClientType(std::forward<ArgsType>(args)...){}
~ClientTypeImp() override {};
protected:
/**
* websocket协议
* @param buf
* @return
*/
int send(const Buffer::Ptr &buf) override{
if(_beforeSendCB){
return _beforeSendCB(buf);
}
return ClientType::send(buf);
}
/** /**
* *
* @param cb * @param cb
*/ */
void setOnBeforeSendCB(const onBeforeSendCB &cb) { void setOnBeforeSendCB(const onBeforeSendCB &cb){
_beforeSendCB = cb; _beforeSendCB = cb;
} }
protected: private:
/**
* send函数截取数据
* @param buf
* @return
*/
int send(const Buffer::Ptr &buf) override {
if (_beforeSendCB) {
return _beforeSendCB(buf);
}
return SessionType::send(buf);
}
private:
onBeforeSendCB _beforeSendCB; 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(); * weksocket TcpClient派生类事件的桥接
if (!strongSelf) { * @tparam ClientType TcpClient派生类
return; * @tparam DataType websocket负载类型TEXT还是BINARY类型
*/
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)));
} }
strongSelf->onErr(ex); ~HttpWsClient(){}
});
temSession->setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf) { /**
auto strongSelf = weakSelf.lock(); * ws握手
if (strongSelf) { * @param ws_url ws连接url
WebSocketHeader header; * @param fTimeOutSec
header._fin = true; */
header._reserved = 0; void startWsClient(const string &ws_url,float fTimeOutSec){
header._opcode = WebSocketHeader::BINARY; string http_url = ws_url;
header._mask_flag = false; replace(http_url,"ws://","http://");
strongSelf->WebSocketSplitter::encode(header, (uint8_t *)buf->data(), buf->size()); 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);
} }
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;
protected: protected:
virtual int64_t onResponseHeader(const string &status,const HttpHeader &headers);; //HttpClientImp override
virtual void onResponseBody(const char *buf,int64_t size,int64_t recvedSize,int64_t totalSize);; /**
* 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回复完毕, * http回复完毕,
*/ */
virtual void onResponseCompleted(); void onResponseCompleted() override {}
//TcpClient override
/** /**
* http链接断开回调 *
* @param ex
*/ */
virtual void onDisconnect(const SockException &ex){} void onManager() override {
if(_onRecv){
//websocket连接成功了
_delegate.onManager();
} else{
//websocket连接中...
HttpClientImp::onManager();
}
}
//HttpRequestSplitter override /**
int64_t onRecvHeader(const char *data, uint64_t len) override; *
*/
void onFlush() override{
if(_onRecv){
//websocket连接成功了
_delegate.onFlush();
} else{
//websocket连接中...
HttpClientImp::onFlush();
}
}
void onRecvContent(const char *data, uint64_t len) override; /**
* tcp连接结果
* @param ex
*/
void onConnect(const SockException &ex) override{
if(ex){
//tcp连接失败直接返回失败
onWebSocketException(ex);
return;
}
//开始websocket握手
HttpClientImp::onConnect(ex);
}
protected: /**
virtual void onConnect(const SockException &ex) override; * tcp收到数据
* @param pBuf
*/
void onRecv(const Buffer::Ptr &pBuf) override{
if(_onRecv){
//完成websocket握手后拦截websocket数据并解析
_onRecv(pBuf);
}else{
//websocket握手数据
HttpClientImp::onRecv(pBuf);
}
}
virtual void onRecv(const Buffer::Ptr &pBuf) override; /**
* tcp连接断开
* @param ex
*/
void onErr(const SockException &ex) override{
//tcp断开或者shutdown导致的断开
onWebSocketException(ex);
}
virtual void onErr(const SockException &ex) override; //WebSocketSplitter override
virtual void onFlush() override {}; /**
* webSocket数据包包头onWebSocketDecodePlayload回调
* @param header
*/
void onWebSocketDecodeHeader(const WebSocketHeader &header) override{
_payload.clear();
}
virtual void onManager() 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);
}
protected:
string generate_websocket_client_handshake(const char* ip, uint16_t port, const char * path, const char * key);
string get_random(size_t n); /**
* 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;
void onWebSocketDecodeHeader(const WebSocketHeader &packet) 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:{
void onWebSocketDecodePlayload(const WebSocketHeader &packet, const uint8_t *ptr, uint64_t len, uint64_t recved) override; }
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;
}
void onWebSocketDecodeComplete(const WebSocketHeader &header_in) override; /**
* 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();
});
virtual void onWebSocketEncodeData(const uint8_t *ptr, uint64_t len); //设置sock否则shutdown等接口都无效
_delegate.setSock(HttpClientImp::_sock);
//触发连接成功事件
_delegate.onConnect(ex);
//拦截websocket数据接收
_onRecv = [this](const Buffer::Ptr &pBuf){
//解析websocket数据包
WebSocketSplitter::decode((uint8_t*)pBuf->data(),pBuf->size());
};
return;
}
//websocket握手失败或者tcp连接失败或者中途断开
if(_onRecv){
//握手成功之后的中途断开
_onRecv = nullptr;
_delegate.onErr(ex);
return;
}
//websocket握手失败或者tcp连接失败
_delegate.onConnect(ex);
}
private: private:
void onResponseCompleted_l(); string _Sec_WebSocket_Key;
function<void(const Buffer::Ptr &pBuf)> _onRecv;
protected: ClientTypeImp<ClientType,DataType> &_delegate;
bool _isHttps; string _payload;
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;
}; };
} /* namespace mediakit */
#endif /* Http_HttpClient_h */ /**
* Tcp客户端转WebSocket客户端模板
* TcpClient派生类任何代码的情况下快速实现WebSocket协议的包装
* @tparam ClientType TcpClient派生类
* @tparam DataType websocket负载类型TEXT还是BINARY类型
* @tparam useWSS 使ws还是wss连接
*/
template <typename ClientType,WebSocketHeader::Type DataType = WebSocketHeader::TEXT,bool useWSS = false >
class WebSocketClient : public ClientTypeImp<ClientType,DataType>{
public:
typedef std::shared_ptr<WebSocketClient> Ptr;
template<typename ...ArgsType>
WebSocketClient(ArgsType &&...args) : ClientTypeImp<ClientType,DataType>(std::forward<ArgsType>(args)...){
_wsClient.reset(new HttpWsClient<ClientType,DataType>(*this));
}
~WebSocketClient() override {}
/**
* startConnect方法
* TcpClient的连接服务器行为使WebSocket握手
* @param strUrl websocket服务器ip或域名
* @param iPort websocket服务器端口
* @param fTimeOutSec
*/
void startConnect(const string &strUrl, uint16_t iPort, float fTimeOutSec = 3) override {
string ws_url;
if(useWSS){
//加密的ws
ws_url = StrPrinter << "wss://" + strUrl << ":" << iPort << "/" ;
}else{
//明文ws
ws_url = StrPrinter << "ws://" + strUrl << ":" << iPort << "/" ;
}
_wsClient->startWsClient(ws_url,fTimeOutSec);
}
private:
typename HttpWsClient<ClientType,DataType>::Ptr _wsClient;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_WebSocketClient_H

View File

@ -34,7 +34,7 @@
* WebSock协议下的具体业务协议WebSocket协议的Rtmp协议等 * WebSock协议下的具体业务协议WebSocket协议的Rtmp协议等
* @tparam SessionType TcpSession类 * @tparam SessionType TcpSession类
*/ */
template <class SessionType,class HttpSessionType = HttpSession> template <class SessionType,class HttpSessionType = HttpSession,WebSocketHeader::Type DataType = WebSocketHeader::TEXT>
class WebSocketSession : public HttpSessionType { class WebSocketSession : public HttpSessionType {
public: public:
WebSocketSession(const Socket::Ptr &pSock) : HttpSessionType(pSock){} WebSocketSession(const Socket::Ptr &pSock) : HttpSessionType(pSock){}
@ -61,6 +61,37 @@ public:
_weakServer = const_cast<TcpServer &>(server).shared_from_this(); _weakServer = const_cast<TcpServer &>(server).shared_from_this();
} }
protected: protected:
/**
* websocket客户端连接上事件
* @param header http头
* @return true代表允许websocket连接
*/
bool onWebSocketConnect(const Parser &header) override{
//创建websocket session类
_session = std::make_shared<SessionImp>(HttpSessionType::getIdentifier(),HttpSessionType::_sock);
auto strongServer = _weakServer.lock();
if(strongServer){
_session->attachServer(*strongServer);
}
//此处截取数据并进行websocket协议打包
weak_ptr<WebSocketSession> weakSelf = dynamic_pointer_cast<WebSocketSession>(HttpSessionType::shared_from_this());
_session->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 = false;
strongSelf->WebSocketSplitter::encode(header,buf);
}
return buf->size();
});
//允许websocket客户端
return true;
}
/** /**
* webSocket数据包 * webSocket数据包
* @param packet * @param packet
@ -68,34 +99,6 @@ protected:
void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{ void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{
//新包,原来的包残余数据清空掉 //新包,原来的包残余数据清空掉
_remian_data.clear(); _remian_data.clear();
if(_firstPacket){
//这是个WebSocket会话而不是普通的Http会话
_firstPacket = false;
_session = std::make_shared<SessionImp>(HttpSessionType::getIdentifier(),HttpSessionType::_sock);
auto strongServer = _weakServer.lock();
if(strongServer){
//此处截取数据并进行websocket协议打包
weak_ptr<WebSocketSession> weakSelf = dynamic_pointer_cast<WebSocketSession>(HttpSessionType::shared_from_this());
_session->setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf) {
auto strongSelf = weakSelf.lock();
if (strongSelf) {
WebSocketHeader header;
header._fin = true;
header._reserved = 0;
header._opcode = WebSocketHeader::TEXT;
header._mask_flag = false;
strongSelf->WebSocketSplitter::encode(header, (uint8_t *)buf->data(), buf->size());
}
return buf->size();
});
_session->attachServer(*strongServer);
}
}
} }
/** /**
@ -124,7 +127,7 @@ protected:
} }
break; break;
case WebSocketHeader::PING:{ case WebSocketHeader::PING:{
const_cast<WebSocketHeader&>(header)._opcode = WebSocketHeader::PONG; header._opcode = WebSocketHeader::PONG;
HttpSessionType::encode(header,std::make_shared<BufferString>(_remian_data)); HttpSessionType::encode(header,std::make_shared<BufferString>(_remian_data));
} }
break; break;
@ -191,42 +194,10 @@ private:
string _identifier; string _identifier;
}; };
private: private:
bool _firstPacket = true;
string _remian_data; string _remian_data;
weak_ptr<TcpServer> _weakServer; weak_ptr<TcpServer> _weakServer;
std::shared_ptr<SessionImp> _session; std::shared_ptr<SessionImp> _session;
}; };
/**
*
*/
class EchoSession : public TcpSession {
public:
EchoSession(const Socket::Ptr &pSock) : TcpSession(pSock){
DebugL;
}
virtual ~EchoSession(){
DebugL;
}
void attachServer(const TcpServer &server) override{
DebugL << getIdentifier() << " " << TcpSession::getIdentifier();
}
void onRecv(const Buffer::Ptr &buffer) override {
send(buffer);
}
void onError(const SockException &err) override{
WarnL << err.what();
}
//每隔一段时间触发,用来做超时管理
void onManager() override{
DebugL;
}
};
typedef WebSocketSession<EchoSession,HttpSession> EchoWebSocketSession;
typedef WebSocketSession<EchoSession,HttpsSession> SSLEchoWebSocketSession;
#endif //ZLMEDIAKIT_WEBSOCKETSESSION_H #endif //ZLMEDIAKIT_WEBSOCKETSESSION_H

32
tests/README.md Normal file
View File

@ -0,0 +1,32 @@
此目录下的所有.cpp文件将被编译成可执行程序(不包含此目录下的子目录).
子目录DeviceHK为海康IPC的适配程序,需要先下载海康的SDK才能编译,
由于操作麻烦,所以仅把源码放在这仅供参考.
- test_benchmark.cpp
rtsp/rtmp性能测试客户端
- test_httpApi.cpp
http api 测试服务器
- test_httpClient.cpp
http 测试客户端
- test_player.cpp
rtsp/rtmp带视频渲染的客户端
- test_pusher.cpp
先拉流再推流的测试客户端
- test_pusherMp4.cpp
解复用mp4文件再推流的测试客户端
- test_server.cpp
rtsp/rtmp/http等服务器
- test_wsClient.cpp
websocket测试客户端
- test_wsServer.cpp
websocket回显测试服务器

View File

@ -1,3 +0,0 @@
此目录下的所有.cpp文件将被编译成可执行程序(不包含此目录下的子目录).
子目录DeviceHK为海康IPC的适配程序,需要先下载海康的SDK才能编译,
由于操作麻烦,所以仅把源码放在这仅供参考.

View File

@ -124,11 +124,11 @@ int main(int argc,char *argv[]){
//开启http服务器 //开启http服务器
TcpServer::Ptr httpSrv(new TcpServer()); TcpServer::Ptr httpSrv(new TcpServer());
httpSrv->start<EchoWebSocketSession>(mINI::Instance()[Http::kPort]);//默认80 httpSrv->start<HttpSession>(mINI::Instance()[Http::kPort]);//默认80
//如果支持ssl还可以开启https服务器 //如果支持ssl还可以开启https服务器
TcpServer::Ptr httpsSrv(new TcpServer()); TcpServer::Ptr httpsSrv(new TcpServer());
httpsSrv->start<SSLEchoWebSocketSession>(mINI::Instance()[Http::kSSLPort]);//默认443 httpsSrv->start<HttpsSession>(mINI::Instance()[Http::kSSLPort]);//默认443
InfoL << "你可以在浏览器输入:http://127.0.0.1/api/my_api?key0=val0&key1=参数1" << endl; InfoL << "你可以在浏览器输入:http://127.0.0.1/api/my_api?key0=val0&key1=参数1" << endl;

View File

@ -300,13 +300,13 @@ int main(int argc,char *argv[]) {
shellSrv->start<ShellSession>(shellPort); shellSrv->start<ShellSession>(shellPort);
rtspSrv->start<RtspSession>(rtspPort);//默认554 rtspSrv->start<RtspSession>(rtspPort);//默认554
rtmpSrv->start<RtmpSession>(rtmpPort);//默认1935 rtmpSrv->start<RtmpSession>(rtmpPort);//默认1935
//http服务器,支持websocket //http服务器
httpSrv->start<EchoWebSocketSession>(httpPort);//默认80 httpSrv->start<HttpSession>(httpPort);//默认80
//如果支持ssl还可以开启https服务器 //如果支持ssl还可以开启https服务器
TcpServer::Ptr httpsSrv(new TcpServer()); TcpServer::Ptr httpsSrv(new TcpServer());
//https服务器,支持websocket //https服务器
httpsSrv->start<SSLEchoWebSocketSession>(httpsPort);//默认443 httpsSrv->start<HttpsSession>(httpsPort);//默认443
//支持ssl加密的rtsp服务器可用于诸如亚马逊echo show这样的设备访问 //支持ssl加密的rtsp服务器可用于诸如亚马逊echo show这样的设备访问
TcpServer::Ptr rtspSSLSrv(new TcpServer()); TcpServer::Ptr rtspSSLSrv(new TcpServer());
@ -332,12 +332,12 @@ int main(int argc,char *argv[]) {
} }
if(httpPort != mINI::Instance()[Http::kPort].as<uint16_t>()){ if(httpPort != mINI::Instance()[Http::kPort].as<uint16_t>()){
httpPort = mINI::Instance()[Http::kPort]; httpPort = mINI::Instance()[Http::kPort];
httpSrv->start<EchoWebSocketSession>(httpPort); httpSrv->start<HttpSession>(httpPort);
InfoL << "重启http服务器" << httpPort; InfoL << "重启http服务器" << httpPort;
} }
if(httpsPort != mINI::Instance()[Http::kSSLPort].as<uint16_t>()){ if(httpsPort != mINI::Instance()[Http::kSSLPort].as<uint16_t>()){
httpsPort = mINI::Instance()[Http::kSSLPort]; httpsPort = mINI::Instance()[Http::kSSLPort];
httpsSrv->start<SSLEchoWebSocketSession>(httpsPort); httpsSrv->start<HttpsSession>(httpsPort);
InfoL << "重启https服务器" << httpsPort; InfoL << "重启https服务器" << httpsPort;
} }

84
tests/test_wsClient.cpp Normal file
View File

@ -0,0 +1,84 @@
/*
* 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.
*/
#include <signal.h>
#include <string>
#include <iostream>
#include "Util/MD5.h"
#include "Util/logger.h"
#include "Http/WebSocketClient.h"
using namespace std;
using namespace toolkit;
using namespace mediakit;
class EchoTcpClient : public TcpClient {
public:
EchoTcpClient(const EventPoller::Ptr &poller = nullptr){
InfoL;
}
~EchoTcpClient() override {
InfoL;
}
protected:
void onRecv(const Buffer::Ptr &pBuf) override {
DebugL << pBuf->toString();
}
//被动断开连接回调
void onErr(const SockException &ex) override {
WarnL << ex.what();
}
//tcp连接成功后每2秒触发一次该事件
void onManager() override {
send("echo test!");
DebugL << "send echo test";
}
//连接服务器结果回调
void onConnect(const SockException &ex) override{
DebugL << ex.what();
}
//数据全部发送完毕后回调
void onFlush() override{
DebugL;
}
};
int main(int argc, char *argv[]) {
//设置退出信号处理函数
static semaphore sem;
signal(SIGINT, [](int) { sem.post(); });// 设置退出信号
//设置日志
Logger::Instance().add(std::make_shared<ConsoleChannel>());
Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>());
WebSocketClient<EchoTcpClient>::Ptr client = std::make_shared<WebSocketClient<EchoTcpClient> >();
client->startConnect("121.40.165.18",8800);
sem.wait();
return 0;
}

88
tests/test_wsServer.cpp Normal file
View File

@ -0,0 +1,88 @@
/*
* 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.
*/
#include <signal.h>
#include <string>
#include <iostream>
#include "Util/MD5.h"
#include "Util/logger.h"
#include "Http/WebSocketSession.h"
using namespace std;
using namespace toolkit;
using namespace mediakit;
/**
*
*/
class EchoSession : public TcpSession {
public:
EchoSession(const Socket::Ptr &pSock) : TcpSession(pSock){
DebugL;
}
virtual ~EchoSession(){
DebugL;
}
void attachServer(const TcpServer &server) override{
DebugL << getIdentifier() << " " << TcpSession::getIdentifier();
}
void onRecv(const Buffer::Ptr &buffer) override {
//回显数据
send(buffer);
}
void onError(const SockException &err) override{
WarnL << err.what();
}
//每隔一段时间触发,用来做超时管理
void onManager() override{
DebugL;
}
};
int main(int argc, char *argv[]) {
//设置日志
Logger::Instance().add(std::make_shared<ConsoleChannel>());
Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>());
SSL_Initor::Instance().loadCertificate((exeDir() + "ssl.p12").data());
TcpServer::Ptr httpSrv(new TcpServer());
//http服务器,支持websocket
httpSrv->start<WebSocketSession<EchoSession,HttpSession>>(80);//默认80
TcpServer::Ptr httpsSrv(new TcpServer());
//https服务器,支持websocket
httpsSrv->start<WebSocketSession<EchoSession,HttpsSession>>(443);//默认443
DebugL << "请打开网页:http://www.websocket-test.com/,连接 ws://127.0.0.1/测试";
//设置退出信号处理函数
static semaphore sem;
signal(SIGINT, [](int) { sem.post(); });// 设置退出信号
sem.wait();
return 0;
}