Basic pushing of data through inputs/outputs.

app/playground/main.cpp

@@ -5,9 +5,15 @@
 
 #include <pls/pls.h>
 #include <pls/dataflow/graph.h>
+#include <pls/dataflow/internal/token.h>
 
 int main() {
   using namespace pls::dataflow;
 
-  graph<inputs<int>, outputs<int, int>, 8> graph;
+  graph<inputs<int, int>, outputs<int, int>, 8> graph;
+  graph.output<0>().connect(graph.input<0>());
+  graph.output<1>().connect(graph.input<1>());
+
+  graph.input<0>().push_token(pls::dataflow::internal::token<int>());
+  graph.input<1>().push_token(pls::dataflow::internal::token<int>());
 }

lib/pls/include/pls/dataflow/dataflow.h

@@ -2,4 +2,8 @@
 #ifndef PLS_DATAFLOW_DATAFLOW_H_
 #define PLS_DATAFLOW_DATAFLOW_H_
 
+#include "graph.h"
+#include "inputs.h"
+#include "outputs.h"
+
 #endif //PLS_DATAFLOW_DATAFLOW_H_

lib/pls/include/pls/dataflow/graph.h

@@ -12,7 +12,22 @@ template<typename I, typename O, int P>
 class graph {
   using internal_inputs = typename I::template internal_inputs<P>;
   using internal_outputs = typename O::template internal_outputs<P>;
+
+ public:
+  internal_inputs inputs_;
+  internal_outputs outputs_;
+
+  template<int N>
+  decltype(outputs_.template get<N>()) output() {
+    return outputs_.template get<N>();
+  }
+
+  template<int N>
+  decltype(inputs_.template get<N>()) input() {
+    return inputs_.template get<N>();
+  }
 };
+
 }
 }
 

lib/pls/include/pls/dataflow/internal/input.h

@@ -2,23 +2,32 @@
 #ifndef PLS_DATAFLOW_INTERNAL_INPUT_H_
 #define PLS_DATAFLOW_INTERNAL_INPUT_H_
 
-#include "token.h"
+#include <array>
 
 #include "pls/internal/base/error_handling.h"
 
+#include "token.h"
+
 namespace pls {
 namespace dataflow {
 namespace internal {
 
-template<typename T>
+class push_token_cb {
+ public:
+  virtual void token_pushed(int pos, token_color color) = 0;
+};
+
+template<int P, typename T>
 class input {
-  template<typename OT>
+  template<int OP, typename OT>
   friend
   class output;
 
-  token<T> token_;
-  bool filled_;
-  bool connected_;
+  std::array<token<T>, P> tokens_;
+  bool connected_{false};
+
+  push_token_cb *cb_{nullptr};
+  int input_pos_{0};
 
   void connect() {
     if (connected_) {
@@ -27,15 +36,16 @@ class input {
     connected_ = true;
   }
 
-  input() : filled_{false}, connected_{false} {};
-
-  bool has_token(token_color color) const {
-    return filled_ && token_.color() == color;
-  }
-  const token<T> &get_token() const { return token_; }
-
  public:
+  void push_token(token<T> token) {
+    tokens_[token.color().get_index(P)] = token;
+    cb_->token_pushed(input_pos_, token.color());
+  }
 
+  void set_cb(push_token_cb *cb, int input_pos) {
+    cb_ = cb;
+    input_pos_ = input_pos;
+  }
 };
 
 }

lib/pls/include/pls/dataflow/internal/inputs.h

@@ -3,6 +3,7 @@
 #define PLS_DATAFLOW_INTERNAL_INPUTS_H_
 
 #include <tuple>
+#include <array>
 
 #include "input.h"
 
@@ -11,15 +12,45 @@ namespace dataflow {
 namespace internal {
 
 template<int P, typename ...I>
-class inputs {
-  using values_type = std::tuple<input<I>...>;
+class inputs : push_token_cb {
+  using values_type = std::tuple<input<P, I>...>;
   values_type values_;
+  static constexpr unsigned int num_values = std::tuple_size<values_type>::value;
+
+  std::array<std::atomic<unsigned int>, P> required_inputs_;
 
  public:
+  inputs() {
+    for (int i = 0; i < P; i++) {
+      required_inputs_[i] = num_values;
+    }
+    init_cb<0, I...>::call(this);
+  }
+
+  void token_pushed(int /*pos*/, token_color color) override {
+    int current_required = --required_inputs_[color.get_index(P)];
+    if (current_required == 0) {
+      std::cout << "All Inputs Avaliable" << std::endl; // TODO: Add proper callback in here
+    }
+  }
+
   template<int N>
   typename std::tuple_element<N, values_type>::type &get() {
     return std::get<N>(values_);
   }
+
+// TODO: Change CB code using proper templating to save method calls during execution...
+  template<int POS, typename ...TAIL>
+  struct init_cb {
+    static void call(inputs<P, I...> *inputs) { }
+  };
+  template<int POS, typename HEAD, typename ...TAIL>
+  struct init_cb<POS, HEAD, TAIL...> {
+    static void call(inputs<P, I...> *inputs) {
+      inputs->get<POS>().set_cb(inputs, POS);
+      init_cb<POS + 1, TAIL...>::call(inputs);
+    }
+  };
 };
 
 }

lib/pls/include/pls/dataflow/internal/output.h

@@ -11,15 +11,15 @@ namespace pls {
 namespace dataflow {
 namespace internal {
 
-template<typename T>
+template<int P, typename T>
 class output {
-  input<T> *target_;
+  input<P, T> *target_;
   bool connected_;
 
  public:
   output() : target_{nullptr}, connected_{false} {};
 
-  void connect(input<T> &target) {
+  void connect(input<P, T> &target) {
     if (connected_) {
       PLS_ERROR("Must only connect output once. Please disconnect it before reconnecting.")
     }
@@ -28,6 +28,10 @@ class output {
     target_ = &target;
     connected_ = true;
   }
+
+  void push_token(token<T> token) {
+    target_->push_token(token);
+  }
 };
 
 }

lib/pls/include/pls/dataflow/internal/outputs.h

@@ -12,7 +12,7 @@ namespace internal {
 
 template<int P, typename ...O>
 class outputs {
-  using values_type = std::tuple<output<O>...>;
+  using values_type = std::tuple<output<P, O>...>;
   values_type values_;
 
  public:

lib/pls/include/pls/dataflow/internal/token.h

@@ -13,6 +13,10 @@ namespace internal {
  */
 struct token_color {
   unsigned int clock_;
+
+  int get_index(int parallel_limit) const {
+    return clock_ % parallel_limit;
+  }
 };
 
 template<typename T>