Replace C-style cast with move() in let* algos

include/unifex/let_done.hpp

@@ -219,15 +219,22 @@ class _op<Source, Done, Receiver>::type {
 
 public:
   template <typename Done2, typename Receiver2>
-  explicit type(Source&& source, Done2&& done, Receiver2&& dest)
+  explicit type(
+      Source&& source, // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved)
+      Done2&& done, Receiver2&& dest)
       noexcept(std::is_nothrow_move_constructible_v<Receiver> &&
                std::is_nothrow_move_constructible_v<Done> &&
                is_nothrow_connectable_v<Source, source_receiver>)
   : done_((Done2&&)done)
   , receiver_((Receiver2&&)dest)
   {
+    // Note: 'Source' is not a forwarding reference since it's not deduced
+    // in this constructor. It can either be a Sender&& or Sender& for
+    // some concrete type Sender. Here, we want the forwarding behavior when
+    // the operation is constructed based on the type of Source, even though
+    // it's not a idiomatic use for std::forward.
     unifex::activate_union_member_with(sourceOp_, [&] {
-        return unifex::connect((Source&&)source, source_receiver{this});
+        return unifex::connect(std::forward<Source>(source), source_receiver{this});
       });
     startedOp_ = 0 + 1;
   }

include/unifex/let_error.hpp

@@ -257,14 +257,21 @@ class _op<Source, Func, Receiver>::type final {
 
 public:
   template <typename Func2, typename Receiver2>
-  explicit type(Source&& source, Func2&& func, Receiver2&& dest) noexcept(
+  explicit type(
+      Source&& source, // NOLINT(cppcoreguidelines-rvalue-reference-param-not-moved)
+      Func2&& func, Receiver2&& dest) noexcept(
       std::is_nothrow_constructible_v<Receiver, Receiver2&&>&&
           std::is_nothrow_constructible_v<Func, Func2&&>&&
               is_nothrow_connectable_v<Source, source_receiver>)
     : func_((Func2 &&) func)
     , receiver_((Receiver2 &&) dest) {
+    // Note: 'Source' is not a forwarding reference since it's not deduced
+    // in this constructor. It can either be a Sender&& or Sender& for
+    // some concrete type Sender. Here, we want the forwarding behavior when
+    // the operation is constructed based on the type of Source, even though
+    // it's not a idiomatic use for std::forward.
     unifex::activate_union_member_with(sourceOp_, [&] {
-      return unifex::connect((Source &&) source, source_receiver{this});
+      return unifex::connect(std::forward<Source>(source), source_receiver{this});
     });
   }
 
@@ -418,8 +425,8 @@ class _sndr<Source, Func>::type final {
         is_nothrow_connectable_v<member_t<Sender, Source>, SourceReceiver> &&
         std::is_nothrow_constructible_v<Func, member_t<Sender, Func>> &&
         std::is_nothrow_constructible_v<remove_cvref_t<Receiver>, Receiver>)
-      -> operation_type<Source, Func, Receiver> {
-    return operation_type<Source, Func, Receiver>{
+      -> operation_type<member_t<Sender, Source>, Func, Receiver> {
+    return operation_type<member_t<Sender, Source> , Func, Receiver>{
       static_cast<Sender&&>(s).source_,
       static_cast<Sender&&>(s).func_,
       static_cast<Receiver&&>(r)

include/unifex/let_value.hpp

@@ -31,6 +31,7 @@
 #include <functional>
 #include <tuple>
 #include <type_traits>
+#include <utility>
 
 #include <unifex/detail/prologue.hpp>
 
@@ -85,7 +86,7 @@ struct _successor_receiver<Operation, Values...>::type {
   void set_error(Error error) && noexcept {
     auto& op = op_;
     cleanup();
-    unifex::set_error(std::move(op.receiver_), (Error &&) error);
+    unifex::set_error(std::move(op.receiver_), std::move(error));
   }
 
 private:
@@ -178,7 +179,7 @@ struct _predecessor_receiver<Operation>::type {
   void set_error(Error error) && noexcept {
     auto& op = op_;
     unifex::deactivate_union_member(op.predOp_);
-    unifex::set_error(std::move(op.receiver_), (Error &&) error);
+    unifex::set_error(std::move(op.receiver_), std::move(error));
   }
 
   template(typename CPO)
@@ -227,16 +228,16 @@ struct _op<Predecessor, SuccessorFactory, Receiver>::type {
   template <typename Operation, typename... Values>
   friend struct _successor_receiver;
 
-  template <typename SuccessorFactory2, typename Receiver2>
+  template <typename Predecessor2, typename SuccessorFactory2, typename Receiver2>
   explicit type(
-      Predecessor&& pred,
+      Predecessor2&& pred,
       SuccessorFactory2&& func,
       Receiver2&& receiver)
       : func_((SuccessorFactory2 &&) func),
         receiver_((Receiver2 &&) receiver) {
     unifex::activate_union_member_with(predOp_, [&] {
       return unifex::connect(
-          (Predecessor &&) pred, predecessor_receiver<operation>{*this});
+          std::forward<Predecessor2>(pred), predecessor_receiver<operation>{*this});
     });
   }
 

include/unifex/let_value_with.hpp

@@ -23,6 +23,8 @@
 #include <unifex/execution_policy.hpp>
 #include <unifex/type_list.hpp>
 
+#include <utility>
+
 #include <unifex/detail/prologue.hpp>
 
 namespace unifex {
@@ -100,12 +102,12 @@ template<typename StateFactory, typename SuccessorFactory, typename Receiver>
 struct _operation<StateFactory, SuccessorFactory, Receiver>::type {
   template <typename StateFactory2, typename SuccessorFactory2, typename Receiver2>
   type(StateFactory2&& stateFactory, SuccessorFactory2&& func, Receiver2&& r) :
-    stateFactory_(static_cast<StateFactory2&&>(stateFactory)),
+    stateFactory_(std::forward<StateFactory2>(stateFactory)),
     func_(static_cast<SuccessorFactory2&&>(func)),
-    state_(static_cast<StateFactory&&>(stateFactory_)()),
+    state_(std::move(stateFactory_)()),
     innerOp_(
         unifex::connect(
-        static_cast<SuccessorFactory&&>(func_)(state_),
+        std::move(func_)(state_),
         static_cast<Receiver2&&>(r))) {
   }
 

include/unifex/let_value_with_stop_source.hpp

@@ -25,6 +25,8 @@
 #include <unifex/execution_policy.hpp>
 #include <unifex/type_list.hpp>
 
+#include <utility>
+
 #include <unifex/detail/prologue.hpp>
 
 namespace unifex {
@@ -165,7 +167,7 @@ struct _stop_source_operation<SuccessorFactory, Receiver>::type {
           nothrow_connectable) {
     return unifex::connect(
         static_cast<SuccessorFactory&&>(func)(stopSource_),
-        receiver_t{*this, static_cast<Receiver&&>(r)});
+        receiver_t{*this, std::move(r)});
   }
 
 public:

include/unifex/let_value_with_stop_token.hpp

@@ -25,6 +25,8 @@
 #include <unifex/sender_concepts.hpp>
 #include <unifex/type_list.hpp>
 
+#include <utility>
+
 #include <unifex/detail/prologue.hpp>
 
 namespace unifex {

test/let_done_test.cpp

@@ -21,6 +21,7 @@
 #include <unifex/just.hpp>
 #include <unifex/just_done.hpp>
 #include <unifex/on.hpp>
+#include <unifex/repeat_effect_until.hpp>
 #include <unifex/then.hpp>
 #include <unifex/sequence.hpp>
 #include <unifex/stop_when.hpp>
@@ -104,3 +105,11 @@ TEST(TransformDone, WithValue) {
   EXPECT_TRUE(multiple.has_value());
   EXPECT_EQ(*multiple, std::tuple(42, 1, 2));
 }
+
+TEST(TransformDone, LvalueConnectable) {
+  int n = 0;
+  sync_wait(repeat_effect_until(
+        let_done(just(), [] { return just(); }),
+    [&n]() mutable noexcept { return n++ == 5; }));
+  EXPECT_EQ(n, 6);
+}

test/let_error_test.cpp

@@ -20,6 +20,7 @@
 #include <unifex/let_done.hpp>
 #include <unifex/let_error.hpp>
 #include <unifex/on.hpp>
+#include <unifex/repeat_effect_until.hpp>
 #include <unifex/scheduler_concepts.hpp>
 #include <unifex/sequence.hpp>
 #include <unifex/stop_when.hpp>
@@ -173,6 +174,14 @@ TEST(TransformError, SequenceFwd) {
   EXPECT_EQ(*one, 42);
 }
 
+TEST(TransformError, LvalueConnectable) {
+  int n = 0;
+  sync_wait(repeat_effect_until(
+        let_error(just(), [](auto&&) { return just(); }),
+    [&n]() mutable noexcept { return n++ == 5; }));
+  EXPECT_EQ(n, 6);
+}
+
 #if !UNIFEX_NO_COROUTINES
 TEST(TransformError, WithTask) {
   auto value = let_error(

test/let_value_test.cpp

@@ -13,10 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-#include <unifex/just.hpp>
 #include <unifex/let_value.hpp>
+
+#include <unifex/just.hpp>
+#include <unifex/let_done.hpp>
 #include <unifex/let_value_with.hpp>
 #include <unifex/scheduler_concepts.hpp>
+#include <unifex/repeat_effect_until.hpp>
 #include <unifex/sync_wait.hpp>
 #include <unifex/timed_single_thread_context.hpp>
 #include <unifex/then.hpp>
@@ -281,3 +284,11 @@ TEST(Let, LetValueWithTraitlessPredecessor) {
   ASSERT_TRUE(ret);
   EXPECT_EQ(*ret, 42);
 }
+
+TEST(Let, LvalueConnectable) {
+  int n = 0;
+  sync_wait(repeat_effect_until(
+      let_value(let_done(just(), [] { return just(); }), [] { return just(); }),
+    [&n]() mutable noexcept { return n++ == 5; }));
+  EXPECT_EQ(n, 6);
+}

test/let_value_with_test.cpp

@@ -0,0 +1,51 @@
+#include <unifex/let_value_with.hpp>
+
+#include <unifex/just.hpp>
+#include <unifex/let_value_with.hpp>
+#include <unifex/sync_wait.hpp>
+#include <unifex/then.hpp>
+#include <unifex/timed_single_thread_context.hpp>
+#include <chrono>
+#include <iostream>
+#include <optional>
+#include <variant>
+
+#include <gtest/gtest.h>
+
+using namespace unifex;
+
+namespace {
+constexpr auto async = [](auto& context, auto&& func) {
+    return then(
+        schedule_after(context.get_scheduler(), std::chrono::milliseconds(10)),
+        (decltype(func))func);
+};
+}
+
+TEST(LetValueWith, StatefulFactory) {
+  // Verifies the let_value_with operation state holds onto the
+  // Factory object until the operation is complete.
+  timed_single_thread_context context;
+  std::optional<int> result = sync_wait(
+      let_value(just(), [&] {
+        return let_value_with([x = std::make_unique<int>(10)]() -> int* { return x.get(); }, [&](int*& v) {
+            return async(context, [&v]() { return *v; });
+          });
+      })
+    );
+  ASSERT_TRUE(!!result);
+  EXPECT_EQ(*result, 10);
+}
+
+TEST(LetValueWith, CallOperatorRvalueRefOverload) {
+  struct Factory {
+    int operator()() && {
+      return 10;
+    }
+  };
+  std::optional<int> result = sync_wait(let_value_with(Factory{}, [&](int& v) {
+      return just(v);
+    }));
+  ASSERT_TRUE(!!result);
+  EXPECT_EQ(*result, 10);
+}

test/static_thread_pool_test.cpp

@@ -16,8 +16,12 @@
 #include <unifex/static_thread_pool.hpp>
 
 #include <unifex/just.hpp>
+#include <unifex/let_done.hpp>
+#include <unifex/let_error.hpp>
 #include <unifex/on.hpp>
+#include <unifex/repeat_effect_until.hpp>
 #include <unifex/scheduler_concepts.hpp>
+#include <unifex/stop_when.hpp>
 #include <unifex/sync_wait.hpp>
 #include <unifex/then.hpp>
 #include <unifex/when_all.hpp>
@@ -62,3 +66,36 @@ TEST(StaticThreadPool, Smoke) {
 
   EXPECT_EQ(x, 3);
 }
+
+TEST(StaticThreadPool, ScheduleCancelationThreadSafety) {
+    static_thread_pool tpContext;
+    auto sch = tpContext.get_scheduler();
+
+    unifex::sync_wait(unifex::repeat_effect_until(
+        unifex::let_done(
+            unifex::stop_when(
+                unifex::repeat_effect(unifex::schedule(sch)),
+                unifex::schedule(sch)),
+            [] { return unifex::just(); }),
+        [n=0]() mutable noexcept { return n++ == 1000; }));
+
+    unifex::sync_wait(unifex::repeat_effect_until(
+        unifex::let_done(
+          unifex::let_error(
+            unifex::stop_when(
+                unifex::repeat_effect(unifex::schedule(sch)),
+                unifex::schedule(sch)),
+            [](auto&&) { return unifex::just(); }),
+          [] { return unifex::just(); }),
+        [n=0]() mutable noexcept { return n++ == 1000; }));
+
+    unifex::sync_wait(unifex::repeat_effect_until(
+        unifex::let_error(
+          unifex::let_done(
+            unifex::stop_when(
+                unifex::repeat_effect(unifex::schedule(sch)),
+                unifex::schedule(sch)),
+            [] { return unifex::just(); }),
+          [](auto&&) { return unifex::just(); }),
+        [n=0]() mutable noexcept { return n++ == 1000; }));
+}