gras: moved flush output to consume function in output queues

This cleans up some code in block actor.
The message sending code is now in task_main,
and the consume routine is now the shared one.

1 files changed, 14 insertions, 2 deletions

diff --git a/lib/task_main.cpp b/lib/task_main.cpp
index 9ab786a..6c3c23d 100644
--- a/lib/task_main.cpp
+++ b/lib/task_main.cpp
@@ -52,7 +52,7 @@ void BlockActor::task_main(void)
             //copy buffer reference but push with zero length, same offset
             SBuffer new_obuff = buff;
             new_obuff.length = 0;
-            this->flush_output(output_inline_index);
+            this->output_queues.consume(output_inline_index);
             this->output_queues.push(output_inline_index, new_obuff); //you got inlined!
             output_inline_index++; //done do this output port again
         }
@@ -103,7 +103,19 @@ void BlockActor::task_main(void)
     //------------------------------------------------------------------
     for (size_t i = 0; i < num_outputs; i++)
     {
-        this->flush_output(i);
+        //buffer may be popped by one of the special buffer api hooks
+        if GRAS_UNLIKELY(this->output_queues.empty(i)) continue;
+
+        //grab a copy of the front buffer then consume from the queue
+        InputBufferMessage buff_msg;
+        buff_msg.buffer = this->output_queues.front(i);
+        this->output_queues.consume(i);
+
+        //Post a buffer message downstream only if the produce flag was marked.
+        //So this explicitly after consuming the output queues so pop is called.
+        //This is because pop may have special hooks in it to prepare the buffer.
+        if GRAS_LIKELY(this->produce_outputs[i]) this->post_downstream(i, buff_msg);
+        this->produce_outputs[i] = false;
     }
 
     //------------------------------------------------------------------