Skip to content

Z
ZLMediaKit
Commit 05e6d325 by xiongziliang
Options

完善webSocket协议实现 

优化代码
parent bf88cc01
正在显示 包含 273 行增加 和 153 行删除
ZLToolKit @ bb2ba900
Subproject commit 4916fec31fcbe383ae4d38570b84858f6ec7747d
Subproject commit bb2ba9005b0191f709897bf51e5bb719eac856bb
src/Http/HttpClientImp.cpp
......@@ -57,7 +57,7 @@ void HttpClientImp::sendRequest(const string& url,float fTimeOutSec) {
#if defined(__GNUC__) && (__GNUC__ < 5)
public_send(data,len);
#else//defined(__GNUC__) && (__GNUC__ < 5)
HttpClient::send(data,len);
HttpClient::send(obtainBuffer(data,len));
#endif//defined(__GNUC__) && (__GNUC__ < 5)
});
#endif //ENABLE_OPENSSL
......@@ -78,28 +78,12 @@ void HttpClientImp::onRecvBytes(const char* data, int size) {
}
}
int HttpClientImp::send(const string& str) {
int HttpClientImp::send(const Buffer::Ptr &buf) {
if(_sslBox){
_sslBox->onSend(str.data(),str.size());
return str.size();
_sslBox->onSend(buf->data(),buf->size());
return buf->size();
}
return HttpClient::send(str);
}
int HttpClientImp::send(string &&str){
if(_sslBox){
_sslBox->onSend(str.data(),str.size());
return str.size();
}
return HttpClient::send(std::move(str));
}
int HttpClientImp::send(const char* str, int len) {
if(_sslBox){
_sslBox->onSend(str,len);
return len;
}
return HttpClient::send(str,len);
return HttpClient::send(buf);
}
#endif //ENABLE_OPENSSL
......
src/Http/HttpClientImp.h
......@@ -48,15 +48,13 @@ public:
HttpClient::onRecvBytes(data,len);
}
void public_send(const char *data, uint32_t len){
HttpClient::send(data,len);
HttpClient::send(obtainBuffer(data,len));
}
#endif //defined(__GNUC__) && (__GNUC__ < 5)
private:
#ifdef ENABLE_OPENSSL
virtual void onRecvBytes(const char *data,int size) override;
virtual int send(const string &str) override;
virtual int send(string &&str) override;
virtual int send(const char *str, int len) override;
virtual int send(const Buffer::Ptr &buf) override;
std::shared_ptr<SSL_Box> _sslBox;
#endif //ENABLE_OPENSSL
};
......
src/Http/HttpRequestSplitter.cpp
......@@ -43,6 +43,16 @@ void HttpRequestSplitter::input(const char *data,uint64_t len) {
splitPacket:
/*确保ptr最后一个字节是0，防止strstr越界
*由于ZLToolKit确保内存最后一个字节是保留未使用字节并置0，
*所以此处可以不用再次置0
*但是上层数据可能来自其他渠道，保险起见还是置0
*/
char &tail_ref = ((char *) ptr)[len];
char tail_tmp = tail_ref;
tail_ref = 0;
//数据按照请求头处理
const char *index = nullptr;
while (_content_len == 0 && (index = strstr(ptr,"\r\n\r\n")) != nullptr) {
......@@ -51,6 +61,11 @@ splitPacket:
ptr = index + 4;
}
/*
* 恢复末尾字节
*/
tail_ref = tail_tmp;
uint64_t remain = len - (ptr - data);
if(remain <= 0){
//没有剩余数据，清空缓存
......
src/Http/HttpSession.cpp
......@@ -49,7 +49,7 @@ using namespace ZL::Util;
namespace ZL {
namespace Http {
static int sock_flags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
static int kSockFlags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
string dateStr() {
char buf[64];
......@@ -235,7 +235,7 @@ inline bool HttpSession::checkLiveFlvStream(){
//开始发送rtmp负载
//关闭tcp_nodelay ,优化性能
SockUtil::setNoDelay(_sock->rawFD(),false);
(*this) << SocketFlags(sock_flags);
(*this) << SocketFlags(kSockFlags);
try{
start(mediaSrc);
......@@ -425,29 +425,10 @@ inline bool HttpSession::Handle_Req_GET(int64_t &content_len) {
//关闭tcp_nodelay ,优化性能
SockUtil::setNoDelay(_sock->rawFD(),false);
//设置MSG_MORE，优化性能
(*this) << SocketFlags(sock_flags);
//后台线程执行onFlush
auto onFlushWrapper = [onFlush,weakSelf](){
auto strongSelf = weakSelf.lock();
if(!strongSelf){
return false;
}
strongSelf->async([onFlush,weakSelf](){
//在后台线程完成文件读取，释放主线程性能
if(!onFlush()){
//如果onFlush返回false，则说明不再监听flush事件
auto strongSelf = weakSelf.lock();
if(strongSelf){
strongSelf->_sock->setOnFlush(nullptr);
}
}
});
return true;
};
(*this) << SocketFlags(kSockFlags);
onFlush();
_sock->setOnFlush(onFlushWrapper);
_sock->setOnFlush(onFlush);
return true;
}
......
src/Http/HttpSession.h
......@@ -126,34 +126,185 @@ private:
const string &contentOut);
};
/**
* 回显WebSocket会话
* 通过该模板类可以透明化WebSocket协议，
* 用户只要实现WebSock协议下的具体业务协议，譬如基于WebSocket协议的Rtmp协议等
* @tparam SessionType 业务协议的TcpSession类
*/
class EchoWebSocketSession : public HttpSession {
template <typename SessionType>
class WebSocketSession : public HttpSession {
public:
EchoWebSocketSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : HttpSession(pTh,pSock){};
virtual ~EchoWebSocketSession(){};
WebSocketSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : HttpSession(pTh,pSock){
_session = std::make_shared<SessionImp>(pTh,pSock);
}
virtual ~WebSocketSession(){};
//收到eof或其他导致脱离TcpServer事件的回调
void onError(const SockException &err) override{
HttpSession::onError(err);
_session->onError(err);
}
//每隔一段时间触发，用来做超时管理
void onManager() override{
HttpSession::onManager();
_session->onManager();
}
//在创建TcpSession后，TcpServer会把自身的配置参数通过该函数传递给TcpSession
void attachServer(const TcpServer &server) override{
HttpSession::attachServer(server);
_session->attachServer(server);
//此处截取数据并进行websocket协议打包
weak_ptr<WebSocketSession> weakSelf = dynamic_pointer_cast<WebSocketSession>(shared_from_this());
_session->setOnBeforeSendCB([weakSelf](const Buffer::Ptr &buf){
auto strongSelf = weakSelf.lock();
if(strongSelf){
bool mask_flag = strongSelf->_mask_flag;
strongSelf->_mask_flag = false;
strongSelf->WebSocketSplitter::encode((uint8_t *)buf->data(),buf->size());
strongSelf->_mask_flag = mask_flag;
}
return buf->size();
});
}
//作为该TcpSession的唯一标识符
string getIdentifier() const override{
return _session->getIdentifier();
}
protected:
/**
* 开始收到一个webSocket数据包
* @param packet
*/
void onWebSocketDecodeHeader(const WebSocketHeader &packet) override{
DebugL << packet._playload_len;
};
//新包，原来的包残余数据清空掉
_remian_data.clear();
}
/**
* 收到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 {
DebugL << string((char *)ptr,len) << " " << recved;
if(packet._playload_len == recved){
//收到完整的包
if(_remian_data.empty()){
onRecvWholePacket((char *)ptr,len);
}else{
_remian_data.append((char *)ptr,len);
onRecvWholePacket(_remian_data);
_remian_data.clear();
}
} else {
//部分数据
_remian_data.append((char *)ptr,len);
}
}
//webSocket服务器不允许对数据进行掩码加密
bool mask_flag = _mask_flag;
_mask_flag = false;
WebSocketSplitter::encode((uint8_t *)ptr,len);
_mask_flag = mask_flag;
/**
* 发送数据进行websocket协议打包后回调
* @param ptr
* @param len
*/
void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len) override{
_session->realSend(_session->obtainBuffer((char *)ptr,len));
}
};
/**
* 收到一个完整的websock数据包
* @param data
* @param len
*/
void onRecvWholePacket(const char *data,uint64_t len){
BufferRaw::Ptr buffer = _session->obtainBuffer(data,len);
_session->onRecv(buffer);
}
void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len) override{
send((char *)ptr,len);
/**
* 收到一个完整的websock数据包
* @param str
*/
void onRecvWholePacket(const string &str){
BufferString::Ptr buffer = std::make_shared<BufferString>(str);
_session->onRecv(buffer);
}
private:
typedef function<int(const Buffer::Ptr &buf)> onBeforeSendCB;
/**
* 该类实现了TcpSession派生类发送数据的截取
* 目的是发送业务数据前进行websocket协议的打包
*/
class SessionImp : public SessionType{
public:
SessionImp(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : SessionType(pTh,pSock){};
~SessionImp(){}
/**
* 截取到数据后，再进行webSocket协议打包
* 然后真正的发送数据到socket
* @param buf 数据
* @return 数据字节数
*/
int realSend(const Buffer::Ptr &buf){
return SessionType::send(buf);
}
/**
* 设置发送数据截取回调函数
* @param cb 截取回调函数
*/
void setOnBeforeSendCB(const onBeforeSendCB &cb){
_beforeSendCB = cb;
}
protected:
/**
* 重载send函数截取数据
* @param buf 需要截取的数据
* @return 数据字节数
*/
int send(const Buffer::Ptr &buf) override {
if(_beforeSendCB){
return _beforeSendCB(buf);
}
return SessionType::send(buf);
}
private:
onBeforeSendCB _beforeSendCB;
};
private:
std::shared_ptr<SessionImp> _session;
string _remian_data;
};
/**
* 回显会话
*/
class EchoSession : public TcpSession {
public:
EchoSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) : TcpSession(pTh,pSock){};
virtual ~EchoSession(){};
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> EchoWebSocketSession;
} /* namespace Http */
} /* namespace ZL */
......
src/Http/HttpsSession.h
......@@ -44,7 +44,7 @@ public:
#if defined(__GNUC__) && (__GNUC__ < 5)
public_send(data,len);
#else//defined(__GNUC__) && (__GNUC__ < 5)
HttpSession::send(data,len);
HttpSession::send(obtainBuffer(data,len));
#endif//defined(__GNUC__) && (__GNUC__ < 5)
});
m_sslBox.setOnDecData([&](const char *data, uint32_t len){
......@@ -64,27 +64,17 @@ public:
}
#if defined(__GNUC__) && (__GNUC__ < 5)
int public_send(const char *data, uint32_t len){
return HttpSession::send(data,len);
return HttpSession::send(obtainBuffer(data,len));
}
void public_onRecv(const char *data, uint32_t len){
HttpSession::onRecv(data,len);
}
#endif//defined(__GNUC__) && (__GNUC__ < 5)
private:
virtual int send(const string &buf) override{
virtual int send(const Buffer::Ptr &buf) override{
TimeTicker();
m_sslBox.onSend(buf.data(), buf.size());
return buf.size();
}
virtual int send(string &&buf) override{
TimeTicker();
m_sslBox.onSend(buf.data(), buf.size());
return buf.size();
}
virtual int send(const char *buf, int size) override{
TimeTicker();
m_sslBox.onSend(buf, size);
return size;
m_sslBox.onSend(buf->data(), buf->size());
return buf->size();
}
SSL_Box m_sslBox;
};
......
src/Rtmp/RtmpPlayer.h
......@@ -105,11 +105,8 @@ private:
//fro RtmpProtocol
void onRtmpChunk(RtmpPacket &chunkData) override;
void onStreamDry(uint32_t ui32StreamId) override;
void onSendRawData(const char *pcRawData, int iSize) override {
send(pcRawData, iSize);
}
void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
_sock->send(buffer,flags);
void onSendRawData(const Buffer::Ptr &buffer) override{
send(buffer);
}
template<typename FUN>
......
src/Rtmp/RtmpProtocol.cpp
......@@ -190,7 +190,7 @@ void RtmpProtocol::sendRequest(int iCmd, const string& str) {
}
void RtmpProtocol::sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId,
const std::string& strBuf, uint32_t ui32TimeStamp, int iChunkId , bool msg_more) {
const std::string& strBuf, uint32_t ui32TimeStamp, int iChunkId) {
if (iChunkId < 2 || iChunkId > 63) {
auto strErr = StrPrinter << "不支持发送该类型的块流 ID:" << iChunkId << endl;
throw std::runtime_error(strErr);
......@@ -235,7 +235,7 @@ void RtmpProtocol::sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId,
pos += chunk;
}
buffer->setSize(totalSize);
onSendRawData(buffer,msg_more ? SOCKET_DEFAULE_FLAGS : (SOCKET_DEFAULE_FLAGS | FLAG_MORE));
onSendRawData(buffer);
m_ui32ByteSent += totalSize;
if (m_ui32WinSize > 0 && m_ui32ByteSent - m_ui32LastSent >= m_ui32WinSize) {
m_ui32LastSent = m_ui32ByteSent;
......@@ -253,9 +253,9 @@ void RtmpProtocol::onParseRtmp(const char *pcRawData, int iSize) {
void RtmpProtocol::startClientSession(const function<void()> &callBack) {
//发送 C0C1
char handshake_head = HANDSHAKE_PLAINTEXT;
onSendRawData(&handshake_head, 1);
onSendRawData(obtainBuffer(&handshake_head, 1));
RtmpHandshake c1(0);
onSendRawData((char *) (&c1), sizeof(c1));
onSendRawData(obtainBuffer((char *) (&c1), sizeof(c1)));
m_nextHandle = [this,callBack]() {
//等待 S0+S1+S2
handle_S0S1S2(callBack);
......@@ -271,7 +271,7 @@ void RtmpProtocol::handle_S0S1S2(const function<void()> &callBack) {
}
//发送 C2
const char *pcC2 = m_strRcvBuf.data() + 1;
onSendRawData(pcC2, C1_HANDSHARK_SIZE);
onSendRawData(obtainBuffer(pcC2, C1_HANDSHARK_SIZE));
m_strRcvBuf.erase(0, 1 + 2 * C1_HANDSHARK_SIZE);
//握手结束
m_nextHandle = [this]() {
......@@ -306,12 +306,12 @@ void RtmpProtocol::handle_C0C1() {
void RtmpProtocol::handle_C1_simple(){
//发送S0
char handshake_head = HANDSHAKE_PLAINTEXT;
onSendRawData(&handshake_head, 1);
onSendRawData(obtainBuffer(&handshake_head, 1));
//发送S1
RtmpHandshake s1(0);
onSendRawData((char *) &s1, C1_HANDSHARK_SIZE);
onSendRawData(obtainBuffer((char *) &s1, C1_HANDSHARK_SIZE));
//发送S2
onSendRawData(m_strRcvBuf.c_str() + 1, C1_HANDSHARK_SIZE);
onSendRawData(obtainBuffer(m_strRcvBuf.c_str() + 1, C1_HANDSHARK_SIZE));
//等待C2
m_nextHandle = [this]() {
handle_C2();
......@@ -433,7 +433,7 @@ void RtmpProtocol::send_complex_S0S1S2(int schemeType,const string &digest){
//S1S2计算参考自:https://github.com/hitYangfei/golang/blob/master/rtmpserver.go
//发送S0
char handshake_head = HANDSHAKE_PLAINTEXT;
onSendRawData(&handshake_head, 1);
onSendRawData(obtainBuffer(&handshake_head, 1));
//S1
RtmpHandshake s1(0);
memcpy(s1.zero,"\x04\x05\x00\x01",4);
......@@ -460,7 +460,7 @@ void RtmpProtocol::send_complex_S0S1S2(int schemeType,const string &digest){
s1_joined.erase(digestPos - s1_start,C1_DIGEST_SIZE);
string s1_digest = openssl_HMACsha256(FMSKey,S1_FMS_KEY_SIZE,s1_joined.data(),s1_joined.size());
memcpy(digestPos,s1_digest.data(),s1_digest.size());
onSendRawData((char *) &s1, sizeof(s1));
onSendRawData(obtainBuffer((char *) &s1, sizeof(s1)));
//S2
string s2_key = openssl_HMACsha256(FMSKey,S2_FMS_KEY_SIZE,digest.data(),digest.size());
......@@ -468,7 +468,7 @@ void RtmpProtocol::send_complex_S0S1S2(int schemeType,const string &digest){
s2.random_generate((char *)&s2,8);
string s2_digest = openssl_HMACsha256(s2_key.data(),s2_key.size(),&s2,sizeof(s2) - C1_DIGEST_SIZE);
memcpy((char *)&s2 + C1_HANDSHARK_SIZE - C1_DIGEST_SIZE,s2_digest.data(),C1_DIGEST_SIZE);
onSendRawData((char *)&s2, sizeof(s2));
onSendRawData(obtainBuffer((char *)&s2, sizeof(s2)));
//等待C2
m_nextHandle = [this]() {
handle_C2();
......@@ -691,5 +691,11 @@ BufferRaw::Ptr RtmpProtocol::obtainBuffer() {
return std::make_shared<BufferRaw>() ;//_bufferPool.obtain();
}
BufferRaw::Ptr RtmpProtocol::obtainBuffer(const void *data, int len) {
auto buffer = obtainBuffer();
buffer->assign((const char *)data,len);
return buffer;
}
} /* namespace Rtmp */
} /* namespace ZL */
src/Rtmp/RtmpProtocol.h
......@@ -54,11 +54,8 @@ public:
void onParseRtmp(const char *pcRawData,int iSize);
void reset();
protected:
virtual void onSendRawData(const char *pcRawData,int iSize) = 0;
virtual void onSendRawData(const Buffer::Ptr &buffer,int flags) = 0;
virtual void onSendRawData(const Buffer::Ptr &buffer) = 0;
virtual void onRtmpChunk(RtmpPacket &chunkData) = 0;
virtual void onStreamBegin(uint32_t ui32StreamId){
m_ui32StreamId = ui32StreamId;
}
......@@ -78,19 +75,19 @@ protected:
void sendInvoke(const string &strCmd, const AMFValue &val);
void sendRequest(int iCmd, const string &str);
void sendResponse(int iType, const string &str);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const std::string &strBuf, uint32_t ui32TimeStamp, int iChunkID,bool msg_more = false);
void sendRtmp(uint8_t ui8Type, uint32_t ui32StreamId, const std::string &strBuf, uint32_t ui32TimeStamp, int iChunkID);
protected:
int m_iReqID = 0;
uint32_t m_ui32StreamId = STREAM_CONTROL;
int m_iNowStreamID = 0;
int m_iNowChunkID = 0;
bool m_bDataStarted = false;
BufferRaw::Ptr obtainBuffer();
//ResourcePool<BufferRaw,MAX_SEND_PKT> m_bufferPool;
inline BufferRaw::Ptr obtainBuffer();
inline BufferRaw::Ptr obtainBuffer(const void *data, int len);
//ResourcePool<BufferRaw,MAX_SEND_PKT> m_bufferPool;
private:
void handle_S0S1S2(const function<void()> &cb);
void handle_C0C1();
void handle_C1_simple();
#ifdef ENABLE_OPENSSL
void handle_C1_complex();
......@@ -104,6 +101,7 @@ private:
void handle_rtmp();
void handle_rtmpChunk(RtmpPacket &chunkData);
private:
////////////ChunkSize////////////
size_t m_iChunkLenIn = DEFAULT_CHUNK_LEN;
size_t m_iChunkLenOut = DEFAULT_CHUNK_LEN;
......
src/Rtmp/RtmpPusher.cpp
......@@ -35,6 +35,8 @@ using namespace ZL::Util;
namespace ZL {
namespace Rtmp {
static int kSockFlags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
unordered_map<string, RtmpPusher::rtmpCMDHandle> RtmpPusher::g_mapCmd;
RtmpPusher::RtmpPusher(const char *strVhost,const char *strApp,const char *strStream) {
auto src = dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA,strVhost,strApp,strStream));
......@@ -200,7 +202,7 @@ inline void RtmpPusher::send_metaData(){
sendRequest(MSG_DATA, enc.data());
src->getConfigFrame([&](const RtmpPacket::Ptr &pkt){
sendRtmp(pkt->typeId, m_ui32StreamId, pkt->strBuf, pkt->timeStamp, pkt->chunkId , true);
sendRtmp(pkt->typeId, m_ui32StreamId, pkt->strBuf, pkt->timeStamp, pkt->chunkId );
});
m_pRtmpReader = src->getRing()->attach();
......@@ -210,7 +212,7 @@ inline void RtmpPusher::send_metaData(){
if(!strongSelf) {
return;
}
strongSelf->sendRtmp(pkt->typeId, strongSelf->m_ui32StreamId, pkt->strBuf, pkt->timeStamp, pkt->chunkId , true);
strongSelf->sendRtmp(pkt->typeId, strongSelf->m_ui32StreamId, pkt->strBuf, pkt->timeStamp, pkt->chunkId);
});
m_pRtmpReader->setDetachCB([weakSelf](){
auto strongSelf = weakSelf.lock();
......@@ -220,6 +222,9 @@ inline void RtmpPusher::send_metaData(){
}
});
onPublishResult(SockException(Err_success,"success"));
//提高发送性能
(*this) << SocketFlags(kSockFlags);
SockUtil::setNoDelay(_sock->rawFD(),false);
}
void RtmpPusher::onCmd_result(AMFDecoder &dec){
auto iReqId = dec.load<int>();
......
src/Rtmp/RtmpPusher.h
......@@ -54,19 +54,15 @@ public:
}
protected:
//for Tcpclient
//for Tcpclient override
void onRecv(const Buffer::Ptr &pBuf) override;
void onConnect(const SockException &err) override;
void onErr(const SockException &ex) override;
//fro RtmpProtocol
//for RtmpProtocol override
void onRtmpChunk(RtmpPacket &chunkData) override;
void onSendRawData(const char *pcRawData, int iSize) override {
send(pcRawData, iSize);
}
void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
_sock->send(buffer,flags);
void onSendRawData(const Buffer::Ptr &buffer) override{
send(buffer);
}
private:
void init(const RtmpMediaSource::Ptr &src);
......
src/Rtmp/RtmpSession.cpp
......@@ -32,6 +32,8 @@
namespace ZL {
namespace Rtmp {
static int kSockFlags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
RtmpSession::RtmpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) :
TcpSession(pTh, pSock) {
DebugL << get_peer_ip();
......@@ -366,6 +368,10 @@ void RtmpSession::doPlayResponse(const string &err,bool tryDelay,const std::shar
if (src->getRing()->readerCount() == 1) {
src->seekTo(0);
}
//提高发送性能
(*this) << SocketFlags(kSockFlags);
SockUtil::setNoDelay(_sock->rawFD(),false);
}
void RtmpSession::doPlay(AMFDecoder &dec){
......@@ -539,7 +545,7 @@ void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) {
CLEAR_ARR(m_aui32FirstStamp);
modifiedStamp = 0;
}
sendRtmp(pkt->typeId, pkt->streamId, pkt->strBuf, modifiedStamp, pkt->chunkId , true);
sendRtmp(pkt->typeId, pkt->streamId, pkt->strBuf, modifiedStamp, pkt->chunkId);
}
void RtmpSession::doDelay(int delaySec, const std::function<void()> &fun) {
......
src/Rtmp/RtmpSession.h
......@@ -69,13 +69,9 @@ private:
void setMetaData(AMFDecoder &dec);
void onSendMedia(const RtmpPacket::Ptr &pkt);
void onSendRawData(const char *pcRawData,int iSize) override{
m_ui64TotalBytes += iSize;
send(pcRawData, iSize);
}
void onSendRawData(const Buffer::Ptr &buffer,int flags) override{
void onSendRawData(const Buffer::Ptr &buffer) override{
m_ui64TotalBytes += buffer->size();
_sock->send(buffer,flags);
send(buffer);
}
void onRtmpChunk(RtmpPacket &chunkData) override;
......
src/Rtsp/RtpBroadCaster.cpp
......@@ -130,7 +130,7 @@ RtpBroadCaster::RtpBroadCaster(const string &strLocalIp,const string &strVhost,c
auto &pSock = m_apUdpSock[i];
auto &peerAddr = m_aPeerUdpAddr[i];
BufferRtp::Ptr buffer(new BufferRtp(pkt,4));
pSock->send(buffer,SOCKET_DEFAULE_FLAGS,(struct sockaddr *)(&peerAddr));
pSock->send(buffer,SOCKET_DEFAULE_FLAGS | FLAG_MORE,(struct sockaddr *)(&peerAddr));
});
m_pReader->setDetachCB([this](){
unordered_map<void * , onDetach > m_mapDetach_copy;
......
src/Rtsp/RtspSession.cpp
......@@ -43,6 +43,8 @@ using namespace ZL::Network;
namespace ZL {
namespace Rtsp {
static int kSockFlags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
string dateHeader() {
char buf[200];
time_t tt = time(NULL);
......@@ -210,7 +212,7 @@ bool RtspSession::handleReq_Options() {
m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
return true;
}
......@@ -277,7 +279,7 @@ void RtspSession::onAuthSuccess(const weak_ptr<RtspSession> &weakSelf) {
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), strongSelf->m_strUrl.data(),
(int) strongSelf->m_strSdp.length(), strongSelf->m_strSdp.data());
strongSelf->send(response, n);
strongSelf->SocketHelper::send(response, n);
});
}
void RtspSession::onAuthFailed(const weak_ptr<RtspSession> &weakSelf,const string &realm) {
......@@ -320,7 +322,7 @@ void RtspSession::onAuthFailed(const weak_ptr<RtspSession> &weakSelf,const strin
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), realm.data());
}
strongSelf->send(response, n);
strongSelf->SocketHelper::send(response, n);
});
}
......@@ -471,7 +473,7 @@ inline void RtspSession::send_StreamNotFound() {
m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
}
inline void RtspSession::send_UnsupportedTransport() {
int n = sprintf(m_pcBuf, "RTSP/1.0 461 Unsupported Transport\r\n"
......@@ -482,7 +484,7 @@ inline void RtspSession::send_UnsupportedTransport() {
m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
}
inline void RtspSession::send_SessionNotFound() {
......@@ -494,7 +496,7 @@ inline void RtspSession::send_SessionNotFound() {
m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
/*40 Method Not Allowed*/
......@@ -562,7 +564,7 @@ bool RtspSession::handleReq_Setup() {
trackRef.type * 2 + 1,
printSSRC(trackRef.ssrc).data(),
m_strSession.data());
send(m_pcBuf, iLen);
SocketHelper::send(m_pcBuf, iLen);
}
break;
case PlayerBase::RTP_UDP: {
......@@ -607,7 +609,7 @@ bool RtspSession::handleReq_Setup() {
pSockRtp->get_local_port(), pSockRtcp->get_local_port(),
printSSRC(trackRef.ssrc).data(),
m_strSession.data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
}
break;
case PlayerBase::RTP_MULTICAST: {
......@@ -650,7 +652,7 @@ bool RtspSession::handleReq_Setup() {
get_local_ip().data(), iSrvPort, pSockRtcp->get_local_port(),
udpTTL,printSSRC(trackRef.ssrc).data(),
m_strSession.data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
}
break;
default:
......@@ -685,7 +687,7 @@ bool RtspSession::handleReq_Play() {
m_iCseq, SERVER_NAME,
RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(),(int)err.size(),err.data());
send(m_pcBuf,n);
SocketHelper::send(m_pcBuf,n);
shutdown();
return;
}
......@@ -757,7 +759,11 @@ bool RtspSession::handleReq_Play() {
iLen -= 1;
(m_pcBuf)[iLen] = '\0';
iLen += sprintf(m_pcBuf + iLen, "\r\n\r\n");
send(m_pcBuf, iLen);
SocketHelper::send(m_pcBuf, iLen);
//提高发送性能
(*this) << SocketFlags(kSockFlags);
SockUtil::setNoDelay(m_pSender->rawFD(),false);
};
weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
......@@ -793,7 +799,7 @@ bool RtspSession::handleReq_Pause() {
"%s"
"Session: %s\r\n\r\n", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), m_strSession.data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
if(m_pRtpReader){
m_pRtpReader->setReadCB(nullptr);
}
......@@ -809,7 +815,7 @@ bool RtspSession::handleReq_Teardown() {
"Session: %s\r\n\r\n", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), m_strSession.data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
TraceL << "播放器断开连接!";
return false;
}
......@@ -827,7 +833,7 @@ bool RtspSession::handleReq_Get() {
lock_guard<recursive_mutex> lock(g_mtxGetter);
g_mapGetter[m_strSessionCookie] = dynamic_pointer_cast<RtspSession>(shared_from_this());
//InfoL << m_strSessionCookie;
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
return true;
}
......@@ -862,7 +868,7 @@ bool RtspSession::handleReq_SET_PARAMETER() {
"%s"
"Session: %s\r\n\r\n", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data(), m_strSession.data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
return true;
}
......@@ -873,7 +879,7 @@ inline void RtspSession::send_NotAcceptable() {
"%s"
"Connection: Close\r\n\r\n", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME,
dateHeader().data());
send(m_pcBuf, n);
SocketHelper::send(m_pcBuf, n);
}
......@@ -939,7 +945,7 @@ inline void RtspSession::sendRtpPacket(const RtpPacket::Ptr & pkt) {
}
BufferRtp::Ptr buffer(new BufferRtp(pkt,4));
m_ui64TotalBytes += buffer->size();
pSock->send(buffer,SOCKET_DEFAULE_FLAGS, peerAddr.get());
pSock->send(buffer,kSockFlags, peerAddr.get());
}
break;
default:
......
src/Rtsp/RtspSession.h
......@@ -89,21 +89,9 @@ protected:
int64_t onRecvHeader(const char *data,uint64_t len) override ;
private:
void inputRtspOrRtcp(const char *data,uint64_t len);
int send(const string &strBuf) override {
m_ui64TotalBytes += strBuf.size();
return m_pSender->send(strBuf);
}
int send(string &&strBuf) override {
m_ui64TotalBytes += strBuf.size();
return m_pSender->send(std::move(strBuf));
}
int send(const char *pcBuf, int iSize) override {
m_ui64TotalBytes += iSize;
return m_pSender->send(pcBuf, iSize);
}
int send(const Buffer::Ptr &pkt) override{
m_ui64TotalBytes += pkt->size();
return m_pSender->send(pkt,SOCKET_DEFAULE_FLAGS | FLAG_MORE);
return m_pSender->send(pkt,_flags);
}
void shutdown() override ;
void shutdown_l(bool close);
......
tests/test_httpClient.cpp
......@@ -53,8 +53,11 @@ int main(int argc, char *argv[]) {
//下载器map
map<string, HttpDownloader::Ptr> downloaderMap;
//下载两个文件，一个是http下载，一个https下载
auto urlList = {"http://img3.imgtn.bdimg.com/it/u=158031390,1321729164&fm=214&gp=0.jpg",
"https://ss0.bdstatic.com/70cFvHSh_Q1YnxGkpoWK1HF6hhy/it/u=931786003,1029770543&fm=27&gp=0.jpg"};
auto urlList = {"https://timgsa.baidu.com/timg?image&quality=80&"
"size=b9999_10000&sec=1537717640404&"
"di=f602efbebbc1e7f6b9ccb0bf0def89d0&"
"imgtype=0&"
"src=http%3A%2F%2Fimgsrc.baidu.com%2Fimgad%2Fpic%2Fitem%2F241f95cad1c8a786ff65052a6d09c93d70cf5042.jpg",};
for (auto &url : urlList) {
//创建下载器
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论