From df4a75ea7bf42555c4ffb3aee0fa8a16298a9efd Mon Sep 17 00:00:00 2001 From: holmesyxli Date: Tue, 2 Nov 2021 23:30:54 +0800 Subject: [PATCH] delete channel in WFWebSocketClient::deinit() --- src/client/WFWebSocketClient.cc | 81 +++++++++++++++++++------------- src/client/WFWebSocketClient.h | 6 +-- src/factory/WFChannel.inl | 8 ++-- src/factory/WebSocketTaskImpl.cc | 20 ++------ 4 files changed, 59 insertions(+), 56 deletions(-) diff --git a/src/client/WFWebSocketClient.cc b/src/client/WFWebSocketClient.cc index 526363d8..53b747bb 100644 --- a/src/client/WFWebSocketClient.cc +++ b/src/client/WFWebSocketClient.cc @@ -16,50 +16,80 @@ Author: Li Yingxin (liyingxin@sogou-inc.com) */ -#include "WFChannel.h" #include "HttpMessage.h" +#include "WFFacilities.h" +#include "WFChannel.h" #include "WebSocketMessage.h" #include "WFWebSocketClient.h" +struct websocket_series_ctx_t +{ + WebSocketClient *client; + WFFacilities::WaitGroup *wait_group; +}; + WFWebSocketTask *WebSocketClient::create_websocket_task(websocket_callback_t cb) { - return new ComplexWebSocketOutTask(&this->channel, WFGlobal::get_scheduler(), + return new ComplexWebSocketOutTask(this->channel, + WFGlobal::get_scheduler(), std::move(cb)); } int WebSocketClient::init(const std::string& url) { ParsedURI uri; + if (URIParser::parse(url, uri) != 0) + return -1; - if (pthread_cond_init(&this->shutdown_cond, NULL) == 0) + this->channel = new ComplexWebSocketChannel(NULL, + WFGlobal::get_scheduler(), + std::move(process)); + this->channel->set_idle_timeout(this->params.idle_timeout); + this->channel->set_size_limit(this->params.size_limit); + this->channel->set_uri(uri); + + this->channel->set_callback([this](WFChannel *channel) { - if (URIParser::parse(url, uri) == 0) + pthread_mutex_lock(&this->channel->mutex); + if (this->channel->is_established() == 0) { - this->channel.set_uri(uri); - return 0; + this->channel->set_state(WFT_STATE_SYS_ERROR); + this->channel->set_sending(false); } + pthread_mutex_unlock(&this->channel->mutex); + }); - pthread_cond_destroy(&this->shutdown_cond); - } - - return -1; + return 0; } void WebSocketClient::deinit() { - pthread_mutex_lock(&this->channel.mutex); + WFFacilities::WaitGroup wait_group(1); - while (this->channel.get_sending()) - pthread_cond_wait(&this->shutdown_cond, &this->channel.mutex); + struct websocket_series_ctx_t ctx = { + .client = this, + .wait_group = &wait_group, + }; - pthread_mutex_unlock(&this->channel.mutex); - pthread_cond_destroy(&this->shutdown_cond); + SeriesWork *series = Workflow::create_series_work(this->channel, + [](const SeriesWork *series) + { + struct websocket_series_ctx_t *ctx; + ctx = (struct websocket_series_ctx_t *)series->get_context(); + delete ctx->client->channel; + ctx->wait_group->done(); + }); + + series->set_context(&ctx); + series->start(); + + wait_group.wait(); } WFWebSocketTask *WebSocketClient::create_ping_task(websocket_callback_t cb) { ComplexWebSocketOutTask *ping_task; - ping_task = new ComplexWebSocketOutTask(&this->channel, + ping_task = new ComplexWebSocketOutTask(this->channel, WFGlobal::get_scheduler(), std::move(cb)); @@ -72,7 +102,7 @@ WFWebSocketTask *WebSocketClient::create_ping_task(websocket_callback_t cb) WFWebSocketTask *WebSocketClient::create_close_task(websocket_callback_t cb) { ComplexWebSocketOutTask *close_task; - close_task = new ComplexWebSocketOutTask(&this->channel, + close_task = new ComplexWebSocketOutTask(this->channel, WFGlobal::get_scheduler(), std::move(cb)); @@ -84,23 +114,8 @@ WFWebSocketTask *WebSocketClient::create_close_task(websocket_callback_t cb) WebSocketClient::WebSocketClient(const struct WFWebSocketParams *params, websocket_process_t process) : - channel(NULL, WFGlobal::get_scheduler(), std::move(process)) + process(std::move(process)) { this->params = *params; - this->channel.set_idle_timeout(this->params.idle_timeout); - this->channel.set_size_limit(this->params.size_limit); - - this->channel.set_callback([this](WFChannel *channel) - { - pthread_mutex_lock(&this->channel.mutex); - - if (this->channel.is_established() == 0) - this->channel.set_sending(false); - - this->channel.condition.signal(NULL); - pthread_cond_signal(&this->shutdown_cond); - - pthread_mutex_unlock(&this->channel.mutex); - }); } diff --git a/src/client/WFWebSocketClient.h b/src/client/WFWebSocketClient.h index 3d29c32f..e0413572 100644 --- a/src/client/WFWebSocketClient.h +++ b/src/client/WFWebSocketClient.h @@ -20,9 +20,9 @@ #define _WFWEBSOCKETCLIENT_H_ #include -#include #include #include "HttpUtil.h" +#include "HttpMessage.h" #include "WFChannel.h" #include "WebSocketMessage.h" @@ -52,9 +52,9 @@ public: void deinit(); private: - ComplexWebSocketChannel channel; + ComplexWebSocketChannel *channel; struct WFWebSocketParams params; - pthread_cond_t shutdown_cond; + websocket_process_t process; public: WebSocketClient(const struct WFWebSocketParams *params, diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index 569445af..eefdc05d 100644 --- a/src/factory/WFChannel.inl +++ b/src/factory/WFChannel.inl @@ -226,11 +226,11 @@ SubTask *WFComplexChannel::done() if (this->callback) this->callback(this); - if (this->state == WFT_STATE_SUCCESS || // after upgrade - this->established == 0) // after close - { + if (this->state == WFT_STATE_SUCCESS) this->state = WFT_STATE_UNDEFINED; - } + +// if (this->established == 0 && this->state == WFT_STATE_SYS_ERROR) // sending == false +// delete this; return series->pop(); } diff --git a/src/factory/WebSocketTaskImpl.cc b/src/factory/WebSocketTaskImpl.cc index 00087e77..72b4f231 100644 --- a/src/factory/WebSocketTaskImpl.cc +++ b/src/factory/WebSocketTaskImpl.cc @@ -102,7 +102,6 @@ SubTask *ComplexWebSocketInTask::done() SubTask *ComplexWebSocketOutTask::done() { - SeriesWork *series = series_of(this); auto *channel = (ComplexWebSocketChannel *)this->get_request_channel(); if (channel->get_state() == WFT_STATE_UNDEFINED || @@ -117,21 +116,10 @@ SubTask *ComplexWebSocketOutTask::done() this->error = channel->get_error(); } - const websocket_parser_t *parser = this->get_msg()->get_parser(); - - if (parser->opcode == WebSocketFrameConnectionClose && - this->get_state() == WFT_STATE_SUCCESS && - channel->is_established()) - { - series->push_front(channel); - } - else - { - pthread_mutex_lock(&channel->mutex); - channel->set_sending(false); - channel->condition.signal(NULL); - pthread_mutex_unlock(&channel->mutex); - } + pthread_mutex_lock(&channel->mutex); + channel->set_sending(false); + channel->condition.signal(NULL); + pthread_mutex_unlock(&channel->mutex); return WFChannelOutTask::done(); }