Split notify_error() in two functions.

Author: mzimbres

include/boost/redis/detail/multiplexer.hpp

@@ -49,7 +49,7 @@ struct multiplexer {
          done_();
       }
 
-      auto notify_error(system::error_code ec) noexcept -> void;
+      auto set_error(system::error_code const& ec) noexcept -> void;
 
       [[nodiscard]]
       auto is_waiting() const noexcept
@@ -115,7 +115,9 @@ struct multiplexer {
       std::size_t read_size_;
    };
 
-   auto remove(std::shared_ptr<elem> const& ptr) -> bool;
+   using elem_ptr_type = std::shared_ptr<elem>;
+
+   auto remove(elem_ptr_type const& ptr) -> bool;
 
    [[nodiscard]]
    auto prepare_write() -> std::size_t;
@@ -129,7 +131,7 @@ struct multiplexer {
    [[nodiscard]]
    auto consume_next(system::error_code& ec) -> std::pair<tribool, std::size_t>;
 
-   auto add(std::shared_ptr<elem> const& ptr) -> void;
+   auto add(elem_ptr_type const& ptr) -> void;
    auto reset() -> void;
 
    [[nodiscard]]
@@ -203,7 +205,7 @@ struct multiplexer {
 
    std::string read_buffer_;
    std::string write_buffer_;
-   std::deque<std::shared_ptr<elem>> reqs_;
+   std::deque<elem_ptr_type> reqs_;
    resp3::parser parser_{};
    bool on_push_ = false;
    bool cancel_run_called_ = false;
@@ -212,7 +214,7 @@ struct multiplexer {
 };
 
 auto make_elem(request const& req, multiplexer::pipeline_adapter_type adapter)
-   -> std::shared_ptr<multiplexer::elem>;
+   -> multiplexer::elem_ptr_type;
 
 }  // namespace detail
 }  // namespace boost::redis

include/boost/redis/impl/multiplexer.ipp

@@ -25,13 +25,10 @@ multiplexer::elem::elem(request const& req, pipeline_adapter_type adapter)
    };
 }
 
-auto multiplexer::elem::notify_error(system::error_code ec) noexcept -> void
+auto multiplexer::elem::set_error(system::error_code const& ec) noexcept -> void
 {
-   if (!ec_) {
+   if (!ec_)
       ec_ = ec;
-   }
-
-   notify_done();
 }
 
 auto multiplexer::elem::commit_response(std::size_t read_size) -> void
@@ -40,7 +37,7 @@ auto multiplexer::elem::commit_response(std::size_t read_size) -> void
    --remaining_responses_;
 }
 
-bool multiplexer::remove(std::shared_ptr<elem> const& ptr)
+bool multiplexer::remove(elem_ptr_type const& ptr)
 {
    if (ptr->is_waiting()) {
       reqs_.erase(std::remove(std::begin(reqs_), std::end(reqs_), ptr));
@@ -68,7 +65,7 @@ std::size_t multiplexer::commit_write()
    return release_push_requests();
 }
 
-void multiplexer::add(std::shared_ptr<elem> const& info)
+void multiplexer::add(elem_ptr_type const& info)
 {
    reqs_.push_back(info);
 
@@ -118,7 +115,8 @@ std::pair<tribool, std::size_t> multiplexer::consume_next(system::error_code& ec
       return std::make_pair(std::nullopt, 0);
 
    if (ec) {
-      reqs_.front()->notify_error(ec);
+      reqs_.front()->set_error(ec);
+      reqs_.front()->notify_done();
       reqs_.pop_front();
       return std::make_pair(std::make_optional(true), 0);
    }
@@ -179,7 +177,8 @@ std::size_t multiplexer::cancel_waiting()
    auto const ret = std::distance(point, std::end(reqs_));
 
    std::for_each(point, std::end(reqs_), [](auto const& ptr) {
-      ptr->notify_error({asio::error::operation_aborted});
+      ptr->set_error({asio::error::operation_aborted});
+      ptr->notify_done();
    });
 
    reqs_.erase(point, std::end(reqs_));
@@ -210,7 +209,8 @@ auto multiplexer::cancel_on_conn_lost() -> std::size_t
    auto const ret = std::distance(point, std::end(reqs_));
 
    std::for_each(point, std::end(reqs_), [](auto const& ptr) {
-      ptr->notify_error({asio::error::operation_aborted});
+      ptr->set_error({asio::error::operation_aborted});
+      ptr->notify_done();
    });
 
    reqs_.erase(point, std::end(reqs_));
@@ -306,7 +306,7 @@ bool multiplexer::is_waiting_response() const noexcept
 bool multiplexer::is_writing() const noexcept { return !write_buffer_.empty(); }
 
 auto make_elem(request const& req, multiplexer::pipeline_adapter_type adapter)
-   -> std::shared_ptr<multiplexer::elem>
+   -> multiplexer::elem_ptr_type
 {
    return std::make_shared<multiplexer::elem>(req, std::move(adapter));
 }