Fix test::stream cancellation

Author: ashtum

include/boost/beast2/impl/read.hpp

@@ -64,6 +64,11 @@ class read_until_op
                     pr_.parse(ec);
                     if(ec == http_proto::condition::need_more_input)
                     {
+                        if(!!self.cancelled())
+                        {
+                            ec = asio::error::operation_aborted;
+                            goto upcall;
+                        }
                         // specific to http_io::async_read_some
                         if(total_bytes_ != 0 && condition_(pr_))
                         {

include/boost/beast2/test/detail/stream_state.hpp

@@ -116,9 +116,6 @@ struct stream_state
 
     void
     notify_read();
-
-    void
-    cancel_read();
 };
 
 //------------------------------------------------------------------------------
@@ -236,22 +233,6 @@ notify_read()
         cv.notify_all();
     }
 }
-
-inline
-void
-stream_state::
-cancel_read()
-{
-    std::unique_ptr<stream_read_op_base> p;
-    {
-        std::lock_guard<std::mutex> lock(m);
-        code = stream_status::eof;
-        p = std::move(op);
-    }
-    if(p != nullptr)
-        (*p)(asio::error::operation_aborted);
-}
-
 } // detail
 } // test
 } // beast2

include/boost/beast2/test/impl/stream.hpp

@@ -74,10 +74,18 @@ class basic_stream<Executor>::read_op : public detail::stream_read_op_base
             void
             operator()(asio::cancellation_type type) const
             {
-                if (type != asio::cancellation_type::none)
+                if(type != asio::cancellation_type::none)
                 {
-                    if (auto sp = wp_.lock())
-                        sp->cancel_read();
+                    if(auto sp = wp_.lock())
+                    {
+                        std::unique_ptr<stream_read_op_base> p;
+                        {
+                            std::lock_guard<std::mutex> lock(sp->m);
+                            p = std::move(sp->op);
+                        }
+                        if(p != nullptr)
+                            (*p)(asio::error::operation_aborted);            
+                    }
                 }
             }
 
@@ -515,7 +523,11 @@ void
 basic_stream<Executor>::
 close()
 {
-    in_->cancel_read();
+    {
+        std::lock_guard<std::mutex> lock(in_->m);
+        in_->code = detail::stream_status::eof;
+        in_->notify_read();
+    }
 
     // disconnect
     {

test/unit/read.cpp

@@ -133,18 +133,44 @@ class read_test
             pr.reset();
             pr.start();
 
-            // read header
+            // async_read_some cancels after reading 0 bytes
+            async_read_some(
+                ts,
+                pr,
+                asio::bind_cancellation_slot(
+                    c_signal.slot(),
+                    [](system::error_code ec, std::size_t n)
+                    {
+                        BOOST_TEST_EQ(n, 0);
+                        BOOST_TEST_EQ(ec, asio::error::operation_aborted);
+                    }));
+            c_signal.emit(asio::cancellation_type::total);
+            test::run(ioc);
+
+            // append 8 bytes of the msg
+            ts.append(msg.substr(0, 8));
+
+            // async_read_some cancels after reading 8 bytes
             async_read_some(
                 ts,
                 pr,
                 asio::bind_cancellation_slot(
                     c_signal.slot(),
-                    test::fail_handler(asio::error::operation_aborted)));
+                    [](system::error_code ec, std::size_t n)
+                    {
+                        BOOST_TEST_EQ(n, 8);
+                        BOOST_TEST_EQ(ec, asio::error::operation_aborted);
+                    }));
+            c_signal.emit(asio::cancellation_type::total);
+            test::run(ioc);
 
-            // send a cancellation
-            asio::post(ioc, [&](){ c_signal.emit(asio::cancellation_type::total); });
+            // append rest of the msg
+            ts.append(msg.substr(8, msg.npos));
 
+            // async_read_some succeeds
+            async_read_some(ts, pr, test::success_handler());
             test::run(ioc);
+            BOOST_TEST(pr.got_header());
         }
     }