refactor: remove unnecessary layer of consumer inheritance

phlex/core/CMakeLists.txt

@@ -24,7 +24,6 @@ cet_make_library(
   node_catalog.cpp
   producer_catalog.cpp
   product_selector.cpp
-  products_consumer.cpp
   provider_node.cpp
   registrar.cpp
   registration_api.cpp
@@ -69,7 +68,6 @@ install(
     observer_node.hpp
     predicate_node.hpp
     product_selector.hpp
-    products_consumer.hpp
     provider_node.hpp
     registrar.hpp
     registration_api.hpp

phlex/core/consumer.cpp

@@ -1,28 +1,108 @@
 #include "phlex/core/consumer.hpp"
 
+#include "phlex/core/message.hpp"
+#include "phlex/core/product_selector.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/identifier.hpp"
+#include "phlex/utilities/bulleted_list.hpp"
 
+#include <fmt/format.h>
+#include <oneapi/tbb/flow_graph.h>
+
+#include <cstddef>
+#include <memory>
+#include <stdexcept>
 #include <string>
 #include <utility>
 #include <vector>
 
+namespace {
+  std::vector<phlex::experimental::identifier> layers_from(phlex::product_selectors const& queries)
+  {
+    using namespace phlex::experimental::literals;
+    std::vector<phlex::experimental::identifier> result;
+    result.reserve(queries.size());
+    for (auto const& query : queries) {
+      if (query.layer) {
+        result.push_back(query.layer);
+      } else {
+        result.push_back("*"_id);
+      }
+    }
+    return result;
+  }
+
+  void validate_layers(phlex::detail::require_layers layers_required,
+                       phlex::product_selectors const& inputs,
+                       phlex::experimental::algorithm_name const& algo)
+  {
+    using namespace phlex::detail;
+    if (layers_required == require_layers::multi_input_only && inputs.size() <= 1) {
+      return;
+    }
+    std::vector<std::string> err_selectors{};
+    for (auto const& p : inputs) {
+      if (!p.layer) {
+        err_selectors.push_back(p.to_string());
+      }
+    }
+    if (!err_selectors.empty()) {
+      std::string const error =
+        fmt::format("Must specify layers in the product selectors for node {}:\n"
+                    "  (Only invalid selectors are listed)\n{}",
+                    algo.to_string(),
+                    bulleted_list(err_selectors));
+      throw std::runtime_error(error);
+    }
+  }
+}
+
 namespace phlex::detail {
   consumer::consumer(phlex::experimental::algorithm_name name,
-                     std::vector<std::string> predicates) :
-    name_{std::move(name)}, predicates_{std::move(predicates)}
+                     std::vector<std::string> predicates,
+                     product_selectors input_products,
+                     tbb::flow::graph& graph,
+                     require_layers layers_required) :
+    name_{std::move(name)},
+    predicates_{std::move(predicates)},
+    graph_{graph},
+    input_products_{std::move(input_products)},
+    layers_{layers_from(input_products_)}
   {
+    validate_layers(layers_required, input_products_, this->name());
   }
 
+  consumer::~consumer() = default;
+
   phlex::experimental::algorithm_name const& consumer::name() const noexcept { return name_; }
-  phlex::experimental::identifier const& consumer::plugin() const noexcept
+
+  std::vector<std::string> const& consumer::when() const noexcept { return predicates_; }
+
+  tbb::flow::receiver<message>& consumer::port(product_selector const& input_product)
   {
-    return name_.plugin();
+    auto& next = port_for(input_product);
+
+    // If input_product doesn't have a layer, it must be for a node that allows layer omission
+    if (input_product.layer) {
+      auto& layer_check = layer_checkers_.emplace_back(std::make_unique<layer_check_node_t>(
+        graph_,
+        tbb::flow::unlimited,
+        [&layer = static_cast<experimental::identifier const&>(input_product.layer)](
+          message const& msg, auto& output) {
+          if (msg.store->layer_name() == layer) {
+            std::get<0>(output).try_put(msg);
+          }
+        }));
+      make_edge(tbb::flow::output_port<0>(*layer_check), next);
+      return *layer_check;
+    }
+    // else
+    return next;
   }
-  phlex::experimental::identifier const& consumer::algorithm() const noexcept
+
+  product_selectors const& consumer::input() const noexcept { return input_products_; }
+  std::vector<phlex::experimental::identifier> const& consumer::layers() const noexcept
   {
-    return name_.algorithm();
+    return layers_;
   }
-
-  std::vector<std::string> const& consumer::when() const noexcept { return predicates_; }
 }

phlex/core/consumer.hpp

@@ -1,25 +1,60 @@
 #ifndef PHLEX_CORE_CONSUMER_HPP
 #define PHLEX_CORE_CONSUMER_HPP
 
+#include "phlex/core/fwd.hpp"
+#include "phlex/core/input_arguments.hpp"
+#include "phlex/core/message.hpp"
+#include "phlex/core/product_selector.hpp"
 #include "phlex/model/algorithm_name.hpp"
+#include "phlex/model/identifier.hpp"
 #include "phlex/phlex_core_export.hpp"
 
+#include <oneapi/tbb/flow_graph.h>
+
+#include <cstddef>
+#include <functional>
+#include <memory>
 #include <string>
 #include <vector>
 
 namespace phlex::detail {
+  enum class require_layers : char { multi_input_only, always };
   class PHLEX_CORE_EXPORT consumer {
+    using layer_check_node_t = tbb::flow::multifunction_node<message, message_tuple<1UZ>>;
+
   public:
-    consumer(phlex::experimental::algorithm_name name, std::vector<std::string> predicates);
+    consumer(phlex::experimental::algorithm_name name,
+             std::vector<std::string> predicates,
+             product_selectors input_products,
+             tbb::flow::graph& graph,
+             require_layers layers_required);
+    virtual ~consumer();
 
     phlex::experimental::algorithm_name const& name() const noexcept;
-    phlex::experimental::identifier const& plugin() const noexcept;
-    phlex::experimental::identifier const& algorithm() const noexcept;
     std::vector<std::string> const& when() const noexcept;
 
+    product_selectors const& input() const noexcept;
+    std::vector<phlex::experimental::identifier> const& layers() const noexcept;
+    tbb::flow::receiver<message>& port(product_selector const& input_product);
+
+    virtual named_index_ports index_ports() = 0;
+    virtual std::size_t num_calls() const = 0;
+
+  protected:
+    template <typename InputParameterTuple>
+    auto input_arguments()
+    {
+      return form_input_arguments<InputParameterTuple>(input_products_);
+    }
+
   private:
+    virtual tbb::flow::receiver<message>& port_for(product_selector const& input_product) = 0;
     phlex::experimental::algorithm_name name_;
     std::vector<std::string> predicates_;
+    std::reference_wrapper<tbb::flow::graph> graph_;
+    product_selectors input_products_;
+    std::vector<phlex::experimental::identifier> layers_;
+    std::vector<std::unique_ptr<layer_check_node_t>> layer_checkers_;
   };
 }
 

phlex/core/declared_fold.cpp

@@ -1,7 +1,7 @@
 #include "phlex/core/declared_fold.hpp"
 
+#include "phlex/core/consumer.hpp"
 #include "phlex/core/product_selector.hpp"
-#include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 
 #include <oneapi/tbb/flow_graph.h>
@@ -16,11 +16,11 @@ namespace phlex::detail {
                                product_selectors input_products,
                                tbb::flow::graph& graph,
                                std::string partition_layer) :
-    products_consumer{std::move(name),
-                      std::move(predicates),
-                      std::move(input_products),
-                      graph,
-                      require_layers::always},
+    consumer{std::move(name),
+             std::move(predicates),
+             std::move(input_products),
+             graph,
+             require_layers::always},
     partition_layer_{std::move(partition_layer)}
   {
   }

phlex/core/declared_fold.hpp

@@ -1,10 +1,10 @@
 #ifndef PHLEX_CORE_DECLARED_FOLD_HPP
 #define PHLEX_CORE_DECLARED_FOLD_HPP
 
+#include "phlex/core/consumer.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/producer.hpp"
 #include "phlex/core/product_selector.hpp"
-#include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/identifier.hpp"
 #include "phlex/phlex_core_export.hpp"
@@ -17,7 +17,7 @@
 #include <vector>
 
 namespace phlex::detail {
-  class PHLEX_CORE_EXPORT declared_fold : public products_consumer, public producer {
+  class PHLEX_CORE_EXPORT declared_fold : public consumer, public producer {
   public:
     declared_fold(phlex::experimental::algorithm_name name,
                   std::vector<std::string> predicates,
@@ -28,7 +28,7 @@ namespace phlex::detail {
 
     phlex::experimental::algorithm_name const& name() const noexcept override
     {
-      return products_consumer::name();
+      return consumer::name();
     }
 
     virtual tbb::flow::receiver<index_message>& partition_port() = 0;

phlex/core/declared_observer.cpp

@@ -1,7 +1,7 @@
 #include "phlex/core/declared_observer.hpp"
 
+#include "phlex/core/consumer.hpp"
 #include "phlex/core/product_selector.hpp"
-#include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 
 #include <oneapi/tbb/flow_graph.h>
@@ -15,11 +15,11 @@ namespace phlex::detail {
                                        std::vector<std::string> predicates,
                                        product_selectors input_products,
                                        tbb::flow::graph& graph) :
-    products_consumer{std::move(name),
-                      std::move(predicates),
-                      std::move(input_products),
-                      graph,
-                      require_layers::multi_input_only}
+    consumer{std::move(name),
+             std::move(predicates),
+             std::move(input_products),
+             graph,
+             require_layers::multi_input_only}
   {
   }
 

phlex/core/declared_observer.hpp

@@ -1,8 +1,8 @@
 #ifndef PHLEX_CORE_DECLARED_OBSERVER_HPP
 #define PHLEX_CORE_DECLARED_OBSERVER_HPP
 
+#include "phlex/core/consumer.hpp"
 #include "phlex/core/product_selector.hpp"
-#include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/phlex_core_export.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
@@ -14,7 +14,7 @@
 #include <vector>
 
 namespace phlex::detail {
-  class PHLEX_CORE_EXPORT declared_observer : public products_consumer {
+  class PHLEX_CORE_EXPORT declared_observer : public consumer {
   public:
     declared_observer(phlex::experimental::algorithm_name name,
                       std::vector<std::string> predicates,

phlex/core/declared_output.cpp

@@ -1,8 +1,8 @@
 #include "phlex/core/declared_output.hpp"
 
-#include "phlex/core/consumer.hpp"
 #include "phlex/core/detail/make_algorithm_name.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/model/algorithm_name.hpp"
 
 #include <oneapi/tbb/flow_graph.h>
 
@@ -17,7 +17,8 @@ namespace phlex::detail {
                                    std::vector<std::string> predicates,
                                    tbb::flow::graph& g,
                                    internal::output_function_t&& ft) :
-    consumer{std::move(name), std::move(predicates)},
+    name_{std::move(name)},
+    predicates_{std::move(predicates)},
     node_{g, concurrency, [this, f = std::move(ft)](message const& msg) -> tbb::flow::continue_msg {
             f(*msg.store);
             ++calls_;
@@ -26,5 +27,11 @@ namespace phlex::detail {
   {
   }
 
+  phlex::experimental::algorithm_name const& declared_output::name() const noexcept
+  {
+    return name_;
+  }
+  std::vector<std::string> const& declared_output::when() const noexcept { return predicates_; }
+
   tbb::flow::receiver<message>& declared_output::port() noexcept { return node_; }
 }

phlex/core/declared_output.hpp

@@ -1,10 +1,10 @@
 #ifndef PHLEX_CORE_DECLARED_OUTPUT_HPP
 #define PHLEX_CORE_DECLARED_OUTPUT_HPP
 
-#include "phlex/core/consumer.hpp"
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/model/algorithm_name.hpp"
+#include "phlex/model/identifier.hpp"
 #include "phlex/model/product_store.hpp"
 #include "phlex/phlex_core_export.hpp"
 #include "phlex/utilities/simple_ptr_map.hpp"
@@ -21,18 +21,23 @@ namespace phlex::detail {
   namespace internal {
     using output_function_t = std::function<void(phlex::experimental::product_store const&)>;
   }
-  class PHLEX_CORE_EXPORT declared_output : public consumer {
+  class PHLEX_CORE_EXPORT declared_output {
   public:
     declared_output(phlex::experimental::algorithm_name name,
                     std::size_t concurrency,
                     std::vector<std::string> predicates,
                     tbb::flow::graph& g,
                     internal::output_function_t&& ft);
 
+    phlex::experimental::algorithm_name const& name() const noexcept;
+    std::vector<std::string> const& when() const noexcept;
+
     tbb::flow::receiver<message>& port() noexcept;
     std::size_t num_calls() const { return calls_; }
 
   private:
+    phlex::experimental::algorithm_name name_;
+    std::vector<std::string> predicates_;
     tbb::flow::function_node<message> node_;
     std::atomic<std::size_t> calls_;
   };

phlex/core/declared_predicate.cpp

@@ -1,7 +1,7 @@
 #include "phlex/core/declared_predicate.hpp"
 
+#include "phlex/core/consumer.hpp"
 #include "phlex/core/product_selector.hpp"
-#include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 
 #include <oneapi/tbb/flow_graph.h>
@@ -15,11 +15,11 @@ namespace phlex::detail {
                                          std::vector<std::string> predicates,
                                          product_selectors input_products,
                                          tbb::flow::graph& graph) :
-    products_consumer{std::move(name),
-                      std::move(predicates),
-                      std::move(input_products),
-                      graph,
-                      require_layers::multi_input_only}
+    consumer{std::move(name),
+             std::move(predicates),
+             std::move(input_products),
+             graph,
+             require_layers::multi_input_only}
   {
   }