diff --git a/src/client/WFWebSocketClient.cc b/src/client/WFWebSocketClient.cc index 96a93342..dc8209d2 100644 --- a/src/client/WFWebSocketClient.cc +++ b/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 *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(); } diff --git a/src/factory/WFChannel.inl b/src/factory/WFChannel.inl index b1b16a9b..3fb929a4 100644 --- a/src/factory/WFChannel.inl +++ b/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::done() return series->pop(); } + // dispacth() by handle_terminate() or deinit() + if (this->established == 0 && + (WFComplexChannel *)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 void WFComplexChannel::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::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 +void WFComplexChannel::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::wait_callback, + this, std::placeholders::_1)); + } + pthread_mutex_unlock(&this->mutex); + + series_of(task)->push_back(next); } template diff --git a/tutorial/tutorial-14-websocket_cli.cc b/tutorial/tutorial-14-websocket_cli.cc index cabb8421..9e277358 100644 --- a/tutorial/tutorial-14-websocket_cli.cc +++ b/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);