fix

victimsnino · Oct 15, 2024 · bc99305 · bc99305
1 parent 82927fe
commit bc99305
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 2 deletions.
diff --git a/src/rpp/rpp/operators/details/combining_strategy.hpp b/src/rpp/rpp/operators/details/combining_strategy.hpp
@@ -49,7 +49,7 @@ namespace rpp::operators::details
     template<typename TState>
     struct combining_observer_strategy
     {
-        using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
+        using preferred_disposable_strategy = rpp::details::observers::locally_disposable_strategy;
 
         std::shared_ptr<TState> state{};
 

diff --git a/src/tests/rpp/test_merge.cpp b/src/tests/rpp/test_merge.cpp
@@ -216,18 +216,21 @@ TEST_CASE_TEMPLATE("merge handles race condition", TestType, rpp::memory_model::
             extracted_obs.emplace(std::forward<decltype(obs)>(obs).as_dynamic());
         });
 
+
         auto test = [&](auto source) {
             SUBCASE("subscribe on it")
             {
                 SUBCASE("on_error can't interleave with on_next")
                 {
+                    std::mutex m{};
                     source
                         | rpp::ops::as_blocking()
                         | rpp::ops::subscribe([&](auto&&) {
                                     REQUIRE(extracted_obs.has_value());
                                     CHECK(!on_error_called);
-                                    std::thread{[extracted_obs]
+                                    std::thread{[extracted_obs,&m]
                                     {
+                                        std::lock_guard lock{m};
                                         extracted_obs->on_error(std::exception_ptr{});
                                     }}.detach();
                                     std::this_thread::sleep_for(std::chrono::seconds{1});