add missing tests

fbshipit-source-id: 30a9f3571fb0a26d0023e3596105d2a38cf2d5e1

folly/experimental/pushmi/.gitignore

+# Prerequisites
+*.d
+
+# Compiled Object files
+*.slo
+*.lo
+*.o
+*.obj
+
+# Precompiled Headers
+*.gch
+*.pch
+
+# Compiled Dynamic libraries
+*.so
+*.dylib
+*.dll
+
+# Fortran module files
+*.mod
+*.smod
+
+# Compiled Static libraries
+*.lai
+*.la
+*.a
+*.lib
+
+# Executables
+*.exe
+*.out
+*.app
+
+build/**
\ No newline at end of file

folly/experimental/pushmi/test/CMakeLists.txt

 
 
-add_executable(PushmiTest catch.cpp PushmiTest.cpp)
+add_executable(PushmiTest catch.cpp CompileTest.cpp NewThreadTest.cpp TrampolineTest.cpp PushmiTest.cpp)
 target_link_libraries(PushmiTest
   pushmi
   Threads::Threads)

folly/experimental/pushmi/test/CompileTest.cpp

+#include <type_traits>
+
+#include <chrono>
+using namespace std::literals;
+
+#include "pushmi/flow_single_deferred.h"
+#include "pushmi/o/empty.h"
+#include "pushmi/o/just.h"
+#include "pushmi/o/on.h"
+#include "pushmi/o/transform.h"
+#include "pushmi/o/tap.h"
+#include "pushmi/o/via.h"
+#include "pushmi/o/submit.h"
+#include "pushmi/o/extension_operators.h"
+
+#include "pushmi/trampoline.h"
+#include "pushmi/new_thread.h"
+
+using namespace pushmi::aliases;
+
+void none_test() {
+  auto out0 = pushmi::none{};
+  auto out1 = pushmi::none{pushmi::abortEF{}};
+  auto out2 = pushmi::none{pushmi::abortEF{}, pushmi::ignoreDF{}};
+  auto out3 = pushmi::none{[](auto e) noexcept{ e.get(); }};
+  auto out5 = pushmi::none{
+      pushmi::on_error{[](auto e)noexcept {
+        e.get(); }, [](std::exception_ptr e) noexcept{}
+      }};
+  auto out6 = pushmi::none{
+      pushmi::on_done{[]() {  }}};
+
+  using Out0 = decltype(out0);
+
+  auto proxy0 = pushmi::none{out0};
+  auto proxy2 = pushmi::none{out0, pushmi::passDEF{}};
+  auto proxy3 = pushmi::none{
+      out0, pushmi::passDEF{}, pushmi::passDDF{}};
+  auto proxy4 = pushmi::none{out0, [](auto d, auto e)noexcept {
+    d.error(e.get());
+  }};
+  auto proxy5 = pushmi::none{
+      out0,
+      pushmi::on_error{[](Out0&, auto e) noexcept{ e.get(); },
+                       [](Out0&, std::exception_ptr e) noexcept{}}};
+  auto proxy6 = pushmi::none{
+      out0,
+      pushmi::on_done{[](Out0&) { }}};
+
+  std::promise<void> p0;
+  auto promise0 = pushmi::none{std::move(p0)};
+  promise0.done();
+
+  std::promise<void> p1;
+
+  auto any0 = pushmi::any_none<>(std::move(p1));
+  auto any1 = pushmi::any_none<>(std::move(promise0));
+  auto any2 = pushmi::any_none<>(out0);
+  auto any3 = pushmi::any_none<>(proxy0);
+}
+
+void deferred_test(){
+  auto in0 = pushmi::deferred{};
+  auto in1 = pushmi::deferred{pushmi::ignoreSF{}};
+  auto in3 = pushmi::deferred{[&](auto out){
+    in0.submit(pushmi::none{std::move(out), [](auto d, auto e) noexcept {
+      pushmi::set_error(d, e);
+    }});
+  }};
+  in3.submit(pushmi::none{});
+
+  std::promise<void> p0;
+  auto promise0 = pushmi::none{std::move(p0)};
+  in0 | ep::submit(std::move(promise0));
+
+  auto out0 = pushmi::none{[](auto e) noexcept {  }};
+  auto out1 = pushmi::none{out0, [](auto d, auto e) noexcept {}};
+  out1.error(std::exception_ptr{});
+  in3.submit(out1);
+
+  auto any0 = pushmi::any_deferred<>(in0);
+}
+
+void single_test() {
+  auto out0 = pushmi::single{};
+  auto out1 = pushmi::single{pushmi::ignoreVF{}};
+  auto out2 = pushmi::single{pushmi::ignoreVF{}, pushmi::abortEF{}};
+  auto out3 =
+      pushmi::single{pushmi::ignoreVF{}, pushmi::abortEF{}, pushmi::ignoreDF{}};
+  auto out4 = pushmi::single{[](auto v) { v.get(); }};
+  auto out5 = pushmi::single{
+      pushmi::on_value{[](auto v) { v.get(); }, [](int v) {}},
+      pushmi::on_error{
+        [](auto e)noexcept { e.get(); },
+        [](std::exception_ptr e) noexcept{}}};
+  auto out6 = pushmi::single{
+      pushmi::on_error{
+        [](auto e) noexcept{ e.get(); },
+        [](std::exception_ptr e) noexcept{}}};
+  auto out7 = pushmi::single{
+      pushmi::on_done{[]() {  }}};
+
+  using Out0 = decltype(out0);
+
+  auto proxy0 = pushmi::single{out0};
+  auto proxy1 = pushmi::single{out0, pushmi::passDVF{}};
+  auto proxy2 = pushmi::single{out0, pushmi::passDVF{}, pushmi::passDEF{}};
+  auto proxy3 = pushmi::single{
+      out0, pushmi::passDVF{}, pushmi::passDEF{}, pushmi::passDDF{}};
+  auto proxy4 = pushmi::single{out0, [](auto d, auto v) {
+    pushmi::set_value(d, v.get());
+  }};
+  auto proxy5 = pushmi::single{
+      out0,
+      pushmi::on_value{[](Out0&, auto v) { v.get(); }, [](Out0&, int v) {}},
+      pushmi::on_error{[](Out0&, auto e) noexcept { e.get(); },
+                       [](Out0&, std::exception_ptr e) noexcept {}}};
+  auto proxy6 = pushmi::single{
+      out0,
+      pushmi::on_error{[](Out0&, auto e) noexcept { e.get(); },
+                       [](Out0&, std::exception_ptr e) noexcept {}}};
+  auto proxy7 = pushmi::single{
+      out0,
+      pushmi::on_done{[](Out0&) { }}};
+
+  std::promise<int> p0;
+  auto promise0 = pushmi::single{std::move(p0)};
+  promise0.value(0);
+
+  std::promise<int> p1;
+
+  auto any0 = pushmi::any_single<int>(std::move(p1));
+  auto any1 = pushmi::any_single<int>(std::move(promise0));
+  auto any2 = pushmi::any_single<int>(out0);
+  auto any3 = pushmi::any_single<int>(proxy0);
+}
+
+void single_deferred_test(){
+  auto in0 = pushmi::single_deferred{};
+  auto in1 = pushmi::single_deferred{pushmi::ignoreSF{}};
+  auto in3 = pushmi::single_deferred{[&](auto out){
+    in0.submit(pushmi::single{std::move(out),
+      pushmi::on_value{[](auto d, int v){ pushmi::set_value(d, v); }}
+    });
+  }};
+
+  std::promise<int> p0;
+  auto promise0 = pushmi::single{std::move(p0)};
+  in0 | ep::submit(std::move(promise0));
+
+  auto out0 = pushmi::single{};
+  auto out1 = pushmi::single{out0, pushmi::on_value{[](auto d, int v){
+    pushmi::set_value(d, v);
+  }}};
+  in3.submit(out1);
+
+  auto any0 = pushmi::any_single_deferred<int>(in0);
+}
+
+void time_single_deferred_test(){
+  auto in0 = pushmi::time_single_deferred{};
+  auto in1 = pushmi::time_single_deferred{pushmi::ignoreSF{}};
+  auto in3 = pushmi::time_single_deferred{[&](auto tp, auto out){
+    in0.submit(tp, pushmi::single{std::move(out),
+      pushmi::on_value{[](auto d, int v){ pushmi::set_value(d, v); }}
+    });
+  }};
+  auto in4 = pushmi::time_single_deferred{pushmi::ignoreSF{}, pushmi::systemNowF{}};
+
+  std::promise<int> p0;
+  auto promise0 = pushmi::single{std::move(p0)};
+  in0.submit(in0.now(), std::move(promise0));
+
+  auto out0 = pushmi::single{};
+  auto out1 = pushmi::single{out0, pushmi::on_value{[](auto d, int v){
+    pushmi::set_value(d, v);
+  }}};
+  in3.submit(in0.now(), out1);
+
+  auto any0 = pushmi::any_time_single_deferred<int>(in0);
+
+  in3 | op::submit();
+  in3 | op::submit_at(in3.now() + 1s);
+  in3 | op::submit_after(1s);
+
+#if 0
+  auto tr = v::trampoline();
+  tr |
+    op::transform([]<class TR>(TR tr) {
+      return v::get_now(tr);
+    }) |
+    // op::submit(v::single{});
+    op::get<std::chrono::system_clock::time_point>();
+
+  std::vector<std::string> times;
+  auto push = [&](int time) {
+    return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); });
+  };
+
+  auto nt = v::new_thread();
+
+  auto out = v::any_single<v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>>{v::single{}};
+  (v::any_time_executor_ref{nt}).submit(v::get_now(nt), v::single{});
+
+  nt |
+    op::transform([&]<class NT>(NT nt){
+      // auto out = v::any_single<v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>>{v::single{}};
+      // nt.submit(v::get_now(nt), std::move(out));
+      // nt.submit(v::get_now(nt), v::single{});
+      // nt | op::submit_at(v::get_now(nt), std::move(out));
+      // nt |
+      // op::submit(v::single{});
+      // op::submit_at(nt | ep::get_now(), v::on_value{[](auto){}}) |
+      // op::submit_at(nt | ep::get_now(), v::on_value{[](auto){}}) |
+      // op::submit_after(20ms, v::single{}) ;//|
+      // op::submit_after(20ms, v::on_value{[](auto){}}) |
+      // op::submit_after(40ms, push(42));
+      return v::get_now(nt);
+    }) |
+   // op::submit(v::single{});
+   op::blocking_submit(v::single{});
+    // op::get<decltype(v::get_now(nt))>();
+    // op::get<std::chrono::system_clock::time_point>();
+
+  auto ref = v::any_time_executor_ref<std::exception_ptr, std::chrono::system_clock::time_point>{nt};
+#endif
+}
+
+void flow_single_test() {
+  auto out0 = pushmi::flow_single{};
+  auto out1 = pushmi::flow_single{pushmi::ignoreVF{}};
+  auto out2 = pushmi::flow_single{pushmi::ignoreVF{}, pushmi::abortEF{}};
+  auto out3 =
+      pushmi::flow_single{
+        pushmi::ignoreVF{},
+        pushmi::abortEF{},
+        pushmi::ignoreDF{}};
+  auto out4 = pushmi::flow_single{[](auto v) { v.get(); }};
+  auto out5 = pushmi::flow_single{
+      pushmi::on_value{[](auto v) { v.get(); }, [](int v) {}},
+      pushmi::on_error{
+        [](auto e)noexcept { e.get(); },
+        [](std::exception_ptr e) noexcept{}}};
+  auto out6 = pushmi::flow_single{
+      pushmi::on_error{
+        [](auto e) noexcept{ e.get(); },
+        [](std::exception_ptr e) noexcept{}}};
+  auto out7 = pushmi::flow_single{
+      pushmi::on_done{[]() {  }}};
+
+  auto out8 =
+      pushmi::flow_single{
+        pushmi::ignoreVF{},
+        pushmi::abortEF{},
+        pushmi::ignoreDF{},
+        pushmi::ignoreStpF{}};
+  auto out9 =
+      pushmi::flow_single{
+        pushmi::ignoreVF{},
+        pushmi::abortEF{},
+        pushmi::ignoreDF{},
+        pushmi::ignoreStpF{},
+        pushmi::ignoreStrtF{}};
+
+  using Out0 = decltype(out0);
+
+  auto proxy0 = pushmi::flow_single{out0};
+  auto proxy1 = pushmi::flow_single{out0, pushmi::passDVF{}};
+  auto proxy2 = pushmi::flow_single{out0, pushmi::passDVF{}, pushmi::passDEF{}};
+  auto proxy3 = pushmi::flow_single{
+      out0, pushmi::passDVF{}, pushmi::passDEF{}, pushmi::passDDF{}};
+  auto proxy4 = pushmi::flow_single{out0, [](auto d, auto v) {
+    pushmi::set_value(d, v.get());
+  }};
+  auto proxy5 = pushmi::flow_single{
+      out0,
+      pushmi::on_value{[](Out0&, auto v) { v.get(); }, [](Out0&, int v) {}},
+      pushmi::on_error{[](Out0&, auto e) noexcept { e.get(); },
+                       [](Out0&, std::exception_ptr e) noexcept {}}};
+  auto proxy6 = pushmi::flow_single{
+      out0,
+      pushmi::on_error{[](Out0&, auto e) noexcept { e.get(); },
+                       [](Out0&, std::exception_ptr e) noexcept {}}};
+  auto proxy7 = pushmi::flow_single{
+      out0,
+      pushmi::on_done{[](Out0&) { }}};
+
+  auto proxy8 = pushmi::flow_single{out0,
+    pushmi::passDVF{},
+    pushmi::passDEF{},
+    pushmi::passDDF{}};
+  auto proxy9 = pushmi::flow_single{out0,
+    pushmi::passDVF{},
+    pushmi::passDEF{},
+    pushmi::passDDF{},
+    pushmi::passDStpF{}};
+
+  auto any2 = pushmi::any_flow_single<int>(out0);
+  auto any3 = pushmi::any_flow_single<int>(proxy0);
+}
+
+void flow_single_deferred_test(){
+  auto in0 = pushmi::flow_single_deferred{};
+  auto in1 = pushmi::flow_single_deferred{pushmi::ignoreSF{}};
+  auto in3 = pushmi::flow_single_deferred{[&](auto out){
+    in0.submit(pushmi::flow_single{std::move(out),
+      pushmi::on_value{[](auto d, int v){ pushmi::set_value(d, v); }}
+    });
+  }};
+
+  auto out0 = pushmi::flow_single{};
+  auto out1 = pushmi::flow_single{out0, pushmi::on_value{[](auto d, int v){
+    pushmi::set_value(d, v);
+  }}};
+  in3.submit(out1);
+
+  auto any0 = pushmi::any_flow_single_deferred<int>(in0);
+}

folly/experimental/pushmi/test/NewThreadTest.cpp

+#include "catch.hpp"
+
+#include <type_traits>
+
+#include <chrono>
+using namespace std::literals;
+
+#include "pushmi/flow_single_deferred.h"
+#include "pushmi/o/empty.h"
+#include "pushmi/o/just.h"
+#include "pushmi/o/on.h"
+#include "pushmi/o/transform.h"
+#include "pushmi/o/tap.h"
+#include "pushmi/o/via.h"
+#include "pushmi/o/submit.h"
+#include "pushmi/o/extension_operators.h"
+
+#include "pushmi/trampoline.h"
+#include "pushmi/new_thread.h"
+
+using namespace pushmi::aliases;
+
+struct countdownsingle {
+  countdownsingle(int& c)
+      : counter(&c) {}
+
+  int* counter;
+
+  template <class ExecutorRef>
+  void operator()(ExecutorRef exec) {
+    if (--*counter > 0) {
+      exec | op::submit(*this);
+    }
+  }
+};
+
+SCENARIO( "new_thread executor", "[new_thread][deferred]" ) {
+
+  GIVEN( "A new_thread time_single_deferred" ) {
+    auto nt = v::new_thread();
+    using NT = decltype(nt);
+
+    // REQUIRE( v::TimeSingleDeferred<
+    //   NT, v::archetype_single,
+    //   NT&, std::exception_ptr> );
+    // REQUIRE( v::TimeExecutor<
+    //   NT&, v::archetype_single,
+    //   std::exception_ptr> );
+
+    auto any = v::any_time_executor{nt};
+
+    WHEN( "blocking submit now" ) {
+      auto signals = 0;
+      auto start = v::now(nt);
+      auto signaled = v::now(nt);
+      nt |
+        op::transform([](auto nt){ return nt | ep::now(); }) |
+        op::blocking_submit(
+          [&](auto at){
+            signaled = at;
+            signals += 100; },
+          [&](auto e) noexcept {  signals += 1000; },
+          [&](){ signals += 10; });
+
+      THEN( "the value signal is recorded once and the signal did not drift much" ) {
+        REQUIRE( signals == 100 );
+        INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
+        REQUIRE( signaled - start < 10s );
+      }
+    }
+
+    WHEN( "blocking get now" ) {
+      auto start = v::now(nt);
+      auto signaled = nt |
+        op::transform([](auto nt){
+          return v::now(nt);
+        }) |
+        op::get<std::chrono::system_clock::time_point>;
+
+      THEN( "the signal did not drift much" ) {
+        INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
+        REQUIRE( signaled - start < 10s );
+      }
+    }
+
+    WHEN( "submissions are ordered in time" ) {
+      std::vector<std::string> times;
+      auto push = [&](int time) {
+        return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); });
+      };
+      nt | op::blocking_submit(v::on_value{[push](auto nt) {
+        nt |
+            op::submit_after(40ms, push(40)) |
+            op::submit_after(10ms, push(10)) |
+            op::submit_after(20ms, push(20)) |
+            op::submit_after(10ms, push(11));
+      }});
+
+      THEN( "the items were pushed in time order not insertion order" ) {
+        REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"});
+      }
+    }
+
+    WHEN( "now is called" ) {
+      bool done = false;
+      nt | ep::now();
+      nt | op::blocking_submit([&](auto nt) {
+        nt | ep::now();
+        done = true;
+      });
+
+      THEN( "both calls to now() complete" ) {
+        REQUIRE( done == true );
+      }
+    }
+
+    WHEN( "virtual derecursion is triggered" ) {
+      int counter = 100'000;
+      std::function<void(pushmi::archtype_any_time_executor_ref exec)> recurse;
+      recurse = [&](pushmi::archtype_any_time_executor_ref nt) {
+        if (--counter <= 0)
+          return;
+        nt | op::submit(recurse);
+      };
+      nt | op::blocking_submit([&](auto nt) { recurse(nt); });
+
+      THEN( "all nested submissions complete" ) {
+        REQUIRE( counter == 0 );
+      }
+    }
+
+    WHEN( "static derecursion is triggered" ) {
+      int counter = 100'000;
+      countdownsingle single{counter};
+      nt | op::blocking_submit(single);
+      THEN( "all nested submissions complete" ) {
+        REQUIRE( counter == 0 );
+      }
+    }
+
+    WHEN( "used with on" ) {
+      std::vector<std::string> values;
+      auto deferred = pushmi::single_deferred([](auto out) {
+        ::pushmi::set_value(out, 2.0);
+        // ignored
+        ::pushmi::set_value(out, 1);
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
+      });
+      deferred | op::on([&](){return nt;}) |
+          op::blocking_submit(v::on_value{[&](auto v) { values.push_back(std::to_string(v)); }});
+      THEN( "only the first item was pushed" ) {
+        REQUIRE(values == std::vector<std::string>{"2.000000"});
+      }
+    }
+
+    WHEN( "used with via" ) {
+      std::vector<std::string> values;
+      auto deferred = pushmi::single_deferred([](auto out) {
+        ::pushmi::set_value(out, 2.0);
+        // ignored
+        ::pushmi::set_value(out, 1);
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
+      });
+      deferred | op::via([&](){return nt;}) |
+          op::blocking_submit(v::on_value{[&](auto v) { values.push_back(std::to_string(v)); }});
+      THEN( "only the first item was pushed" ) {
+        REQUIRE(values == std::vector<std::string>{"2.000000"});
+      }
+    }
+  }
+}

folly/experimental/pushmi/test/TrampolineTest.cpp

+#include "catch.hpp"
+
+#include <type_traits>
+
+#include <chrono>
+using namespace std::literals;
+
+#include "pushmi/flow_single_deferred.h"
+#include "pushmi/o/empty.h"
+#include "pushmi/o/just.h"
+#include "pushmi/o/on.h"
+#include "pushmi/o/transform.h"
+#include "pushmi/o/tap.h"
+#include "pushmi/o/via.h"
+#include "pushmi/o/submit.h"
+#include "pushmi/o/extension_operators.h"
+
+#include "pushmi/trampoline.h"
+#include "pushmi/new_thread.h"
+
+using namespace pushmi::aliases;
+
+struct countdownsingle {
+  countdownsingle(int& c)
+      : counter(&c) {}
+
+  int* counter;
+
+  template <class ExecutorRef>
+  void operator()(ExecutorRef exec) {
+    if (--*counter > 0) {
+      exec | op::submit(*this);
+    }
+  }
+};
+
+SCENARIO( "trampoline executor", "[trampoline][deferred]" ) {
+
+  GIVEN( "A trampoline time_single_deferred" ) {
+    auto tr = v::trampoline();
+    using TR = decltype(tr);
+
+    // REQUIRE( v::TimeSingleDeferred<
+    //   TR, v::archetype_single,
+    //   TR&, std::exception_ptr> );
+    // REQUIRE( v::TimeExecutor<
+    //   TR&, v::archetype_single,
+    //   std::exception_ptr> );
+
+    WHEN( "submit now" ) {
+      auto signals = 0;
+      auto start = v::now(tr);
+      auto signaled = v::now(tr);
+      tr |
+        op::transform([](auto tr){ return v::now(tr); }) |
+        op::submit(
+          [&](auto at){ signaled = at;
+            signals += 100; },
+          [&](auto e) noexcept {  signals += 1000; },
+          [&](){ signals += 10; });
+
+      THEN( "the value signal is recorded once and the signal did not drift much" ) {
+        REQUIRE( signals == 100 );
+        INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
+        REQUIRE( signaled - start < 10s );
+      }
+    }
+
+    WHEN( "blocking get now" ) {
+      auto start = v::now(tr);
+      auto signaled = tr |
+        op::transform([](auto tr){ return v::now(tr); }) |
+        op::get<decltype(v::now(tr))>;
+
+      THEN( "the signal did not drift much" ) {
+        INFO("The delay is " << ::Catch::Detail::stringify(signaled - start));
+        REQUIRE( signaled - start < 10s );
+      }
+    }
+
+    WHEN( "submissions are ordered in time" ) {
+      std::vector<std::string> times;
+      auto push = [&](int time) {
+        return v::on_value([&, time](auto) { times.push_back(std::to_string(time)); });
+      };
+      tr | op::submit(v::on_value{[push](auto tr) {
+        tr |
+            op::submit_after(40ms, push(40)) |
+            op::submit_after(10ms, push(10)) |
+            op::submit_after(20ms, push(20)) |
+            op::submit_after(10ms, push(11));
+      }});
+
+      THEN( "the items were pushed in time order not insertion order" ) {
+        REQUIRE( times == std::vector<std::string>{"10", "11", "20", "40"});
+      }
+    }
+
+    WHEN( "now is called" ) {
+      bool done = false;
+      tr | ep::now();
+      tr | op::submit([&](auto tr) {
+        tr | ep::now();
+        done = true;
+      });
+
+      THEN( "both calls to now() complete" ) {
+        REQUIRE( done == true );
+      }
+    }
+
+    WHEN( "virtual derecursion is triggered" ) {
+      int counter = 100'000;
+      std::function<void(pushmi::archtype_any_time_executor_ref exec)> recurse;
+      recurse = [&](pushmi::archtype_any_time_executor_ref tr) {
+        if (--counter <= 0)
+          return;
+        tr | op::submit(recurse);
+      };
+      tr | op::submit([&](auto exec) { recurse(exec); });
+
+      THEN( "all nested submissions complete" ) {
+        REQUIRE( counter == 0 );
+      }
+    }
+
+    WHEN( "static derecursion is triggered" ) {
+      int counter = 100'000;
+      countdownsingle single{counter};
+      tr | op::submit(single);
+      THEN( "all nested submissions complete" ) {
+        REQUIRE( counter == 0 );
+      }
+    }
+
+    WHEN( "used with on" ) {
+      std::vector<std::string> values;
+      auto deferred = pushmi::single_deferred([](auto out) {
+        ::pushmi::set_value(out, 2.0);
+        // ignored
+        ::pushmi::set_value(out, 1);
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
+      });
+      deferred | op::on([&](){return tr;}) |
+          op::submit(v::on_value{[&](auto v) { values.push_back(std::to_string(v)); }});
+      THEN( "only the first item was pushed" ) {
+        REQUIRE(values == std::vector<std::string>{"2.000000"});
+      }
+    }
+
+    WHEN( "used with via" ) {
+      std::vector<std::string> values;
+      auto deferred = pushmi::single_deferred([](auto out) {
+        ::pushmi::set_value(out, 2.0);
+        // ignored
+        ::pushmi::set_value(out, 1);
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::min());
+        ::pushmi::set_value(out, std::numeric_limits<int8_t>::max());
+      });
+      deferred | op::via([&](){return tr;}) |
+          op::submit(v::on_value{[&](auto v) { values.push_back(std::to_string(v)); }});
+      THEN( "only the first item was pushed" ) {
+        REQUIRE(values == std::vector<std::string>{"2.000000"});
+      }
+    }
+  }
+}