create the next signal for use by Many

fbshipit-source-id: 77064a2ace9076a80dd8ce4dd69e386b5c5a496a

folly/experimental/pushmi/include/pushmi.h

@@ -1533,6 +1533,11 @@ PUSHMI_TEMPLATE (class S, class V)
 void set_value(S& s, V&& v) noexcept(noexcept(s.value((V&&) v))) {
   s.value((V&&) v);
 }
+PUSHMI_TEMPLATE (class S, class V)
+  (requires requires (std::declval<S&>().next(std::declval<V>())))
+void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
+  s.next((V&&) v);
+}
 
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires (std::declval<S&>().starting(std::declval<Up>())))
@@ -1602,6 +1607,12 @@ void set_value(std::reference_wrapper<S> s, V&& v) noexcept(
   noexcept(set_value(s.get(), (V&&) v))) {
   set_value(s.get(), (V&&) v);
 }
+PUSHMI_TEMPLATE (class S, class V)
+  (requires requires ( set_next(std::declval<S&>(), std::declval<V>()) ))
+void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
+  noexcept(set_next(s.get(), (V&&) v))) {
+  set_next(s.get(), (V&&) v);
+}
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires ( set_starting(std::declval<S&>(), std::declval<Up>()) ))
 void set_starting(std::reference_wrapper<S> s, Up up) noexcept(
@@ -1671,6 +1682,21 @@ struct set_value_fn {
     }
   }
 };
+struct set_next_fn {
+  PUSHMI_TEMPLATE (class S, class V)
+    (requires requires (
+      set_next(std::declval<S&>(), std::declval<V>()),
+      set_error(std::declval<S&>(), std::current_exception())
+    ))
+  void operator()(S&& s, V&& v) const
+      noexcept(noexcept(set_next(s, (V&&) v))) {
+    try {
+      set_next(s, (V&&) v);
+    } catch (...) {
+      set_error(s, std::current_exception());
+    }
+  }
+};
 
 struct set_starting_fn {
   PUSHMI_TEMPLATE (class S, class Up)
@@ -1726,6 +1752,7 @@ struct get_now_fn {
 PUSHMI_INLINE_VAR constexpr __adl::set_done_fn set_done{};
 PUSHMI_INLINE_VAR constexpr __adl::set_error_fn set_error{};
 PUSHMI_INLINE_VAR constexpr __adl::set_value_fn set_value{};
+PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{};
 PUSHMI_INLINE_VAR constexpr __adl::set_starting_fn set_starting{};
 PUSHMI_INLINE_VAR constexpr __adl::do_submit_fn submit{};
 PUSHMI_INLINE_VAR constexpr __adl::get_now_fn now{};
@@ -2142,6 +2169,9 @@ PUSHMI_CONCEPT_DEF(
 PUSHMI_CONCEPT_DEF(
   template (class S, class T, class E = std::exception_ptr)
   (concept ManyReceiver)(S, T, E),
+    requires(S& s, T&& t) (
+      ::pushmi::set_next(s, (T &&) t) // Semantics: called 0-N times.
+    ) &&
     NoneReceiver<S, E> &&
     SemiMovable<T> &&
     SemiMovable<E> &&
@@ -2218,11 +2248,13 @@ PUSHMI_CONCEPT_DEF(
       class S,
       class Up,
       class T,
+      class PT,
       class PE = std::exception_ptr,
       class E = PE)
   (concept FlowManyReceiver)(S, Up, T, PE, E),
     ManyReceiver<S, T, E> &&
-    FlowSingleReceiver<S, Up, T, PE, E>
+    ManyReceiver<Up, PT, PE> &&
+    FlowNoneReceiver<S, Up, PE, E>
 );
 
 PUSHMI_CONCEPT_DEF(
@@ -2368,6 +2400,11 @@ struct ignoreDF {
   void operator()() {}
 };
 
+struct ignoreNF {
+  template <class V>
+  void operator()(V&&) {}
+};
+
 struct ignoreStrtF {
   template <class Up>
   void operator()(Up&&) {}
@@ -2411,6 +2448,16 @@ struct passDDF {
   }
 };
 
+struct passDNXF {
+  PUSHMI_TEMPLATE(class V, class Data)
+    (requires requires (
+      ::pushmi::set_next(std::declval<Data&>(), std::declval<V>())
+    ) && Receiver<Data>)
+  void operator()(Data& out, V&& v) const {
+    ::pushmi::set_next(out, (V&&) v);
+  }
+};
+
 struct passDStrtF {
   PUSHMI_TEMPLATE(class Up, class Data)
     (requires requires (
@@ -2543,6 +2590,17 @@ auto on_done(Fn fn) -> on_done_fn<Fn> {
   return on_done_fn<Fn>{std::move(fn)};
 }
 
+template <class... Fns>
+struct on_next_fn : overload_fn<Fns...> {
+  constexpr on_next_fn() = default;
+  using overload_fn<Fns...>::overload_fn;
+};
+
+template <class... Fns>
+auto on_next(Fns... fns) -> on_next_fn<Fns...> {
+  return on_next_fn<Fns...>{std::move(fns)...};
+}
+
 template <class... Fns>
 struct on_starting_fn : overload_fn<Fns...> {
   constexpr on_starting_fn() = default;

folly/experimental/pushmi/include/pushmi/boosters.h

@@ -56,6 +56,11 @@ struct ignoreDF {
   void operator()() {}
 };
 
+struct ignoreNF {
+  template <class V>
+  void operator()(V&&) {}
+};
+
 struct ignoreStrtF {
   template <class Up>
   void operator()(Up&&) {}
@@ -99,6 +104,16 @@ struct passDDF {
   }
 };
 
+struct passDNXF {
+  PUSHMI_TEMPLATE(class V, class Data)
+    (requires requires (
+      ::pushmi::set_next(std::declval<Data&>(), std::declval<V>())
+    ) && Receiver<Data>)
+  void operator()(Data& out, V&& v) const {
+    ::pushmi::set_next(out, (V&&) v);
+  }
+};
+
 struct passDStrtF {
   PUSHMI_TEMPLATE(class Up, class Data)
     (requires requires (
@@ -231,6 +246,17 @@ auto on_done(Fn fn) -> on_done_fn<Fn> {
   return on_done_fn<Fn>{std::move(fn)};
 }
 
+template <class... Fns>
+struct on_next_fn : overload_fn<Fns...> {
+  constexpr on_next_fn() = default;
+  using overload_fn<Fns...>::overload_fn;
+};
+
+template <class... Fns>
+auto on_next(Fns... fns) -> on_next_fn<Fns...> {
+  return on_next_fn<Fns...>{std::move(fns)...};
+}
+
 template <class... Fns>
 struct on_starting_fn : overload_fn<Fns...> {
   constexpr on_starting_fn() = default;

folly/experimental/pushmi/include/pushmi/concepts.h

@@ -220,6 +220,9 @@ PUSHMI_CONCEPT_DEF(
 PUSHMI_CONCEPT_DEF(
   template (class S, class T, class E = std::exception_ptr)
   (concept ManyReceiver)(S, T, E),
+    requires(S& s, T&& t) (
+      ::pushmi::set_next(s, (T &&) t) // Semantics: called 0-N times.
+    ) &&
     NoneReceiver<S, E> &&
     SemiMovable<T> &&
     SemiMovable<E> &&
@@ -296,11 +299,13 @@ PUSHMI_CONCEPT_DEF(
       class S,
       class Up,
       class T,
+      class PT,
       class PE = std::exception_ptr,
       class E = PE)
   (concept FlowManyReceiver)(S, Up, T, PE, E),
     ManyReceiver<S, T, E> &&
-    FlowSingleReceiver<S, Up, T, PE, E>
+    ManyReceiver<Up, PT, PE> &&
+    FlowNoneReceiver<S, Up, PE, E>
 );
 
 PUSHMI_CONCEPT_DEF(

folly/experimental/pushmi/include/pushmi/extension_points.h

@@ -26,6 +26,11 @@ PUSHMI_TEMPLATE (class S, class V)
 void set_value(S& s, V&& v) noexcept(noexcept(s.value((V&&) v))) {
   s.value((V&&) v);
 }
+PUSHMI_TEMPLATE (class S, class V)
+  (requires requires (std::declval<S&>().next(std::declval<V>())))
+void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
+  s.next((V&&) v);
+}
 
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires (std::declval<S&>().starting(std::declval<Up>())))
@@ -95,6 +100,12 @@ void set_value(std::reference_wrapper<S> s, V&& v) noexcept(
   noexcept(set_value(s.get(), (V&&) v))) {
   set_value(s.get(), (V&&) v);
 }
+PUSHMI_TEMPLATE (class S, class V)
+  (requires requires ( set_next(std::declval<S&>(), std::declval<V>()) ))
+void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
+  noexcept(set_next(s.get(), (V&&) v))) {
+  set_next(s.get(), (V&&) v);
+}
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires ( set_starting(std::declval<S&>(), std::declval<Up>()) ))
 void set_starting(std::reference_wrapper<S> s, Up up) noexcept(
@@ -164,6 +175,21 @@ struct set_value_fn {
     }
   }
 };
+struct set_next_fn {
+  PUSHMI_TEMPLATE (class S, class V)
+    (requires requires (
+      set_next(std::declval<S&>(), std::declval<V>()),
+      set_error(std::declval<S&>(), std::current_exception())
+    ))
+  void operator()(S&& s, V&& v) const
+      noexcept(noexcept(set_next(s, (V&&) v))) {
+    try {
+      set_next(s, (V&&) v);
+    } catch (...) {
+      set_error(s, std::current_exception());
+    }
+  }
+};
 
 struct set_starting_fn {
   PUSHMI_TEMPLATE (class S, class Up)
@@ -219,6 +245,7 @@ struct get_now_fn {
 PUSHMI_INLINE_VAR constexpr __adl::set_done_fn set_done{};
 PUSHMI_INLINE_VAR constexpr __adl::set_error_fn set_error{};
 PUSHMI_INLINE_VAR constexpr __adl::set_value_fn set_value{};
+PUSHMI_INLINE_VAR constexpr __adl::set_next_fn set_next{};
 PUSHMI_INLINE_VAR constexpr __adl::set_starting_fn set_starting{};
 PUSHMI_INLINE_VAR constexpr __adl::do_submit_fn submit{};
 PUSHMI_INLINE_VAR constexpr __adl::get_now_fn now{};