Fixing a bug in http_client where we were waiting for the response stream to close in _complete and not handling any exceptions.

Author: stgates

Release/include/cpprest/astreambuf.h

@@ -422,7 +422,7 @@ namespace streams
             }
 
             // After the flush_internal task completed, "this" object may have been destroyed, 
-            // accessing the memebers is invalid, use shared_from_this to avoid access violation exception.
+            // accessing the members is invalid, use shared_from_this to avoid access violation exception.
             auto this_ptr = std::static_pointer_cast<streambuf_state_manager>(this->shared_from_this());
 
             if (mode & std::ios_base::out && can_write()) {

Release/src/http/common/http_msg.cpp

@@ -170,29 +170,73 @@ size_t http_msg_base::_get_content_length()
 /// </summary>
 void http_msg_base::_complete(utility::size64_t body_size, const std::exception_ptr &exceptionPtr)
 {
-    // Close the write head
-    if ((bool)outstream())
+    const auto hasBody = outstream().is_valid();
+    const auto hasException = exceptionPtr != std::exception_ptr();
+    const auto &completionEvent = _get_data_available();
+    auto closeTask = pplx::task_from_result();
+
+    if (hasBody)
     {
-        if ( !(exceptionPtr == std::exception_ptr()) )
-            outstream().close(exceptionPtr).get();
-        else if ( m_default_outstream )
-            outstream().close().get();
+        if (hasException)
+        {
+            closeTask = outstream().close(exceptionPtr);
+        }
+        else if (m_default_outstream)
+        {
+            closeTask = outstream().close();
+        }
     }
 
-    if(exceptionPtr == std::exception_ptr())
+    if (hasException)
     {
-        _get_data_available().set(body_size);
+        auto setException = [completionEvent, exceptionPtr](pplx::task<void> t)
+        {
+            // If closing stream throws an exception ignore since we already have an error.
+            try { t.get(); } catch (...) {}
+            completionEvent.set_exception(exceptionPtr);
+            pplx::create_task(completionEvent).then([](pplx::task<utility::size64_t> t)
+            {
+                try { t.get(); } catch (...) {}
+            });
+        };
+
+        if (closeTask.is_done())
+        {
+            setException(closeTask);
+        }
+        else
+        {
+            closeTask.then(setException);
+        }
     }
     else
     {
-        _get_data_available().set_exception(exceptionPtr);
-        // The exception for body will be observed by default, because read body is not always required.
-        pplx::create_task(_get_data_available()).then([](pplx::task<utility::size64_t> t) {
-            try {
+        auto setBodySize = [completionEvent, body_size](pplx::task<void> t)
+        {
+            try
+            {
                 t.get();
-            } catch (...) {
+                completionEvent.set(body_size);
             }
-        });
+            catch (...)
+            {
+                // If close throws an exception report back to user.
+                completionEvent.set_exception(std::current_exception());
+                pplx::create_task(completionEvent).then([](pplx::task<utility::size64_t> t)
+                {
+                    try { t.get(); } catch (...) {}
+                });
+            }
+        };
+
+        if (closeTask.is_done())
+        {
+            setBodySize(closeTask);
+        }
+        else
+        {
+            closeTask.then(setBodySize);
+        }
     }
 }
 

Release/tests/functional/http/client/response_stream_tests.cpp

@@ -251,6 +251,83 @@ TEST_FIXTURE(uri_address, response_stream_large_file_stream)
 }
 
 #if !defined(__cplusplus_winrt)
+
+template<typename CharType>
+class basic_throws_buffer : public streams::details::streambuf_state_manager<CharType>
+{
+public:
+    basic_throws_buffer() : streams::details::streambuf_state_manager<CharType>(std::ios_base::out) {}
+
+    bool can_seek() const override { return true; }
+    bool has_size() const override { return false; }
+    size_t buffer_size(std::ios_base::openmode) const override { return 0; }
+    void set_buffer_size(size_t, std::ios_base::openmode) override {}
+    size_t in_avail() const override { return 0; }
+    pos_type getpos(std::ios_base::openmode) const override { return 0; }
+    pos_type seekpos(pos_type, std::ios_base::openmode) { return 0; }
+    pos_type seekoff(off_type, std::ios_base::seekdir, std::ios_base::openmode) override { return 0; }
+    bool acquire(_Out_writes_(count) CharType*&, _In_ size_t&) override { return false; }
+    void release(_Out_writes_(count) CharType *, _In_ size_t) override {}
+
+protected:
+    pplx::task<int_type> _putc(CharType) override { throw std::runtime_error("error"); }
+    pplx::task<size_t> _putn(const CharType *, size_t) override { throw std::runtime_error("error"); }
+    pplx::task<int_type> _bumpc() override { throw std::runtime_error("error"); }
+    int_type _sbumpc() override { throw std::runtime_error("error"); }
+    pplx::task<int_type> _getc() override { throw std::runtime_error("error"); }
+    int_type _sgetc() override { throw std::runtime_error("error"); }
+    pplx::task<int_type> _nextc() override { throw std::runtime_error("error"); }
+    pplx::task<int_type> _ungetc() override { throw std::runtime_error("error"); }
+    pplx::task<size_t> _getn(_Out_writes_(count) CharType *, _In_ size_t) override { throw std::runtime_error("error"); }
+    size_t _scopy(_Out_writes_(count) CharType *, _In_ size_t) override { throw std::runtime_error("error"); }
+    pplx::task<bool> _sync() override { throw std::runtime_error("error"); }
+    CharType* _alloc(size_t) override { throw std::runtime_error("error"); }
+    void _commit(size_t) override { throw std::runtime_error("error"); }
+
+    pplx::task<void> _close_write() override
+    {
+        return pplx::task_from_exception<void>(std::invalid_argument("test"));
+    }
+};
+
+template<typename CharType>
+class close_throws_buffer : public streams::streambuf<CharType>
+{
+public:
+    close_throws_buffer() : streambuf<CharType>(std::shared_ptr<basic_throws_buffer<CharType>>(new basic_throws_buffer<CharType>())) {}
+};
+
+// Tests if an exception occurs and close throws an exception that the close
+// one is ignored and doesn't bring down the process.
+TEST_FIXTURE(uri_address, response_stream_close_throws_with_exception)
+{
+    web::http::experimental::listener::http_listener listener(m_uri);
+    listener.open().wait();
+
+    streams::producer_consumer_buffer<uint8_t> buf;
+
+    listener.support([buf](http_request request)
+    {
+        http_response response(200);
+        response.set_body(streams::istream(buf), U("text/plain"));
+        response.headers().add(header_names::connection, U("close"));
+        request.reply(response);
+    });
+
+    http_client_config config;
+    config.set_timeout(utility::seconds(1));
+    http_client client(m_uri, config);
+
+    close_throws_buffer<uint8_t> responseBody;
+    http_request msg(methods::GET);
+    msg.set_response_stream(responseBody.create_ostream());
+    http_response rsp = client.request(msg).get();
+    VERIFY_THROWS(rsp.content_ready().get(), http_exception);
+
+    buf.close(std::ios_base::out).wait();
+    listener.close().wait();
+}
+
 TEST_FIXTURE(uri_address, content_ready)
 {
     http_client client(m_uri);