diff --git a/ZLToolKit b/ZLToolKit index 666eb9ba..fd59a374 160000 --- a/ZLToolKit +++ b/ZLToolKit @@ -1 +1 @@ -Subproject commit 666eb9ba01939f739bf145abc5f02539ca2b9c97 +Subproject commit fd59a374a8d3f65c379d0395e66cd0f64559b520 diff --git a/src/Http/HttpSession.h b/src/Http/HttpSession.h index 3fbf967b..6a3ddfd0 100644 --- a/src/Http/HttpSession.h +++ b/src/Http/HttpSession.h @@ -124,6 +124,8 @@ private: }; +typedef TcpSessionWithSSL HttpsSession; + } /* namespace mediakit */ #endif /* SRC_HTTP_HTTPSESSION_H_ */ diff --git a/src/Http/WebSocketSession.h b/src/Http/WebSocketSession.h new file mode 100644 index 00000000..cf57fbd7 --- /dev/null +++ b/src/Http/WebSocketSession.h @@ -0,0 +1,235 @@ +/* + * MIT License + * + * Copyright (c) 2016 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_WEBSOCKETSESSION_H +#define ZLMEDIAKIT_WEBSOCKETSESSION_H + +#include "HttpSession.h" + +/** +* 通过该模板类可以透明化WebSocket协议, +* 用户只要实现WebSock协议下的具体业务协议,譬如基于WebSocket协议的Rtmp协议等 +* @tparam SessionType 业务协议的TcpSession类 +*/ +template +class WebSocketSession : public HttpSessionType { +public: + WebSocketSession(const std::shared_ptr &pTh, const Socket::Ptr &pSock) : HttpSessionType(pTh,pSock){} + virtual ~WebSocketSession(){} + + //收到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); + _weakServer = const_cast(server).shared_from_this(); + } +protected: + /** + * 开始收到一个webSocket数据包 + * @param packet + */ + void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{ + //新包,原来的包残余数据清空掉 + _remian_data.clear(); + + if(_firstPacket){ + //这是个WebSocket会话而不是普通的Http会话 + _firstPacket = false; + _session = std::make_shared(HttpSessionType::getIdentifier(),nullptr,HttpSessionType::_sock); + + auto strongServer = _weakServer.lock(); + if(strongServer){ + _session->attachServer(*strongServer); + } + + //此处截取数据并进行websocket协议打包 + weak_ptr weakSelf = dynamic_pointer_cast(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(); + }); + } + } + + /** + * 收到websocket数据包负载 + * @param packet + * @param ptr + * @param len + * @param recved + */ + void onWebSocketDecodePlayload(const WebSocketHeader &packet,const uint8_t *ptr,uint64_t len,uint64_t recved) override { + _remian_data.append((char *)ptr,len); + } + + /** + * 接收到完整的一个webSocket数据包后回调 + * @param header 数据包包头 + */ + void onWebSocketDecodeComplete(const WebSocketHeader &header_in) override { + WebSocketHeader& header = const_cast(header_in); + auto flag = header._mask_flag; + header._mask_flag = false; + + switch (header._opcode){ + case WebSocketHeader::CLOSE:{ + HttpSessionType::encode(header,nullptr,0); + } + break; + case WebSocketHeader::PING:{ + const_cast(header)._opcode = WebSocketHeader::PONG; + HttpSessionType::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(_remian_data); + _session->onRecv(buffer); + } + break; + default: + break; + } + _remian_data.clear(); + header._mask_flag = flag; + } + + /** + * 发送数据进行websocket协议打包后回调 + * @param ptr + * @param len + */ + void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len) override{ + SocketHelper::send((char *)ptr,len); + } +private: + typedef function onBeforeSendCB; + /** + * 该类实现了TcpSession派生类发送数据的截取 + * 目的是发送业务数据前进行websocket协议的打包 + */ + class SessionImp : public SessionType{ + public: + SessionImp(const string &identifier, + const std::shared_ptr &pTh, + const Socket::Ptr &pSock) : + _identifier(identifier),SessionType(pTh,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); + } + string getIdentifier() const override{ + return _identifier; + } + private: + onBeforeSendCB _beforeSendCB; + string _identifier; + }; +private: + bool _firstPacket = true; + string _remian_data; + weak_ptr _weakServer; + std::shared_ptr _session; +}; + +/** +* 回显会话 +*/ +class EchoSession : public TcpSession { +public: + EchoSession(const std::shared_ptr &pTh, const Socket::Ptr &pSock) : + TcpSession(pTh,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 EchoWebSocketSession; +typedef WebSocketSession SSLEchoWebSocketSession; + + +#endif //ZLMEDIAKIT_WEBSOCKETSESSION_H diff --git a/src/Rtsp/RtspSession.h b/src/Rtsp/RtspSession.h index 05ba91af..2cc73f48 100644 --- a/src/Rtsp/RtspSession.h +++ b/src/Rtsp/RtspSession.h @@ -41,7 +41,6 @@ #include "RtspSplitter.h" #include "RtpReceiver.h" #include "RtspToRtmpMediaSource.h" -#include "Http/HttpsSession.h" using namespace std; using namespace toolkit; diff --git a/tests/test_httpApi.cpp b/tests/test_httpApi.cpp index 922be8ce..f1c546af 100644 --- a/tests/test_httpApi.cpp +++ b/tests/test_httpApi.cpp @@ -27,16 +27,15 @@ #include #include #include -#include "Common/config.h" -#include "Util/SSLBox.h" -#include "Http/HttpsSession.h" #include "Util/File.h" +#include "Util/SSLBox.h" #include "Util/logger.h" #include "Util/onceToken.h" +#include "Util/NoticeCenter.h" #include "Network/TcpServer.h" #include "Poller/EventPoller.h" -#include "Thread/WorkThreadPool.h" -#include "Util/NoticeCenter.h" +#include "Common/config.h" +#include "Http/WebSocketSession.h" using namespace std; using namespace toolkit; diff --git a/tests/test_server.cpp b/tests/test_server.cpp index 45589be5..3219f1ef 100644 --- a/tests/test_server.cpp +++ b/tests/test_server.cpp @@ -27,27 +27,23 @@ #include #include #include + +#include "Util/MD5.h" +#include "Util/File.h" +#include "Util/logger.h" +#include "Util/SSLBox.h" +#include "Util/onceToken.h" +#include "Network/TcpServer.h" +#include "Poller/EventPoller.h" + #include "Common/config.h" #include "Rtsp/UDPServer.h" #include "Rtsp/RtspSession.h" #include "Rtmp/RtmpSession.h" -#include "Http/HttpSession.h" #include "Shell/ShellSession.h" -#include "Util/MD5.h" #include "RtmpMuxer/FlvMuxer.h" - -#ifdef ENABLE_OPENSSL -#include "Util/SSLBox.h" -#include "Http/HttpsSession.h" -#endif//ENABLE_OPENSSL - -#include "Util/File.h" -#include "Util/logger.h" -#include "Util/onceToken.h" -#include "Network/TcpServer.h" -#include "Poller/EventPoller.h" -#include "Thread/WorkThreadPool.h" #include "Player/PlayerProxy.h" +#include "Http/WebSocketSession.h" using namespace std; using namespace toolkit; @@ -279,7 +275,6 @@ int main(int argc,char *argv[]) { " rtsp地址 : rtsp://127.0.0.1/live/0\n" " rtmp地址 : rtmp://127.0.0.1/live/0"; -#ifdef ENABLE_OPENSSL //请把证书"test_server.pem"放置在本程序可执行程序同目录下 try { //加载证书,证书包含公钥和私钥 @@ -289,7 +284,6 @@ int main(int argc,char *argv[]) { proxyMap.clear(); return 0; } -#endif //ENABLE_OPENSSL uint16_t shellPort = mINI::Instance()[Shell::kPort]; uint16_t rtspPort = mINI::Instance()[Rtsp::kPort]; @@ -311,7 +305,6 @@ int main(int argc,char *argv[]) { //http服务器,支持websocket httpSrv->start(httpPort);//默认80 -#ifdef ENABLE_OPENSSL //如果支持ssl,还可以开启https服务器 TcpServer::Ptr httpsSrv(new TcpServer()); //https服务器,支持websocket @@ -320,7 +313,6 @@ int main(int argc,char *argv[]) { //支持ssl加密的rtsp服务器,可用于诸如亚马逊echo show这样的设备访问 TcpServer::Ptr rtspSSLSrv(new TcpServer()); rtspSSLSrv->start(rtspsPort);//默认322 -#endif //ENABLE_OPENSSL //服务器支持动态切换端口(不影响现有连接) NoticeCenter::Instance().addListener(ReloadConfigTag,Broadcast::kBroadcastReloadConfig,[&](BroadcastReloadConfigArgs){ @@ -345,7 +337,6 @@ int main(int argc,char *argv[]) { httpSrv->start(httpPort); InfoL << "重启http服务器" << httpPort; } -#ifdef ENABLE_OPENSSL if(httpsPort != mINI::Instance()[Http::kSSLPort].as()){ httpsPort = mINI::Instance()[Http::kSSLPort]; httpsSrv->start(httpsPort); @@ -357,7 +348,6 @@ int main(int argc,char *argv[]) { rtspSSLSrv->start(rtspsPort); InfoL << "重启rtsps服务器" << rtspsPort; } -#endif //ENABLE_OPENSSL }); EventPoller::Instance().runLoop();