I am trying to implement server open http connection with events using Cpp Rest SDK. I just started to investigate CPP Rest. I used this example with implementation:
MyServer::MyServer(utility::string_t url) : mListener(url)
{
mListener.support(methods::GET, bind(&MyServer::handleGet, this, placeholders::_1));
}
void MyServer::handleGet(http_request iRequest)
{
ucout << iRequest.to_string() << endl;
http_response wResponse;
// Setting headers
wResponse.set_status_code(status_codes::OK);
wResponse.headers().add(header_names::access_control_allow_origin, U("*"));
wResponse.headers().add(header_names::content_type, U("text/event-stream"));
// Preparing buffer
streams::producer_consumer_buffer<char> wBuffer;
streams::basic_istream<uint8_t> wStream(wBuffer);
wResponse.set_body(wStream);
auto wReplyTask = iRequest.reply(wResponse);
wBuffer.putn_nocopy("data: a\n",10).wait();
wBuffer.putn_nocopy("data: b\n\n",12).wait();
wBuffer.sync().wait(); // seems equivalent to 'flush'
this_thread::sleep_for(chrono::milliseconds(2000));
wBuffer.putn_nocopy("data: c\n", 10).wait();
wBuffer.putn_nocopy("data: d\n\n", 12).wait();
wBuffer.sync().wait();
// wBuffer.close(std::ios_base::out).wait(); // closes the connection
wReplyTask.wait(); // blocking!
}
I running in debugger, and I never see thread is terminated. When client is aborted connection, still thread is sitting here, as I can see in the VS threads window in the debugger and no breakpoint after wait is happens. I am concerning to terminate thread handleGet, when client aborts connection. How can I do it?
With connection from multiple clients and knowledge about limit of 40 thread in SDK, I am not sure, how properly terminate thread created by the library.
Solved it by checking wReplyTask.is_done(). Removed wait, and processed response in the loop. When client aborted connection, is_done() returns true. Full function:
void MyServer::handleGet(http_request iRequest)
{
ucout << iRequest.to_string() << endl;
http_response wResponse;
auto context = iRequest._get_server_context();
// Setting headers
wResponse.set_status_code(status_codes::OK);
wResponse.headers().add(header_names::access_control_allow_origin, U("*"));
wResponse.headers().add(header_names::content_type, U("text/event-stream"));
// Preparing buffer
streams::producer_consumer_buffer<char> wBuffer;
streams::basic_istream<uint8_t> wStream(wBuffer);
wResponse.set_body(wStream);
auto wReplyTask = iRequest.reply(wResponse);// .then([this](pplx::task<void> t) { handle_error(t); });
wBuffer.putn_nocopy("id: 35\n", 7).wait();
wBuffer.putn_nocopy("data: a\n", 8).wait();
wBuffer.putn_nocopy("data: b\n\n", 9).wait();
wBuffer.sync().wait(); // seems equivalent to 'flush'
while (true)
{
this_thread::sleep_for(chrono::milliseconds(2000));
if (wReplyTask.is_done())
{
wBuffer.close(std::ios_base::out).wait();
return;
}
string datac = "data: c\n";
wBuffer.putn_nocopy(datac.c_str(), datac.size()).wait();
wBuffer.putn_nocopy("data: d\n\n", 9).wait();
wBuffer.sync().wait();
}
// wBuffer.close(std::ios_base::out).wait(); // closes the connection
//wReplyTask.wait(); // blocking!
}