delete channel in WFWebSocketClient::deinit()

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<protocol::WebSocketFrame> *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<protocol::WebSocketFrame> *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);
-	});
 }
 

src/client/WFWebSocketClient.h

@@ -20,9 +20,9 @@
 #define _WFWEBSOCKETCLIENT_H_
 
 #include <string>
-#include <pthread.h>
 #include <functional>
 #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,

src/factory/WFChannel.inl

@@ -226,11 +226,11 @@ SubTask *WFComplexChannel<MSG>::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();
 }

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<WebSocketFrame>::done();
 }