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

websocket客户端触发事件时确保对象强引用有效

Browse Source 
同时修复websocket客户端判断alive返回false的bug
This commit is contained in:
ziyue 2022-01-25 11:44:51 +08:00
parent 4f27894302
commit 756ec1364b
2 changed files with 27 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
3rdpart/ZLToolKit

@ -1 +1 @@
Subproject commit 6e5f08d89cd6fd06a4185f5de937ef629c24a83e Subproject commit 25062620233c62475aaffc0a9960e2689d8418ce

38
src/Http/WebSocketClient.h
View File

@ -72,9 +72,10 @@ class HttpWsClient : public HttpClientImp , public WebSocketSplitter{
public: public:
typedef shared_ptr<HttpWsClient> Ptr; typedef shared_ptr<HttpWsClient> Ptr;
HttpWsClient(ClientTypeImp<ClientType,DataType> &delegate) : _delegate(delegate){ HttpWsClient(const std::shared_ptr<ClientTypeImp<ClientType, DataType> > &delegate) : _weak_delegate(delegate),
_delegate(*delegate) {
_Sec_WebSocket_Key = encodeBase64(makeRandStr(16, false)); _Sec_WebSocket_Key = encodeBase64(makeRandStr(16, false));
setPoller(delegate.getPoller()); setPoller(_delegate.getPoller());
} }
~HttpWsClient(){} ~HttpWsClient(){}
@ -153,14 +154,20 @@ protected:
//TcpClient override //TcpClient override
void onRecv(const Buffer::Ptr &buf) override {
auto strong_ref = _weak_delegate.lock();;
HttpClientImp::onRecv(buf);
}
/** /**
* 定时触发 * 定时触发
*/ */
void onManager() override { void onManager() override {
if(_onRecv){ auto strong_ref = _weak_delegate.lock();;
if (_onRecv) {
//websocket连接成功了 //websocket连接成功了
_delegate.onManager(); _delegate.onManager();
} else{ } else {
//websocket连接中... //websocket连接中...
HttpClientImp::onManager(); HttpClientImp::onManager();
} }
@ -169,11 +176,12 @@ protected:
/** /**
* 数据全部发送完毕后回调 * 数据全部发送完毕后回调
*/ */
void onFlush() override{ void onFlush() override {
if(_onRecv){ auto strong_ref = _weak_delegate.lock();;
if (_onRecv) {
//websocket连接成功了 //websocket连接成功了
_delegate.onFlush(); _delegate.onFlush();
} else{ } else {
//websocket连接中... //websocket连接中...
HttpClientImp::onFlush(); HttpClientImp::onFlush();
} }
@ -182,8 +190,9 @@ protected:
/** /**
* tcp连接结果 * tcp连接结果
*/ */
void onConnect(const SockException &ex) override{ void onConnect(const SockException &ex) override {
if(ex){ auto strong_ref = _weak_delegate.lock();;
if (ex) {
//tcp连接失败，直接返回失败 //tcp连接失败，直接返回失败
onWebSocketException(ex); onWebSocketException(ex);
return; return;
@ -195,7 +204,8 @@ protected:
/** /**
* tcp连接断开 * tcp连接断开
*/ */
void onErr(const SockException &ex) override{ void onErr(const SockException &ex) override {
auto strong_ref = _weak_delegate.lock();;
//tcp断开或者shutdown导致的断开 //tcp断开或者shutdown导致的断开
onWebSocketException(ex); onWebSocketException(ex);
} }
@ -335,6 +345,7 @@ private:
private: private:
string _Sec_WebSocket_Key; string _Sec_WebSocket_Key;
function<void(const char *data, size_t len)> _onRecv; function<void(const char *data, size_t len)> _onRecv;
weak_ptr<ClientTypeImp<ClientType,DataType> > _weak_delegate;
ClientTypeImp<ClientType,DataType> &_delegate; ClientTypeImp<ClientType,DataType> &_delegate;
string _payload_section; string _payload_section;
string _payload_cache; string _payload_cache;
@ -354,7 +365,6 @@ public:
template<typename ...ArgsType> template<typename ...ArgsType>
WebSocketClient(ArgsType &&...args) : ClientTypeImp<ClientType,DataType>(std::forward<ArgsType>(args)...){ WebSocketClient(ArgsType &&...args) : ClientTypeImp<ClientType,DataType>(std::forward<ArgsType>(args)...){
_wsClient.reset(new HttpWsClient<ClientType,DataType>(*this));
} }
~WebSocketClient() override { ~WebSocketClient() override {
_wsClient->closeWsClient(); _wsClient->closeWsClient();
@ -377,10 +387,14 @@ public:
//明文ws //明文ws
ws_url = StrPrinter << "ws://" + host << ":" << port << "/"; ws_url = StrPrinter << "ws://" + host << ":" << port << "/";
} }
_wsClient->startWsClient(ws_url, timeout_sec); startWebSocket(ws_url, timeout_sec);
} }
void startWebSocket(const string &ws_url,float fTimeOutSec = 3){ void startWebSocket(const string &ws_url,float fTimeOutSec = 3){
_wsClient = std::make_shared<HttpWsClient<ClientType,DataType> >(static_pointer_cast<WebSocketClient>(this->shared_from_this()));
_wsClient->setOnCreateSocket([this](const EventPoller::Ptr &){
return this->createSocket();
});
_wsClient->startWsClient(ws_url,fTimeOutSec); _wsClient->startWsClient(ws_url,fTimeOutSec);
} }