From 558fc5e7cfddf48ce2fc6507ee390dc04fcba424 Mon Sep 17 00:00:00 2001
From: Andres Salamanca <andrealebarbaritos@gmail.com>
Date: Mon, 28 Jul 2025 14:29:09 -0500
Subject: [PATCH 1/2] Make bulk pipeable

---
 include/beman/execution/detail/bulk.hpp  | 10 +++++++++-
 tests/beman/execution/exec-bulk.test.cpp | 14 +++++++-------
 2 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/include/beman/execution/detail/bulk.hpp b/include/beman/execution/detail/bulk.hpp
index cbb7d778..d1453ccf 100644
--- a/include/beman/execution/detail/bulk.hpp
+++ b/include/beman/execution/detail/bulk.hpp
@@ -4,6 +4,8 @@
 #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_BULK
 #define INCLUDED_BEMAN_EXECUTION_DETAIL_BULK
 
+#include "beman/execution/detail/sender_adaptor.hpp"
+#include "beman/execution/detail/sender_adaptor_closure.hpp"
 #include <beman/execution/detail/get_completion_signatures.hpp>
 #include <beman/execution/detail/meta_combine.hpp>
 #include <beman/execution/detail/meta_unique.hpp>
@@ -29,7 +31,13 @@
 #include <beman/execution/detail/suppress_push.hpp>
 namespace beman::execution::detail {
 
-struct bulk_t {
+struct bulk_t : ::beman::execution::sender_adaptor_closure<bulk_t> {
+
+    template <class Shape, class f>
+        requires(std::is_integral_v<Shape> && ::beman::execution::detail::movable_value<f>)
+    auto operator()(Shape&& shape, f&& fun) const {
+        return beman::execution::detail::sender_adaptor{*this, std::forward<Shape>(shape), std::forward<f>(fun)};
+    }
 
     template <class Sender, class Shape, class f>
         requires(::beman::execution::sender<Sender> && std::is_integral_v<Shape> &&
diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp
index 001dae5a..91fdf6a4 100644
--- a/tests/beman/execution/exec-bulk.test.cpp
+++ b/tests/beman/execution/exec-bulk.test.cpp
@@ -12,7 +12,7 @@
 
 namespace {
 auto test_bulk() {
-    auto b0 = test_std::bulk(test_std::just(), 1, [](int) {});
+    auto b0 = test_std::just() | test_std::bulk(1, [](int) {});
 
     static_assert(test_std::sender<decltype(b0)>);
     auto b0_env         = test_std::get_env(b0);
@@ -25,7 +25,7 @@ auto test_bulk() {
 
     int counter = 0;
 
-    auto b1 = test_std::bulk(test_std::just(), 5, [&](int i) { counter += i; });
+    auto b1 = test_std::just() | test_std::bulk(5, [&](int i) { counter += i; });
 
     static_assert(test_std::sender<decltype(b1)>);
     auto b1_env         = test_std::get_env(b0);
@@ -43,9 +43,9 @@ auto test_bulk() {
 
     std::vector<int> results(a.size(), 0);
 
-    auto b2 = test_std::bulk(test_std::just(a), a.size(), [&](std::size_t index, const std::vector<int>& vec) {
-        results[index] = vec[index] * b[index];
-    });
+    auto b2 = test_std::just(a) | test_std::bulk(a.size(), [&](std::size_t index, const std::vector<int>& vec) {
+                  results[index] = vec[index] * b[index];
+              });
     static_assert(test_std::sender<decltype(b2)>);
     auto b2_env         = test_std::get_env(b2);
     auto b2_completions = test_std::get_completion_signatures(b2, b2_env);
@@ -65,7 +65,7 @@ auto test_bulk() {
 }
 
 auto test_bulk_noexept() {
-    auto b0             = test_std::bulk(test_std::just(), 1, [](int) noexcept {});
+    auto b0             = test_std::just() | test_std::bulk(1, [](int) noexcept {});
     auto b0_env         = test_std::get_env(b0);
     auto b0_completions = test_std::get_completion_signatures(b0, b0_env);
     static_assert(std::is_same_v<decltype(b0_completions),
@@ -75,7 +75,7 @@ auto test_bulk_noexept() {
 
     int counter = 0;
 
-    auto b1 = test_std::bulk(test_std::just(), 5, [&](int i) noexcept { counter += i; });
+    auto b1 = test_std::just() | test_std::bulk(5, [&](int i) noexcept { counter += i; });
 
     static_assert(test_std::sender<decltype(b1)>);
     auto b1_env         = test_std::get_env(b0);

From 2208ca9d8189c0fa33636a81e3ccd151fd0969a4 Mon Sep 17 00:00:00 2001
From: Andres Salamanca <andrealebarbaritos@gmail.com>
Date: Tue, 29 Jul 2025 10:37:40 -0500
Subject: [PATCH 2/2] Add separate tests for pipable bulk syntax

---
 tests/beman/execution/exec-bulk.test.cpp | 68 +++++++++++++++++++++---
 1 file changed, 61 insertions(+), 7 deletions(-)

diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp
index 91fdf6a4..37409f45 100644
--- a/tests/beman/execution/exec-bulk.test.cpp
+++ b/tests/beman/execution/exec-bulk.test.cpp
@@ -12,7 +12,7 @@
 
 namespace {
 auto test_bulk() {
-    auto b0 = test_std::just() | test_std::bulk(1, [](int) {});
+    auto b0 = test_std::bulk(test_std::just(), 1, [](int) {});
 
     static_assert(test_std::sender<decltype(b0)>);
     auto b0_env         = test_std::get_env(b0);
@@ -25,7 +25,7 @@ auto test_bulk() {
 
     int counter = 0;
 
-    auto b1 = test_std::just() | test_std::bulk(5, [&](int i) { counter += i; });
+    auto b1 = test_std::bulk(test_std::just(), 5, [&](int i) { counter += i; });
 
     static_assert(test_std::sender<decltype(b1)>);
     auto b1_env         = test_std::get_env(b0);
@@ -43,9 +43,9 @@ auto test_bulk() {
 
     std::vector<int> results(a.size(), 0);
 
-    auto b2 = test_std::just(a) | test_std::bulk(a.size(), [&](std::size_t index, const std::vector<int>& vec) {
-                  results[index] = vec[index] * b[index];
-              });
+    auto b2 = test_std::bulk(test_std::just(a), a.size(), [&](std::size_t index, const std::vector<int>& vec) {
+        results[index] = vec[index] * b[index];
+    });
     static_assert(test_std::sender<decltype(b2)>);
     auto b2_env         = test_std::get_env(b2);
     auto b2_completions = test_std::get_completion_signatures(b2, b2_env);
@@ -65,7 +65,7 @@ auto test_bulk() {
 }
 
 auto test_bulk_noexept() {
-    auto b0             = test_std::just() | test_std::bulk(1, [](int) noexcept {});
+    auto b0             = test_std::bulk(test_std::just(), 1, [](int) noexcept {});
     auto b0_env         = test_std::get_env(b0);
     auto b0_completions = test_std::get_completion_signatures(b0, b0_env);
     static_assert(std::is_same_v<decltype(b0_completions),
@@ -75,7 +75,7 @@ auto test_bulk_noexept() {
 
     int counter = 0;
 
-    auto b1 = test_std::just() | test_std::bulk(5, [&](int i) noexcept { counter += i; });
+    auto b1 = test_std::bulk(test_std::just(), 5, [&](int i) noexcept { counter += i; });
 
     static_assert(test_std::sender<decltype(b1)>);
     auto b1_env         = test_std::get_env(b0);
@@ -87,6 +87,60 @@ auto test_bulk_noexept() {
     ASSERT(counter == 10);
 }
 
+auto test_bulk_pipeable() {
+    auto b0 = test_std::just() | test_std::bulk(1, [](int) {});
+
+    static_assert(test_std::sender<decltype(b0)>);
+    auto b0_env         = test_std::get_env(b0);
+    auto b0_completions = test_std::get_completion_signatures(b0, b0_env);
+    static_assert(
+        std::is_same_v<decltype(b0_completions),
+                       beman::execution::completion_signatures<beman::execution::set_value_t(),
+                                                               beman::execution::set_error_t(std::exception_ptr)> >,
+        "Completion signatures do not match!");
+
+    int counter = 0;
+
+    auto b1 = test_std::just() | test_std::bulk(5, [&](int i) { counter += i; });
+
+    static_assert(test_std::sender<decltype(b1)>);
+    auto b1_env         = test_std::get_env(b0);
+    auto b1_completions = test_std::get_completion_signatures(b1, b1_env);
+    static_assert(
+        std::is_same_v<decltype(b1_completions),
+                       beman::execution::completion_signatures<beman::execution::set_value_t(),
+                                                               beman::execution::set_error_t(std::exception_ptr)> >,
+        "Completion signatures do not match!");
+    test_std::sync_wait(b1);
+    ASSERT(counter == 10);
+
+    std::vector<int> a{1, 2, 3, 4, 5, 6, 7, 8};
+    std::vector<int> b{9, 10, 11, 13, 14, 15, 16, 17};
+
+    std::vector<int> results(a.size(), 0);
+
+    auto b2 = test_std::just(a) | test_std::bulk(a.size(), [&](std::size_t index, const std::vector<int>& vec) {
+                  results[index] = vec[index] * b[index];
+              });
+
+    static_assert(test_std::sender<decltype(b2)>);
+    auto b2_env         = test_std::get_env(b2);
+    auto b2_completions = test_std::get_completion_signatures(b2, b2_env);
+    static_assert(
+        std::is_same_v<decltype(b2_completions),
+                       beman::execution::completion_signatures<beman::execution::set_value_t(std::vector<int>),
+                                                               beman::execution::set_error_t(std::exception_ptr)> >,
+        "Completion signatures do not match!");
+    test_std::sync_wait(b2);
+
+    // Expected results: element-wise multiplication of a and b
+    std::vector<int> expected{9, 20, 33, 52, 70, 90, 112, 136};
+
+    for (size_t i = 0; i < results.size(); ++i) {
+        ASSERT(results[i] == expected[i]);
+    }
+}
+
 } // namespace
 
 TEST(exec_bulk) {