websocket : update handle_terminate() and client.deinit() codes

src/client/WFWebSocketClient.cc

@@ -21,6 +21,32 @@
 #include "WebSocketMessage.h"
 #include "WFWebSocketClient.h"
 
+static void deinit_wait_callback(WFMailboxTask *task)
+{
+	ComplexWebSocketChannel *channel;
+	channel = (ComplexWebSocketChannel *)series_of(task)->get_context();
+	SubTask *next;
+
+	if (channel->is_established() == false)
+		return;
+
+	pthread_mutex_lock(&channel->mutex);
+	if (channel->get_sending() == false)
+	{
+		channel->set_sending(true);
+		channel->set_pointer(NULL);
+		next = channel;
+	}
+	else // if out_task started after deinit(), which is not recommanded
+	{
+		next = WFCondTaskFactory::create_wait_task(&channel->condition,
+												   deinit_wait_callback);
+	}
+	pthread_mutex_unlock(&channel->mutex);
+
+	series_of(task)->push_back(next);
+}
+
 WFWebSocketTask *WebSocketClient::create_websocket_task(websocket_callback_t cb)
 {
 	WFWebSocketTask *task = new ComplexWebSocketOutTask(this->channel,
@@ -69,9 +95,29 @@ int WebSocketClient::init(const struct WFWebSocketParams *params)
 void WebSocketClient::deinit()
 {
 	SeriesWork *series;
-	this->channel->set_pointer(NULL);
-	series = Workflow::create_series_work(this->channel,
-										  [](const SeriesWork *series){
+	SubTask *first;
+
+	if (this->channel->is_established() == false)
+	{
+		delete this->channel;
+		return;
+	}
+
+	pthread_mutex_lock(&this->channel->mutex);
+	if (this->channel->get_sending() == false)
+	{
+		this->channel->set_sending(true);
+		this->channel->set_pointer(NULL);
+		first = this->channel;
+	}
+	else
+	{
+		first = WFCondTaskFactory::create_wait_task(&this->channel->condition,
+													deinit_wait_callback);
+	}
+	pthread_mutex_unlock(&this->channel->mutex);
+
+	series = Workflow::create_series_work(first, [](const SeriesWork *series) {
 		ComplexWebSocketChannel *channel;
 		channel = (ComplexWebSocketChannel *)series->get_context();
 		delete channel;
@@ -115,10 +161,10 @@ void WebSocketClient::channel_callback(WFChannel<protocol::WebSocketFrame> *ch,
 	ComplexWebSocketChannel *channel = (ComplexWebSocketChannel *)ch;
 
 	pthread_mutex_lock(&channel->mutex);
-	if (channel->is_established() == 0)
+	if (channel->is_established() == false)
 	{
-		channel->set_state(WFT_STATE_SYS_ERROR);
-		channel->set_sending(false);
+//		channel->set_state(WFT_STATE_SYS_ERROR);
+//		channel->set_sending(false);
 		if (close != nullptr)
 			close();
 	}

src/factory/WFChannel.inl

@@ -165,6 +165,7 @@ protected:
 	virtual void handle_terminated();
 	virtual WFRouterTask *route();
 	void router_callback(WFRouterTask *task);
+	void wait_callback(WFMailboxTask *task);
 
 public:
 	pthread_mutex_t mutex;
@@ -229,41 +230,78 @@ SubTask *WFComplexChannel<MSG>::done()
 		return series->pop();
 	}
 
+	// dispacth() by handle_terminate() or deinit()
+	if (this->established == 0 &&
+		(WFComplexChannel<MSG> *)series->get_context() == this)
+	{
+		pthread_mutex_lock(&this->mutex);
+		this->sending = false;
+		this->condition.signal(NULL);
+		pthread_mutex_unlock(&this->mutex);
+	}
+
 	if (this->callback)
 		this->callback(this);
 
 	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();
 }
 
 template<class MSG>
 void WFComplexChannel<MSG>::handle_terminated()
 {
-	WFMailboxTask *waiter;
-	bool shutdown = false;
+	SeriesWork *series;
+	SubTask *first;
+
+	if (this->established == 0)
+		return;
 
 	pthread_mutex_lock(&this->mutex);
-	Workflow::create_series_work(this, nullptr);
-
 	if (this->sending == false)
 	{
 		this->sending = true;
-		shutdown = true;
-	} else {
-		waiter = WFCondTaskFactory::create_wait_task(&this->condition,
-													 nullptr);
-		series_of(this)->push_front(this);
-		series_of(this)->push_front(waiter);
+		this->set_pointer(NULL);
+		first = this;
+	}
+	else
+	{
+		first = WFCondTaskFactory::create_wait_task(&this->condition,
+					std::bind(&WFComplexChannel<MSG>::wait_callback,
+							  this, std::placeholders::_1));
 	}
 	pthread_mutex_unlock(&this->mutex);
 
-	if (shutdown == true)
-		this->dispatch();
+	series = Workflow::create_series_work(first, nullptr);
+	series->set_context(this);
+	series->start();
+}
+
+template<class MSG>
+void WFComplexChannel<MSG>::wait_callback(WFMailboxTask *task)
+{
+	SubTask *next;
+
+	if (this->established == 0)
+		return;
+
+	pthread_mutex_lock(&this->mutex);
+	if (this->sending == false)
+	{
+		this->sending = true;
+		this->set_pointer(NULL);
+		next = this;
+	}
+	else
+	{
+		next = WFCondTaskFactory::create_wait_task(&this->condition,
+							std::bind(&WFComplexChannel<MSG>::wait_callback,
+									  this, std::placeholders::_1));
+	}
+	pthread_mutex_unlock(&this->mutex);
+
+	series_of(task)->push_back(next);
 }
 
 template<class MSG>

tutorial/tutorial-14-websocket_cli.cc

@@ -44,6 +44,11 @@ void process(WFWebSocketTask *task)
 	}
 }
 
+void closed()
+{
+	fprintf(stderr, "connection is close by system\n");
+}
+
 int main(int argc, char *argv[])
 {
 	if (argc != 2)
@@ -54,7 +59,7 @@ int main(int argc, char *argv[])
 		return 0;
 	}
 
-	WebSocketClient client(process);
+	WebSocketClient client(process, closed);
 	client.init(argv[1]);
 
 	WFFacilities::WaitGroup wg(1);