created API for passive work-flow

3 files changed, 21 insertions, 0 deletions

diff --git a/lib/block.cpp b/lib/block.cpp
index ee7142c..039daf3 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -167,3 +167,13 @@ void Block::mark_done(void)
 {
     (*this)->block->mark_done();
 }
+
+const SBuffer &Block::get_input_buffer(const size_t which_input)
+{
+    return (*this)->block->input_queues.front(which_input);
+}
+
+void Block::post_output_buffer(const size_t which_output, const SBuffer &buffer)
+{
+    (*this)->block->produce_buffer(which_output, buffer);
+}
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 5e1db89..d3e12f6 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -233,3 +233,12 @@ void BlockActor::produce(const size_t i, const size_t items)
     const size_t bytes = items*this->output_items_sizes[i];
     buff.length += bytes;
 }
+
+void BlockActor::produce_buffer(const size_t i, const SBuffer &buffer)
+{
+    const size_t items = buffer.length/output_items_sizes[i];
+    this->items_produced[i] += items;
+    InputBufferMessage buff_msg;
+    buff_msg.buffer = buffer;
+    this->post_downstream(i, buff_msg);
+}
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index 42a05a7..6e2903b 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -92,6 +92,8 @@ struct BlockActor : Apology::Worker
     void trim_tags(const size_t index);
     void produce(const size_t index, const size_t items);
     void consume(const size_t index, const size_t items);
+    void produce_buffer(const size_t index, const SBuffer &buffer);
+
     GRAS_FORCE_INLINE bool any_inputs_done(void)
     {
         if (this->inputs_done.none() or this->input_queues.all_ready()) return false;