amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-26 04:31:37 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

适配ZLToolKit develop分支

Browse Source
This commit is contained in:
xiongziliang 2018-02-23 15:36:51 +08:00
parent b0a64d1e6f
commit f1b4a196c6
30 changed files with 137 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
README.md
View File

@ -1,10 +1,10 @@
# 一个基于C++11简单易用的轻量级流媒体库 # 一个基于C++11简单易用的轻量级流媒体库
平台|编译状态 平台|编译状态
----|------- ----|-------
Linux | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit.svg?branch=master)](https://travis-ci.org/xiongziliang/ZLMediaKit) Linux | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit.svg?branch=2.0-alpha)](https://travis-ci.org/xiongziliang/ZLMediaKit)
macOS | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_mac.svg?branch=master)](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_mac) macOS | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_mac.svg?branch=2.0-alpha)](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_mac)
iOS | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit-build_for_ios.svg?branch=master)](https://travis-ci.org/xiongziliang/ZLMediaKit-build_for_ios) iOS | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit-build_for_ios.svg?branch=2.0-alpha)](https://travis-ci.org/xiongziliang/ZLMediaKit-build_for_ios)
Android | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_android.svg?branch=master)](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_android) Android | [![Build Status](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_android.svg?branch=2.0-alpha)](https://travis-ci.org/xiongziliang/ZLMediaKit_build_for_android)
Windows | 已经完成移植 Windows | 已经完成移植
## 项目特点 ## 项目特点
@ -124,15 +124,15 @@ Windows | 已经完成移植
## 使用方法 ## 使用方法
- 作为服务器： - 作为服务器：
``` ```
TcpServer<RtspSession>::Ptr rtspSrv(new TcpServer<RtspSession>()); TcpServer::Ptr rtspSrv(new TcpServer());
TcpServer<RtmpSession>::Ptr rtmpSrv(new TcpServer<RtmpSession>()); TcpServer::Ptr rtmpSrv(new TcpServer());
TcpServer<HttpSession>::Ptr httpSrv(new TcpServer<HttpSession>()); TcpServer::Ptr httpSrv(new TcpServer());
TcpServer<HttpsSession>::Ptr httpsSrv(new TcpServer<HttpsSession>()); TcpServer::Ptr httpsSrv(new TcpServer());
rtspSrv->start(mINI::Instance()[Config::Rtsp::kPort]); rtspSrv->start<RtspSession>(mINI::Instance()[Config::Rtsp::kPort]);
rtmpSrv->start(mINI::Instance()[Config::Rtmp::kPort]); rtmpSrv->start<RtmpSession>(mINI::Instance()[Config::Rtmp::kPort]);
httpSrv->start(mINI::Instance()[Config::Http::kPort]); httpSrv->start<HttpSession>(mINI::Instance()[Config::Http::kPort]);
httpsSrv->start(mINI::Instance()[Config::Http::kSSLPort]); httpsSrv->start<HttpsSession>(mINI::Instance()[Config::Http::kSSLPort]);
EventPoller::Instance().runLoop(); EventPoller::Instance().runLoop();
``` ```

3
build_for_android.sh
View File

@ -1,12 +1,13 @@
#!/bin/bash #!/bin/bash
path=`pwd` path=`pwd`
wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/master/build_for_android.sh -O toolkit_build.sh wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/develop/build_for_android.sh -O toolkit_build.sh
sudo chmod +x ./toolkit_build.sh sudo chmod +x ./toolkit_build.sh
./toolkit_build.sh ./toolkit_build.sh
cd $path cd $path
cd .. cd ..
git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git
cd ZLMediaKit cd ZLMediaKit
git checkout 2.0-alpha
mkdir -p android_build mkdir -p android_build
rm -rf ./build rm -rf ./build
ln -s ./android_build build ln -s ./android_build build

3
build_for_linux.sh
View File

@ -1,6 +1,6 @@
#!/bin/bash #!/bin/bash
path=`pwd` path=`pwd`
wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/master/build_for_linux.sh -O ./toolkit_build.sh wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/develop/build_for_linux.sh -O ./toolkit_build.sh
sudo chmod +x ./toolkit_build.sh sudo chmod +x ./toolkit_build.sh
./toolkit_build.sh ./toolkit_build.sh
sudo apt-get install libx264-dev sudo apt-get install libx264-dev
@ -13,6 +13,7 @@ cd $path
cd .. cd ..
git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git
cd ZLMediaKit cd ZLMediaKit
git checkout 2.0-alpha
mkdir -p linux_build mkdir -p linux_build
rm -rf ./build rm -rf ./build
ln -s ./linux_build build ln -s ./linux_build build

3
build_for_mac.sh
View File

@ -1,6 +1,6 @@
#!/bin/bash #!/bin/bash
path=`pwd` path=`pwd`
wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/master/build_for_mac.sh -O toolkit_build.sh wget https://raw.githubusercontent.com/xiongziliang/ZLToolKit/develop/build_for_mac.sh -O toolkit_build.sh
sudo chmod +x ./toolkit_build.sh sudo chmod +x ./toolkit_build.sh
./toolkit_build.sh ./toolkit_build.sh
brew install x264 brew install x264
@ -12,6 +12,7 @@ cd $path
cd .. cd ..
git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git git clone --depth=50 https://github.com/xiongziliang/ZLMediaKit.git
cd ZLMediaKit cd ZLMediaKit
git checkout 2.0-alpha
mkdir -p mac_build mkdir -p mac_build
rm -rf ./build rm -rf ./build
ln -s ./mac_build build ln -s ./mac_build build

2
src/Common/config.h
View File

@ -44,8 +44,6 @@ namespace Config {
//默认配置文件名为 /path/to/your/exe.ini //默认配置文件名为 /path/to/your/exe.ini
//加载配置文件成功后返回true，否则返回false //加载配置文件成功后返回true，否则返回false
bool loadIniConfig(const char *ini_path = nullptr); bool loadIniConfig(const char *ini_path = nullptr);
////////////TCP最大连接数///////////
#define MAX_TCP_SESSION 100000
////////////其他宏定义/////////// ////////////其他宏定义///////////
#ifndef MAX #ifndef MAX
#define MAX(a,b) ((a) > (b) ? (a) : (b) ) #define MAX(a,b) ((a) > (b) ? (a) : (b) )

2
src/Http/HttpClient.cpp
View File

@ -110,7 +110,7 @@ void HttpClient::onConnect(const SockException &ex) {
send(_body); send(_body);
} }
} }
void HttpClient::onRecv(const Socket::Buffer::Ptr &pBuf) { void HttpClient::onRecv(const Buffer::Ptr &pBuf) {
onRecvBytes(pBuf->data(),pBuf->size()); onRecvBytes(pBuf->data(),pBuf->size());
} }

2
src/Http/HttpClient.h
View File

@ -111,7 +111,7 @@ protected:
virtual void onDisconnect(const SockException &ex){} virtual void onDisconnect(const SockException &ex){}
private: private:
virtual void onConnect(const SockException &ex) override; virtual void onConnect(const SockException &ex) override;
virtual void onRecv(const Socket::Buffer::Ptr &pBuf) override; virtual void onRecv(const Buffer::Ptr &pBuf) override;
virtual void onErr(const SockException &ex) override; virtual void onErr(const SockException &ex) override;
//send //send

5
src/Http/HttpDownloader.cpp
View File

@ -44,6 +44,7 @@ HttpDownloader::~HttpDownloader() {
void HttpDownloader::startDownload(const string& url, const string& filePath,bool bAppend,uint32_t timeOutSecond) { void HttpDownloader::startDownload(const string& url, const string& filePath,bool bAppend,uint32_t timeOutSecond) {
_filePath = filePath; _filePath = filePath;
_timeOutSecond = timeOutSecond; _timeOutSecond = timeOutSecond;
_downloadTicker.resetTime();
if(_filePath.empty()){ if(_filePath.empty()){
_filePath = exeDir() + "HttpDownloader/" + MD5(url).hexdigest(); _filePath = exeDir() + "HttpDownloader/" + MD5(url).hexdigest();
} }
@ -67,6 +68,7 @@ void HttpDownloader::startDownload(const string& url, const string& filePath,boo
} }
void HttpDownloader::onResponseHeader(const string& status,const HttpHeader& headers) { void HttpDownloader::onResponseHeader(const string& status,const HttpHeader& headers) {
_downloadTicker.resetTime();
if(status != "200" && status != "206"){ if(status != "200" && status != "206"){
//失败 //失败
shutdown(); shutdown();
@ -81,6 +83,7 @@ void HttpDownloader::onResponseHeader(const string& status,const HttpHeader& hea
} }
void HttpDownloader::onResponseBody(const char* buf, size_t size, size_t recvedSize, size_t totalSize) { void HttpDownloader::onResponseBody(const char* buf, size_t size, size_t recvedSize, size_t totalSize) {
_downloadTicker.resetTime();
if(_saveFile){ if(_saveFile){
fwrite(buf,size,1,_saveFile); fwrite(buf,size,1,_saveFile);
} }
@ -126,7 +129,7 @@ void HttpDownloader::closeFile() {
} }
void HttpDownloader::onManager(){ void HttpDownloader::onManager(){
if(elapsedTime() > _timeOutSecond * 1000){ if(_downloadTicker.elapsedTime() > _timeOutSecond * 1000){
//超时 //超时
onDisconnect(SockException(Err_timeout,"download timeout")); onDisconnect(SockException(Err_timeout,"download timeout"));
shutdown(); shutdown();

1
src/Http/HttpDownloader.h
View File

@ -62,6 +62,7 @@ private:
onDownloadResult _onResult; onDownloadResult _onResult;
uint32_t _timeOutSecond; uint32_t _timeOutSecond;
bool _bDownloadSuccess = false; bool _bDownloadSuccess = false;
Ticker _downloadTicker;
}; };
} /* namespace Http */ } /* namespace Http */

40
src/Http/HttpSession.cpp
View File

@ -101,7 +101,7 @@ get_mime_type(const char* name) {
HttpSession::HttpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : HttpSession::HttpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) :
TcpLimitedSession(pTh, pSock) { TcpSession(pTh, pSock) {
GET_CONFIG_AND_REGISTER(string,rootPath,Config::Http::kRootPath); GET_CONFIG_AND_REGISTER(string,rootPath,Config::Http::kRootPath);
m_strPath = rootPath; m_strPath = rootPath;
@ -115,7 +115,7 @@ HttpSession::~HttpSession() {
//DebugL; //DebugL;
} }
void HttpSession::onRecv(const Socket::Buffer::Ptr &pBuf) { void HttpSession::onRecv(const Buffer::Ptr &pBuf) {
onRecv(pBuf->data(),pBuf->size()); onRecv(pBuf->data(),pBuf->size());
} }
void HttpSession::onRecv(const char *data,int size){ void HttpSession::onRecv(const char *data,int size){
@ -256,6 +256,11 @@ inline bool HttpSession::checkLiveFlvStream(){
}); });
//开始发送rtmp负载 //开始发送rtmp负载
//关闭tcp_nodelay ,优化性能
SockUtil::setNoDelay(_sock->rawFD(),false);
(*this) << SocketFlags(sock_flags);
m_pRingReader = mediaSrc->getRing()->attach(); m_pRingReader = mediaSrc->getRing()->attach();
weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this());
m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){ m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){
@ -392,9 +397,9 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
GET_CONFIG_AND_REGISTER(uint32_t,sendBufSize,Config::Http::kSendBufSize); GET_CONFIG_AND_REGISTER(uint32_t,sendBufSize,Config::Http::kSendBufSize);
//不允许主动丢包 //不允许主动丢包
sock->setShouldDropPacket(false); _sock->setShouldDropPacket(false);
//缓存大小为两个包,太大可能导致发送时间太长从而超时 //缓存大小为两个包,太大可能导致发送时间太长从而超时
sock->setSendPktSize(2); _sock->setSendPktSize(2);
weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this());
auto onFlush = [pFilePtr,bClose,weakSelf,piLeft]() { auto onFlush = [pFilePtr,bClose,weakSelf,piLeft]() {
TimeTicker(); TimeTicker();
@ -403,7 +408,7 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
//更新超时定时器 //更新超时定时器
strongSelf->m_ticker.resetTime(); strongSelf->m_ticker.resetTime();
//从循环池获取一个内存片 //从循环池获取一个内存片
auto sendBuf = strongSelf->sock->obtainBuffer(); auto sendBuf = strongSelf->obtainBuffer();
sendBuf->setCapacity(sendBufSize); sendBuf->setCapacity(sendBufSize);
//本次需要读取文件字节数 //本次需要读取文件字节数
int64_t iReq = MIN(sendBufSize,*piLeft); int64_t iReq = MIN(sendBufSize,*piLeft);
@ -421,8 +426,8 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
//InfoL << "send complete!" << iRead << " " << iReq << " " << *piLeft; //InfoL << "send complete!" << iRead << " " << iReq << " " << *piLeft;
if(iRead>0) { if(iRead>0) {
sendBuf->setSize(iRead); sendBuf->setSize(iRead);
strongSelf->sock->setSendPktSize(3);//强制写入socket缓存 strongSelf->_sock->setSendPktSize(3);//强制写入socket缓存
strongSelf->sock->send(sendBuf,sock_flags); strongSelf->send(sendBuf);
} }
if(bClose) { if(bClose) {
strongSelf->shutdown(); strongSelf->shutdown();
@ -431,7 +436,7 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
} }
//文件还未读完 //文件还未读完
sendBuf->setSize(iRead); sendBuf->setSize(iRead);
int iSent = strongSelf->sock->send(sendBuf,sock_flags); int iSent = strongSelf->send(sendBuf);
if(iSent == -1) { if(iSent == -1) {
//send error //send error
//InfoL << "send error"; //InfoL << "send error";
@ -451,9 +456,10 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
return false; return false;
}; };
//关闭tcp_nodelay ,优化性能 //关闭tcp_nodelay ,优化性能
SockUtil::setNoDelay(sock->rawFD(),false); SockUtil::setNoDelay(_sock->rawFD(),false);
(*this) << SocketFlags(sock_flags);
onFlush(); onFlush();
sock->setOnFlush(onFlush); _sock->setOnFlush(onFlush);
return Http_success; return Http_success;
} }
@ -690,7 +696,7 @@ public:
#pragma pack(pop) #pragma pack(pop)
#endif // defined(_WIN32) #endif // defined(_WIN32)
class BufferRtmp : public Socket::Buffer{ class BufferRtmp : public Buffer{
public: public:
typedef std::shared_ptr<BufferRtmp> Ptr; typedef std::shared_ptr<BufferRtmp> Ptr;
BufferRtmp(const RtmpPacket::Ptr & pkt):_rtmp(pkt){} BufferRtmp(const RtmpPacket::Ptr & pkt):_rtmp(pkt){}
@ -708,28 +714,28 @@ private:
void HttpSession::sendRtmp(const RtmpPacket::Ptr &pkt, uint32_t ui32TimeStamp) { void HttpSession::sendRtmp(const RtmpPacket::Ptr &pkt, uint32_t ui32TimeStamp) {
auto size = htonl(m_previousTagSize); auto size = htonl(m_previousTagSize);
sock->send((char *)&size,4,sock_flags);//send PreviousTagSize send((char *)&size,4);//send PreviousTagSize
RtmpTagHeader header; RtmpTagHeader header;
header.type = pkt->typeId; header.type = pkt->typeId;
set_be24(header.data_size, pkt->strBuf.size()); set_be24(header.data_size, pkt->strBuf.size());
header.timestamp_ex = (uint8_t) ((ui32TimeStamp >> 24) & 0xff); header.timestamp_ex = (uint8_t) ((ui32TimeStamp >> 24) & 0xff);
set_be24(header.timestamp,ui32TimeStamp & 0xFFFFFF); set_be24(header.timestamp,ui32TimeStamp & 0xFFFFFF);
sock->send((char *)&header, sizeof(header),sock_flags);//send tag header send((char *)&header, sizeof(header));//send tag header
sock->send(std::make_shared<BufferRtmp>(pkt),sock_flags);//send tag data send(std::make_shared<BufferRtmp>(pkt));//send tag data
m_previousTagSize += (pkt->strBuf.size() + sizeof(header) + 4); m_previousTagSize += (pkt->strBuf.size() + sizeof(header) + 4);
m_ticker.resetTime(); m_ticker.resetTime();
} }
void HttpSession::sendRtmp(uint8_t ui8Type, const std::string& strBuf, uint32_t ui32TimeStamp) { void HttpSession::sendRtmp(uint8_t ui8Type, const std::string& strBuf, uint32_t ui32TimeStamp) {
auto size = htonl(m_previousTagSize); auto size = htonl(m_previousTagSize);
sock->send((char *)&size,4,sock_flags);//send PreviousTagSize send((char *)&size,4);//send PreviousTagSize
RtmpTagHeader header; RtmpTagHeader header;
header.type = ui8Type; header.type = ui8Type;
set_be24(header.data_size, strBuf.size()); set_be24(header.data_size, strBuf.size());
header.timestamp_ex = (uint8_t) ((ui32TimeStamp >> 24) & 0xff); header.timestamp_ex = (uint8_t) ((ui32TimeStamp >> 24) & 0xff);
set_be24(header.timestamp,ui32TimeStamp & 0xFFFFFF); set_be24(header.timestamp,ui32TimeStamp & 0xFFFFFF);
sock->send((char *)&header, sizeof(header),sock_flags);//send tag header send((char *)&header, sizeof(header));//send tag header
sock->send(strBuf,sock_flags);//send tag data send(strBuf);//send tag data
m_previousTagSize += (strBuf.size() + sizeof(header) + 4); m_previousTagSize += (strBuf.size() + sizeof(header) + 4);
m_ticker.resetTime(); m_ticker.resetTime();
} }

6
src/Http/HttpSession.h
View File

@ -29,7 +29,7 @@
#include <functional> #include <functional>
#include "Common/config.h" #include "Common/config.h"
#include "Rtsp/Rtsp.h" #include "Rtsp/Rtsp.h"
#include "Network/TcpLimitedSession.h" #include "Network/TcpSession.h"
#include "Rtmp/RtmpMediaSource.h" #include "Rtmp/RtmpMediaSource.h"
using namespace std; using namespace std;
@ -40,7 +40,7 @@ namespace ZL {
namespace Http { namespace Http {
class HttpSession: public TcpLimitedSession<MAX_TCP_SESSION> { class HttpSession: public TcpSession {
public: public:
typedef StrCaseMap KeyValue; typedef StrCaseMap KeyValue;
typedef std::function<void(const string &codeOut, typedef std::function<void(const string &codeOut,
@ -50,7 +50,7 @@ public:
HttpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock); HttpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock);
virtual ~HttpSession(); virtual ~HttpSession();
virtual void onRecv(const Socket::Buffer::Ptr &) override; virtual void onRecv(const Buffer::Ptr &) override;
virtual void onError(const SockException &err) override; virtual void onError(const SockException &err) override;
virtual void onManager() override; virtual void onManager() override;

2
src/Http/HttpsSession.h
View File

@ -58,7 +58,7 @@ public:
virtual ~HttpsSession(){ virtual ~HttpsSession(){
//m_sslBox.shutdown(); //m_sslBox.shutdown();
} }
void onRecv(const Socket::Buffer::Ptr &pBuf) override{ void onRecv(const Buffer::Ptr &pBuf) override{
TimeTicker(); TimeTicker();
m_sslBox.onRecv(pBuf->data(), pBuf->size()); m_sslBox.onRecv(pBuf->data(), pBuf->size());
} }

2
src/Rtmp/RtmpPlayer.cpp
View File

@ -126,7 +126,7 @@ void RtmpPlayer::onConnect(const SockException &err){
strongSelf->send_connect(); strongSelf->send_connect();
}); });
} }
void RtmpPlayer::onRecv(const Socket::Buffer::Ptr &pBuf){ void RtmpPlayer::onRecv(const Buffer::Ptr &pBuf){
try { try {
onParseRtmp(pBuf->data(), pBuf->size()); onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) { } catch (exception &e) {

6
src/Rtmp/RtmpPlayer.h
View File

@ -99,7 +99,7 @@ private:
} }
//for Tcpclient //for Tcpclient
void onRecv(const Socket::Buffer::Ptr &pBuf) override; void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override; void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override; void onErr(const SockException &ex) override;
//fro RtmpProtocol //fro RtmpProtocol
@ -108,8 +108,8 @@ private:
void onSendRawData(const char *pcRawData, int iSize) override { void onSendRawData(const char *pcRawData, int iSize) override {
send(pcRawData, iSize); send(pcRawData, iSize);
} }
void onSendRawData(const Socket::Buffer::Ptr &buffer,int flags) override{ void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
m_pSock->send(buffer,flags); _sock->send(buffer,flags);
} }
template<typename FUN> template<typename FUN>

2
src/Rtmp/RtmpProtocol.cpp
View File

@ -204,7 +204,7 @@ void RtmpProtocol::sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId,
//估算rtmp包数据大小 //估算rtmp包数据大小
uint32_t capacity = ((bExtStamp ? 5 : 1) * (1 + (strBuf.size() / m_iChunkLenOut))) + strBuf.size() + sizeof(header); uint32_t capacity = ((bExtStamp ? 5 : 1) * (1 + (strBuf.size() / m_iChunkLenOut))) + strBuf.size() + sizeof(header);
uint32_t totalSize = 0; uint32_t totalSize = 0;
Socket::BufferRaw::Ptr buffer = m_bufferPool.obtain(); BufferRaw::Ptr buffer = m_bufferPool.obtain();
buffer->setCapacity(capacity); buffer->setCapacity(capacity);
memcpy(buffer->data() + totalSize,(char *) &header, sizeof(header)); memcpy(buffer->data() + totalSize,(char *) &header, sizeof(header));
totalSize += sizeof(header); totalSize += sizeof(header);

4
src/Rtmp/RtmpProtocol.h
View File

@ -55,7 +55,7 @@ public:
void reset(); void reset();
protected: protected:
virtual void onSendRawData(const char *pcRawData,int iSize) = 0; virtual void onSendRawData(const char *pcRawData,int iSize) = 0;
virtual void onSendRawData(const Socket::Buffer::Ptr &buffer,int flags) = 0; virtual void onSendRawData(const Buffer::Ptr &buffer,int flags) = 0;
virtual void onRtmpChunk(RtmpPacket &chunkData) = 0; virtual void onRtmpChunk(RtmpPacket &chunkData) = 0;
@ -85,7 +85,7 @@ protected:
int m_iNowStreamID = 0; int m_iNowStreamID = 0;
int m_iNowChunkID = 0; int m_iNowChunkID = 0;
bool m_bDataStarted = false; bool m_bDataStarted = false;
ResourcePool<Socket::BufferRaw,MAX_SEND_PKT> m_bufferPool; ResourcePool<BufferRaw,MAX_SEND_PKT> m_bufferPool;
private: private:
void handle_S0S1S2(const function<void()> &cb); void handle_S0S1S2(const function<void()> &cb);
void handle_C0C1(); void handle_C0C1();

2
src/Rtmp/RtmpPusher.cpp
View File

@ -131,7 +131,7 @@ void RtmpPusher::onConnect(const SockException &err){
strongSelf->send_connect(); strongSelf->send_connect();
}); });
} }
void RtmpPusher::onRecv(const Socket::Buffer::Ptr &pBuf){ void RtmpPusher::onRecv(const Buffer::Ptr &pBuf){
try { try {
onParseRtmp(pBuf->data(), pBuf->size()); onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) { } catch (exception &e) {

6
src/Rtmp/RtmpPusher.h
View File

@ -56,7 +56,7 @@ public:
protected: protected:
//for Tcpclient //for Tcpclient
void onRecv(const Socket::Buffer::Ptr &pBuf) override; void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override; void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override; void onErr(const SockException &ex) override;
@ -65,8 +65,8 @@ protected:
void onSendRawData(const char *pcRawData, int iSize) override { void onSendRawData(const char *pcRawData, int iSize) override {
send(pcRawData, iSize); send(pcRawData, iSize);
} }
void onSendRawData(const Socket::Buffer::Ptr &buffer,int flags) override{ void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
m_pSock->send(buffer,flags); _sock->send(buffer,flags);
} }
private: private:
void init(const RtmpMediaSource::Ptr &src); void init(const RtmpMediaSource::Ptr &src);

14
src/Rtmp/RtmpSession.cpp
View File

@ -33,7 +33,7 @@ namespace Rtmp {
unordered_map<string, RtmpSession::rtmpCMDHandle> RtmpSession::g_mapCmd; unordered_map<string, RtmpSession::rtmpCMDHandle> RtmpSession::g_mapCmd;
RtmpSession::RtmpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : RtmpSession::RtmpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) :
TcpLimitedSession(pTh, pSock) { TcpSession(pTh, pSock) {
static onceToken token([]() { static onceToken token([]() {
g_mapCmd.emplace("connect",&RtmpSession::onCmd_connect); g_mapCmd.emplace("connect",&RtmpSession::onCmd_connect);
g_mapCmd.emplace("createStream",&RtmpSession::onCmd_createStream); g_mapCmd.emplace("createStream",&RtmpSession::onCmd_createStream);
@ -43,11 +43,11 @@ RtmpSession::RtmpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::P
g_mapCmd.emplace("play2",&RtmpSession::onCmd_play2); g_mapCmd.emplace("play2",&RtmpSession::onCmd_play2);
g_mapCmd.emplace("seek",&RtmpSession::onCmd_seek); g_mapCmd.emplace("seek",&RtmpSession::onCmd_seek);
g_mapCmd.emplace("pause",&RtmpSession::onCmd_pause);}, []() {}); g_mapCmd.emplace("pause",&RtmpSession::onCmd_pause);}, []() {});
DebugL << getPeerIp(); DebugL << get_peer_ip();
} }
RtmpSession::~RtmpSession() { RtmpSession::~RtmpSession() {
DebugL << getPeerIp(); DebugL << get_peer_ip();
} }
void RtmpSession::onError(const SockException& err) { void RtmpSession::onError(const SockException& err) {
@ -64,20 +64,20 @@ void RtmpSession::onError(const SockException& err) {
void RtmpSession::onManager() { void RtmpSession::onManager() {
if (m_ticker.createdTime() > 10 * 1000) { if (m_ticker.createdTime() > 10 * 1000) {
if (!m_pRingReader && !m_pPublisherSrc) { if (!m_pRingReader && !m_pPublisherSrc) {
WarnL << "非法链接:" << getPeerIp(); WarnL << "非法链接:" << get_peer_ip();
shutdown(); shutdown();
} }
} }
if (m_pPublisherSrc) { if (m_pPublisherSrc) {
//publisher //publisher
if (m_ticker.elapsedTime() > 10 * 1000) { if (m_ticker.elapsedTime() > 10 * 1000) {
WarnL << "数据接收超时:" << getPeerIp(); WarnL << "数据接收超时:" << get_peer_ip();
shutdown(); shutdown();
} }
} }
} }
void RtmpSession::onRecv(const Socket::Buffer::Ptr &pBuf) { void RtmpSession::onRecv(const Buffer::Ptr &pBuf) {
m_ticker.resetTime(); m_ticker.resetTime();
try { try {
m_ui64TotalBytes += pBuf->size(); m_ui64TotalBytes += pBuf->size();
@ -273,7 +273,7 @@ void RtmpSession::doPlay(AMFDecoder &dec){
m_pRingReader = src->getRing()->attach(); m_pRingReader = src->getRing()->attach();
weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this()); weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
SockUtil::setNoDelay(sock->rawFD(), false); SockUtil::setNoDelay(_sock->rawFD(), false);
m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt) { m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt) {
auto strongSelf = weakSelf.lock(); auto strongSelf = weakSelf.lock();
if (!strongSelf) { if (!strongSelf) {

10
src/Rtmp/RtmpSession.h
View File

@ -36,7 +36,7 @@
#include "RtmpToRtspMediaSource.h" #include "RtmpToRtspMediaSource.h"
#include "Util/util.h" #include "Util/util.h"
#include "Util/TimeTicker.h" #include "Util/TimeTicker.h"
#include "Network/TcpLimitedSession.h" #include "Network/TcpSession.h"
using namespace ZL::Util; using namespace ZL::Util;
using namespace ZL::Network; using namespace ZL::Network;
@ -44,12 +44,12 @@ using namespace ZL::Network;
namespace ZL { namespace ZL {
namespace Rtmp { namespace Rtmp {
class RtmpSession: public TcpLimitedSession<MAX_TCP_SESSION> ,public RtmpProtocol , public MediaSourceEvent{ class RtmpSession: public TcpSession ,public RtmpProtocol , public MediaSourceEvent{
public: public:
typedef std::shared_ptr<RtmpSession> Ptr; typedef std::shared_ptr<RtmpSession> Ptr;
RtmpSession(const std::shared_ptr<ThreadPool> &_th, const Socket::Ptr &_sock); RtmpSession(const std::shared_ptr<ThreadPool> &_th, const Socket::Ptr &_sock);
virtual ~RtmpSession(); virtual ~RtmpSession();
void onRecv(const Socket::Buffer::Ptr &pBuf) override; void onRecv(const Buffer::Ptr &pBuf) override;
void onError(const SockException &err) override; void onError(const SockException &err) override;
void onManager() override; void onManager() override;
private: private:
@ -86,9 +86,9 @@ private:
m_ui64TotalBytes += iSize; m_ui64TotalBytes += iSize;
send(pcRawData, iSize); send(pcRawData, iSize);
} }
void onSendRawData(const Socket::Buffer::Ptr &buffer,int flags) override{ void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
m_ui64TotalBytes += buffer->size(); m_ui64TotalBytes += buffer->size();
sock->send(buffer,flags); _sock->send(buffer,flags);
} }
void onRtmpChunk(RtmpPacket &chunkData) override; void onRtmpChunk(RtmpPacket &chunkData) override;

4
src/Rtsp/RtspPlayer.cpp
View File

@ -173,7 +173,7 @@ void RtspPlayer::onConnect(const SockException &err){
})); }));
} }
void RtspPlayer::onRecv(const Socket::Buffer::Ptr& pBuf) { void RtspPlayer::onRecv(const Buffer::Ptr& pBuf) {
const char *buf = pBuf->data(); const char *buf = pBuf->data();
int size = pBuf->size(); int size = pBuf->size();
if (m_onHandshake) { if (m_onHandshake) {
@ -364,7 +364,7 @@ void RtspPlayer::HandleResSETUP(const Parser& parser, unsigned int uiTrackIndex)
} }
auto srcIP = inet_addr(get_peer_ip().data()); auto srcIP = inet_addr(get_peer_ip().data());
weak_ptr<RtspPlayer> weakSelf = dynamic_pointer_cast<RtspPlayer>(shared_from_this()); weak_ptr<RtspPlayer> weakSelf = dynamic_pointer_cast<RtspPlayer>(shared_from_this());
pUdpSockRef->setOnRead([srcIP,i,weakSelf](const Socket::Buffer::Ptr &buf, struct sockaddr *addr) { pUdpSockRef->setOnRead([srcIP,i,weakSelf](const Buffer::Ptr &buf, struct sockaddr *addr) {
auto strongSelf=weakSelf.lock(); auto strongSelf=weakSelf.lock();
if(!strongSelf) { if(!strongSelf) {
return; return;

2
src/Rtsp/RtspPlayer.h
View File

@ -108,7 +108,7 @@ private:
void play(const char* strUrl, const char *strUser, const char *strPwd, eRtpType eType); void play(const char* strUrl, const char *strUser, const char *strPwd, eRtpType eType);
void onConnect(const SockException &err) override; void onConnect(const SockException &err) override;
void onRecv(const Socket::Buffer::Ptr &pBuf) override; void onRecv(const Buffer::Ptr &pBuf) override;
void onErr(const SockException &ex) override; void onErr(const SockException &ex) override;
void HandleResSETUP(const Parser &parser, unsigned int uiTrackIndex); void HandleResSETUP(const Parser &parser, unsigned int uiTrackIndex);

44
src/Rtsp/RtspSession.cpp
View File

@ -56,7 +56,7 @@ recursive_mutex RtspSession::g_mtxGetter; //对quicktime上锁保护
recursive_mutex RtspSession::g_mtxPostter; //对quicktime上锁保护 recursive_mutex RtspSession::g_mtxPostter; //对quicktime上锁保护
unordered_map<string, RtspSession::rtspCMDHandle> RtspSession::g_mapCmd; unordered_map<string, RtspSession::rtspCMDHandle> RtspSession::g_mapCmd;
RtspSession::RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : RtspSession::RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) :
TcpLimitedSession(pTh, pSock), m_pSender(pSock) { TcpSession(pTh, pSock), m_pSender(pSock) {
static onceToken token( []() { static onceToken token( []() {
g_mapCmd.emplace("OPTIONS",&RtspSession::handleReq_Options); g_mapCmd.emplace("OPTIONS",&RtspSession::handleReq_Options);
g_mapCmd.emplace("DESCRIBE",&RtspSession::handleReq_Describe); g_mapCmd.emplace("DESCRIBE",&RtspSession::handleReq_Describe);
@ -76,21 +76,21 @@ RtspSession::RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::P
pSock->setSendPktSize(32); pSock->setSendPktSize(32);
#endif//__x86_64__ #endif//__x86_64__
DebugL << getPeerIp(); DebugL << get_peer_ip();
} }
RtspSession::~RtspSession() { RtspSession::~RtspSession() {
if (m_onDestory) { if (m_onDestory) {
m_onDestory(); m_onDestory();
} }
DebugL << getPeerIp(); DebugL << get_peer_ip();
} }
void RtspSession::shutdown(){ void RtspSession::shutdown(){
if (sock) { if (_sock) {
sock->emitErr(SockException(Err_other, "self shutdown")); _sock->emitErr(SockException(Err_other, "self shutdown"));
} }
if (m_bBase64need && !sock) { if (m_bBase64need && !_sock) {
//quickTime http postter,and self is detached from tcpServer //quickTime http postter,and self is detached from tcpServer
lock_guard<recursive_mutex> lock(g_mtxPostter); lock_guard<recursive_mutex> lock(g_mtxPostter);
g_mapPostter.erase(this); g_mapPostter.erase(this);
@ -108,7 +108,7 @@ void RtspSession::onError(const SockException& err) {
TraceL << err.getErrCode() << " " << err.what(); TraceL << err.getErrCode() << " " << err.what();
if (m_bListenPeerUdpData) { if (m_bListenPeerUdpData) {
//取消UDP端口监听 //取消UDP端口监听
UDPServer::Instance().stopListenPeer(getPeerIp().data(), this); UDPServer::Instance().stopListenPeer(get_peer_ip().data(), this);
m_bListenPeerUdpData = false; m_bListenPeerUdpData = false;
} }
if (!m_bBase64need && m_strSessionCookie.size() != 0) { if (!m_bBase64need && m_strSessionCookie.size() != 0) {
@ -119,7 +119,7 @@ void RtspSession::onError(const SockException& err) {
if (m_bBase64need && err.getErrCode() == Err_eof) { if (m_bBase64need && err.getErrCode() == Err_eof) {
//quickTime http postter,正在发送rtp; QuickTime只是断开了请求连接,请继续发送rtp //quickTime http postter,正在发送rtp; QuickTime只是断开了请求连接,请继续发送rtp
sock = nullptr; _sock = nullptr;
lock_guard<recursive_mutex> lock(g_mtxPostter); lock_guard<recursive_mutex> lock(g_mtxPostter);
//为了保证脱离TCPServer后还能正常运作,需要保持本对象的强引用 //为了保证脱离TCPServer后还能正常运作,需要保持本对象的强引用
g_mapPostter.emplace(this, dynamic_pointer_cast<RtspSession>(shared_from_this())); g_mapPostter.emplace(this, dynamic_pointer_cast<RtspSession>(shared_from_this()));
@ -136,19 +136,19 @@ void RtspSession::onError(const SockException& err) {
void RtspSession::onManager() { void RtspSession::onManager() {
if (m_ticker.createdTime() > 10 * 1000) { if (m_ticker.createdTime() > 10 * 1000) {
if (m_strSession.size() == 0) { if (m_strSession.size() == 0) {
WarnL << "非法链接:" << getPeerIp(); WarnL << "非法链接:" << get_peer_ip();
shutdown(); shutdown();
return; return;
} }
} }
if (m_rtpType != PlayerBase::RTP_TCP && m_ticker.elapsedTime() > 15 * 1000) { if (m_rtpType != PlayerBase::RTP_TCP && m_ticker.elapsedTime() > 15 * 1000) {
WarnL << "RTSP会话超时:" << getPeerIp(); WarnL << "RTSP会话超时:" << get_peer_ip();
shutdown(); shutdown();
return; return;
} }
} }
void RtspSession::onRecv(const Socket::Buffer::Ptr &pBuf) { void RtspSession::onRecv(const Buffer::Ptr &pBuf) {
m_ticker.resetTime(); m_ticker.resetTime();
char tmp[2 * 1024]; char tmp[2 * 1024];
m_pcBuf = tmp; m_pcBuf = tmp;
@ -556,14 +556,14 @@ bool RtspSession::handleReq_Setup() {
break; break;
case PlayerBase::RTP_UDP: { case PlayerBase::RTP_UDP: {
//我们用trackIdx区分rtp和rtcp包 //我们用trackIdx区分rtp和rtcp包
auto pSockRtp = UDPServer::Instance().getSock(getLocalIp().data(),2*trackIdx); auto pSockRtp = UDPServer::Instance().getSock(get_local_ip().data(),2*trackIdx);
if (!pSockRtp) { if (!pSockRtp) {
//分配端口失败 //分配端口失败
WarnL << "分配rtp端口失败"; WarnL << "分配rtp端口失败";
send_NotAcceptable(); send_NotAcceptable();
return false; return false;
} }
auto pSockRtcp = UDPServer::Instance().getSock(getLocalIp().data(),2*trackIdx + 1 ,pSockRtp->get_local_port() + 1); auto pSockRtcp = UDPServer::Instance().getSock(get_local_ip().data(),2*trackIdx + 1 ,pSockRtp->get_local_port() + 1);
if (!pSockRtcp) { if (!pSockRtcp) {
//分配端口失败 //分配端口失败
WarnL << "分配rtcp端口失败"; WarnL << "分配rtcp端口失败";
@ -577,7 +577,7 @@ bool RtspSession::handleReq_Setup() {
struct sockaddr_in peerAddr; struct sockaddr_in peerAddr;
peerAddr.sin_family = AF_INET; peerAddr.sin_family = AF_INET;
peerAddr.sin_port = htons(ui16PeerPort); peerAddr.sin_port = htons(ui16PeerPort);
peerAddr.sin_addr.s_addr = inet_addr(getPeerIp().data()); peerAddr.sin_addr.s_addr = inet_addr(get_peer_ip().data());
bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero); bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero);
m_apPeerUdpAddr[trackIdx].reset((struct sockaddr *) (new struct sockaddr_in(peerAddr))); m_apPeerUdpAddr[trackIdx].reset((struct sockaddr *) (new struct sockaddr_in(peerAddr)));
//尝试获取客户端nat映射地址 //尝试获取客户端nat映射地址
@ -601,7 +601,7 @@ bool RtspSession::handleReq_Setup() {
break; break;
case PlayerBase::RTP_MULTICAST: { case PlayerBase::RTP_MULTICAST: {
if(!m_pBrdcaster){ if(!m_pBrdcaster){
m_pBrdcaster = RtpBroadCaster::get(getLocalIp(),m_mediaInfo.m_vhost, m_mediaInfo.m_app, m_mediaInfo.m_streamid); m_pBrdcaster = RtpBroadCaster::get(get_local_ip(),m_mediaInfo.m_vhost, m_mediaInfo.m_app, m_mediaInfo.m_streamid);
if (!m_pBrdcaster) { if (!m_pBrdcaster) {
send_NotAcceptable(); send_NotAcceptable();
return false; return false;
@ -617,7 +617,7 @@ bool RtspSession::handleReq_Setup() {
} }
int iSrvPort = m_pBrdcaster->getPort(trackid); int iSrvPort = m_pBrdcaster->getPort(trackid);
//我们用trackIdx区分rtp和rtcp包 //我们用trackIdx区分rtp和rtcp包
auto pSockRtcp = UDPServer::Instance().getSock(getLocalIp().data(),2*trackIdx + 1,iSrvPort + 1); auto pSockRtcp = UDPServer::Instance().getSock(get_local_ip().data(),2*trackIdx + 1,iSrvPort + 1);
if (!pSockRtcp) { if (!pSockRtcp) {
//分配端口失败 //分配端口失败
WarnL << "分配rtcp端口失败"; WarnL << "分配rtcp端口失败";
@ -636,7 +636,7 @@ bool RtspSession::handleReq_Setup() {
m_iCseq, SERVER_NAME, m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME, RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), m_pBrdcaster->getIP().data(), dateHeader().data(), m_pBrdcaster->getIP().data(),
getLocalIp().data(), iSrvPort, pSockRtcp->get_local_port(), get_local_ip().data(), iSrvPort, pSockRtcp->get_local_port(),
udpTTL,printSSRC(trackRef.ssrc).data(), udpTTL,printSSRC(trackRef.ssrc).data(),
m_strSession.data()); m_strSession.data());
send(m_pcBuf, n); send(m_pcBuf, n);
@ -934,12 +934,12 @@ inline void RtspSession::sendRtpPacket(const RtpPacket::Ptr & pkt) {
} }
} }
inline void RtspSession::onRcvPeerUdpData(int iTrackIdx, const Socket::Buffer::Ptr &pBuf, const struct sockaddr& addr) { inline void RtspSession::onRcvPeerUdpData(int iTrackIdx, const Buffer::Ptr &pBuf, const struct sockaddr& addr) {
if(iTrackIdx % 2 == 0){ if(iTrackIdx % 2 == 0){
//这是rtp探测包 //这是rtp探测包
if(!m_bGotAllPeerUdp){ if(!m_bGotAllPeerUdp){
//还没有获取完整的rtp探测包 //还没有获取完整的rtp探测包
if(SockUtil::in_same_lan(getLocalIp().data(),getPeerIp().data())){ if(SockUtil::in_same_lan(get_local_ip().data(),get_peer_ip().data())){
//在内网中，客户端上报的端口号是真实的，所以我们忽略udp打洞包 //在内网中，客户端上报的端口号是真实的，所以我们忽略udp打洞包
m_bGotAllPeerUdp = true; m_bGotAllPeerUdp = true;
return; return;
@ -967,8 +967,8 @@ inline void RtspSession::onRcvPeerUdpData(int iTrackIdx, const Socket::Buffer::P
inline void RtspSession::startListenPeerUdpData() { inline void RtspSession::startListenPeerUdpData() {
m_bListenPeerUdpData = true; m_bListenPeerUdpData = true;
weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this()); weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
UDPServer::Instance().listenPeer(getPeerIp().data(), this, UDPServer::Instance().listenPeer(get_peer_ip().data(), this,
[weakSelf](int iTrackIdx,const Socket::Buffer::Ptr &pBuf,struct sockaddr *pPeerAddr)->bool { [weakSelf](int iTrackIdx,const Buffer::Ptr &pBuf,struct sockaddr *pPeerAddr)->bool {
auto strongSelf=weakSelf.lock(); auto strongSelf=weakSelf.lock();
if(!strongSelf) { if(!strongSelf) {
return false; return false;
@ -986,7 +986,7 @@ inline void RtspSession::startListenPeerUdpData() {
} }
inline void RtspSession::initSender(const std::shared_ptr<RtspSession>& session) { inline void RtspSession::initSender(const std::shared_ptr<RtspSession>& session) {
m_pSender = session->sock; m_pSender = session->_sock;
weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this()); weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
session->m_onDestory = [weakSelf]() { session->m_onDestory = [weakSelf]() {
auto strongSelf=weakSelf.lock(); auto strongSelf=weakSelf.lock();

12
src/Rtsp/RtspSession.h
View File

@ -37,7 +37,7 @@
#include "Player/PlayerBase.h" #include "Player/PlayerBase.h"
#include "Util/util.h" #include "Util/util.h"
#include "Util/logger.h" #include "Util/logger.h"
#include "Network/TcpLimitedSession.h" #include "Network/TcpSession.h"
using namespace std; using namespace std;
using namespace ZL::Util; using namespace ZL::Util;
@ -50,7 +50,7 @@ namespace Rtsp {
class RtspSession; class RtspSession;
class BufferRtp : public Socket::Buffer{ class BufferRtp : public Buffer{
public: public:
typedef std::shared_ptr<BufferRtp> Ptr; typedef std::shared_ptr<BufferRtp> Ptr;
BufferRtp(const RtpPacket::Ptr & pkt,uint32_t offset = 0 ):_rtp(pkt),_offset(offset){} BufferRtp(const RtpPacket::Ptr & pkt,uint32_t offset = 0 ):_rtp(pkt),_offset(offset){}
@ -67,7 +67,7 @@ private:
uint32_t _offset; uint32_t _offset;
}; };
class RtspSession: public TcpLimitedSession<MAX_TCP_SESSION> { class RtspSession: public TcpSession {
public: public:
typedef std::shared_ptr<RtspSession> Ptr; typedef std::shared_ptr<RtspSession> Ptr;
typedef std::function<void(const string &realm)> onGetRealm; typedef std::function<void(const string &realm)> onGetRealm;
@ -77,7 +77,7 @@ public:
RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock); RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock);
virtual ~RtspSession(); virtual ~RtspSession();
void onRecv(const Socket::Buffer::Ptr &pBuf) override; void onRecv(const Buffer::Ptr &pBuf) override;
void onError(const SockException &err) override; void onError(const SockException &err) override;
void onManager() override; void onManager() override;
private: private:
@ -94,7 +94,7 @@ private:
m_ui64TotalBytes += iSize; m_ui64TotalBytes += iSize;
return m_pSender->send(pcBuf, iSize); return m_pSender->send(pcBuf, iSize);
} }
int send(const Socket::Buffer::Ptr &pkt) override{ int send(const Buffer::Ptr &pkt) override{
m_ui64TotalBytes += pkt->size(); m_ui64TotalBytes += pkt->size();
return m_pSender->send(pkt,SOCKET_DEFAULE_FLAGS | FLAG_MORE); return m_pSender->send(pkt,SOCKET_DEFAULE_FLAGS | FLAG_MORE);
} }
@ -134,7 +134,7 @@ private:
} }
return -1; return -1;
} }
inline void onRcvPeerUdpData(int iTrackIdx, const Socket::Buffer::Ptr &pBuf, const struct sockaddr &addr); inline void onRcvPeerUdpData(int iTrackIdx, const Buffer::Ptr &pBuf, const struct sockaddr &addr);
inline void startListenPeerUdpData(); inline void startListenPeerUdpData();
//认证相关 //认证相关

2
src/Rtsp/UDPServer.cpp
View File

@ -87,7 +87,7 @@ void UDPServer::onErr(const string& strKey, const SockException& err) {
lock_guard<mutex> lck(m_mtxUpdSock); lock_guard<mutex> lck(m_mtxUpdSock);
m_mapUpdSock.erase(strKey); m_mapUpdSock.erase(strKey);
} }
void UDPServer::onRcvData(int iTrackIndex, const Socket::Buffer::Ptr &pBuf, struct sockaddr* pPeerAddr) { void UDPServer::onRcvData(int iTrackIndex, const Buffer::Ptr &pBuf, struct sockaddr* pPeerAddr) {
//TraceL << trackIndex; //TraceL << trackIndex;
struct sockaddr_in *in = (struct sockaddr_in *) pPeerAddr; struct sockaddr_in *in = (struct sockaddr_in *) pPeerAddr;
string peerIp = inet_ntoa(in->sin_addr); string peerIp = inet_ntoa(in->sin_addr);

4
src/Rtsp/UDPServer.h
View File

@ -44,7 +44,7 @@ namespace Rtsp {
class UDPServer { class UDPServer {
public: public:
typedef function< bool(int, const Socket::Buffer::Ptr &, struct sockaddr *)> onRecvData; typedef function< bool(int, const Buffer::Ptr &, struct sockaddr *)> onRecvData;
UDPServer(); UDPServer();
virtual ~UDPServer(); virtual ~UDPServer();
static UDPServer &Instance() { static UDPServer &Instance() {
@ -58,7 +58,7 @@ public:
void listenPeer(const char *strPeerIp, void *pSelf, const onRecvData &cb); void listenPeer(const char *strPeerIp, void *pSelf, const onRecvData &cb);
void stopListenPeer(const char *strPeerIp, void *pSelf); void stopListenPeer(const char *strPeerIp, void *pSelf);
private: private:
void onRcvData(int iTrackId, const Socket::Buffer::Ptr &pBuf,struct sockaddr *pPeerAddr); void onRcvData(int iTrackId, const Buffer::Ptr &pBuf,struct sockaddr *pPeerAddr);
void onErr(const string &strKey,const SockException &err); void onErr(const string &strKey,const SockException &err);
unordered_map<string, Socket::Ptr> m_mapUpdSock; unordered_map<string, Socket::Ptr> m_mapUpdSock;
mutex m_mtxUpdSock; mutex m_mtxUpdSock;

4
src/Shell/ShellSession.cpp
View File

@ -38,14 +38,14 @@ namespace Shell {
ShellSession::ShellSession(const std::shared_ptr<ThreadPool> &_th, ShellSession::ShellSession(const std::shared_ptr<ThreadPool> &_th,
const Socket::Ptr &_sock) : const Socket::Ptr &_sock) :
TcpLimitedSession(_th, _sock) { TcpSession(_th, _sock) {
pleaseInputUser(); pleaseInputUser();
} }
ShellSession::~ShellSession() { ShellSession::~ShellSession() {
} }
void ShellSession::onRecv(const Socket::Buffer::Ptr&buf) { void ShellSession::onRecv(const Buffer::Ptr&buf) {
//DebugL << hexdump(buf->data(), buf->size()); //DebugL << hexdump(buf->data(), buf->size());
GET_CONFIG_AND_REGISTER(uint32_t,maxReqSize,Config::Shell::kMaxReqSize); GET_CONFIG_AND_REGISTER(uint32_t,maxReqSize,Config::Shell::kMaxReqSize);
if (m_strRecvBuf.size() + buf->size() >= maxReqSize) { if (m_strRecvBuf.size() + buf->size() >= maxReqSize) {

6
src/Shell/ShellSession.h
View File

@ -30,7 +30,7 @@
#include <functional> #include <functional>
#include "Common/config.h" #include "Common/config.h"
#include "Util/TimeTicker.h" #include "Util/TimeTicker.h"
#include "Network/TcpLimitedSession.h" #include "Network/TcpSession.h"
using namespace std; using namespace std;
using namespace ZL::Util; using namespace ZL::Util;
@ -39,12 +39,12 @@ using namespace ZL::Network;
namespace ZL { namespace ZL {
namespace Shell { namespace Shell {
class ShellSession: public TcpLimitedSession<MAX_TCP_SESSION> { class ShellSession: public TcpSession {
public: public:
ShellSession(const std::shared_ptr<ThreadPool> &_th, const Socket::Ptr &_sock); ShellSession(const std::shared_ptr<ThreadPool> &_th, const Socket::Ptr &_sock);
virtual ~ShellSession(); virtual ~ShellSession();
void onRecv(const Socket::Buffer::Ptr &) override; void onRecv(const Buffer::Ptr &) override;
void onError(const SockException &err) override {}; void onError(const SockException &err) override {};
void onManager() override; void onManager() override;

8
tests/test_httpApi.cpp
View File

@ -112,13 +112,13 @@ int main(int argc,char *argv[]){
#endif //ENABLE_OPENSSL #endif //ENABLE_OPENSSL
//开启http服务器 //开启http服务器
TcpServer<HttpSession>::Ptr httpSrv(new TcpServer<HttpSession>()); TcpServer::Ptr httpSrv(new TcpServer());
httpSrv->start(mINI::Instance()[Config::Http::kPort]);//默认80 httpSrv->start<HttpSession>(mINI::Instance()[Config::Http::kPort]);//默认80
#ifdef ENABLE_OPENSSL #ifdef ENABLE_OPENSSL
//如果支持ssl，还可以开启https服务器 //如果支持ssl，还可以开启https服务器
TcpServer<HttpsSession>::Ptr httpsSrv(new TcpServer<HttpsSession>()); TcpServer::Ptr httpsSrv(new TcpServer());
httpsSrv->start(mINI::Instance()[Config::Http::kSSLPort]);//默认443 httpsSrv->start<HttpsSession>(mINI::Instance()[Config::Http::kSSLPort]);//默认443
#endif //ENABLE_OPENSSL #endif //ENABLE_OPENSSL
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;

31
tests/test_server.cpp
View File

@ -195,47 +195,48 @@ int main(int argc,char *argv[]) {
//简单的telnet服务器，可用于服务器调试，但是不能使用23端口，否则telnet上了莫名其妙的现象 //简单的telnet服务器，可用于服务器调试，但是不能使用23端口，否则telnet上了莫名其妙的现象
//测试方法:telnet 127.0.0.1 9000 //测试方法:telnet 127.0.0.1 9000
TcpServer<ShellSession>::Ptr shellSrv(new TcpServer<ShellSession>()); TcpServer::Ptr shellSrv(new TcpServer());
TcpServer<RtspSession>::Ptr rtspSrv(new TcpServer<RtspSession>()); TcpServer::Ptr rtspSrv(new TcpServer());
TcpServer<RtmpSession>::Ptr rtmpSrv(new TcpServer<RtmpSession>()); TcpServer::Ptr rtmpSrv(new TcpServer());
TcpServer<HttpSession>::Ptr httpSrv(new TcpServer<HttpSession>()); TcpServer::Ptr httpSrv(new TcpServer());
shellSrv->start(shellPort);
rtspSrv->start(rtspPort);//默认554 shellSrv->start<ShellSession>(shellPort);
rtmpSrv->start(rtmpPort);//默认1935 rtspSrv->start<RtspSession>(rtspPort);//默认554
httpSrv->start(httpPort);//默认80 rtmpSrv->start<RtmpSession>(rtmpPort);//默认1935
httpSrv->start<HttpSession>(httpPort);//默认80
#ifdef ENABLE_OPENSSL #ifdef ENABLE_OPENSSL
//如果支持ssl，还可以开启https服务器 //如果支持ssl，还可以开启https服务器
TcpServer<HttpsSession>::Ptr httpsSrv(new TcpServer<HttpsSession>()); TcpServer::Ptr httpsSrv(new TcpServer());
httpsSrv->start(httpsPort);//默认443 httpsSrv->start<HttpsSession>(httpsPort);//默认443
#endif //ENABLE_OPENSSL #endif //ENABLE_OPENSSL
NoticeCenter::Instance().addListener(ReloadConfigTag,Config::Broadcast::kBroadcastReloadConfig,[&](BroadcastReloadConfigArgs){ NoticeCenter::Instance().addListener(ReloadConfigTag,Config::Broadcast::kBroadcastReloadConfig,[&](BroadcastReloadConfigArgs){
//重新创建服务器 //重新创建服务器
if(shellPort != mINI::Instance()[Config::Shell::kPort].as<uint16_t>()){ if(shellPort != mINI::Instance()[Config::Shell::kPort].as<uint16_t>()){
shellPort = mINI::Instance()[Config::Shell::kPort]; shellPort = mINI::Instance()[Config::Shell::kPort];
shellSrv->start(shellPort); shellSrv->start<ShellSession>(shellPort);
InfoL << "重启shell服务器:" << shellPort; InfoL << "重启shell服务器:" << shellPort;
} }
if(rtspPort != mINI::Instance()[Config::Rtsp::kPort].as<uint16_t>()){ if(rtspPort != mINI::Instance()[Config::Rtsp::kPort].as<uint16_t>()){
rtspPort = mINI::Instance()[Config::Rtsp::kPort]; rtspPort = mINI::Instance()[Config::Rtsp::kPort];
rtspSrv->start(rtspPort); rtspSrv->start<RtspSession>(rtspPort);
InfoL << "重启rtsp服务器" << rtspPort; InfoL << "重启rtsp服务器" << rtspPort;
} }
if(rtmpPort != mINI::Instance()[Config::Rtmp::kPort].as<uint16_t>()){ if(rtmpPort != mINI::Instance()[Config::Rtmp::kPort].as<uint16_t>()){
rtmpPort = mINI::Instance()[Config::Rtmp::kPort]; rtmpPort = mINI::Instance()[Config::Rtmp::kPort];
rtmpSrv->start(rtmpPort); rtmpSrv->start<RtmpSession>(rtmpPort);
InfoL << "重启rtmp服务器" << rtmpPort; InfoL << "重启rtmp服务器" << rtmpPort;
} }
if(httpPort != mINI::Instance()[Config::Http::kPort].as<uint16_t>()){ if(httpPort != mINI::Instance()[Config::Http::kPort].as<uint16_t>()){
httpPort = mINI::Instance()[Config::Http::kPort]; httpPort = mINI::Instance()[Config::Http::kPort];
httpSrv->start(httpPort); httpSrv->start<HttpSession>(httpPort);
InfoL << "重启http服务器" << httpPort; InfoL << "重启http服务器" << httpPort;
} }
#ifdef ENABLE_OPENSSL #ifdef ENABLE_OPENSSL
if(httpsPort != mINI::Instance()[Config::Http::kSSLPort].as<uint16_t>()){ if(httpsPort != mINI::Instance()[Config::Http::kSSLPort].as<uint16_t>()){
httpsPort = mINI::Instance()[Config::Http::kSSLPort]; httpsPort = mINI::Instance()[Config::Http::kSSLPort];
httpsSrv->start(httpsPort); httpsSrv->start<HttpsSession>(httpsPort);
InfoL << "重启https服务器" << httpsPort; InfoL << "重启https服务器" << httpsPort;
} }
#endif //ENABLE_OPENSSL #endif //ENABLE_OPENSSL