created hooks for input buffer allocator

3 files changed, 45 insertions, 5 deletions

diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp
index 24369c7..ffd052b 100644
--- a/lib/block_allocator.cpp
+++ b/lib/block_allocator.cpp
@@ -86,15 +86,19 @@ void ElementImpl::handle_allocation(const tsbe::TaskInterface &task_iface)
         );
 
         SBufferDeleter deleter = boost::bind(&ElementImpl::buffer_returner, this, i, _1);
+        SBufferToken token = SBufferToken(new SBufferDeleter(deleter));
 
-        this->output_buffer_tokens[i] = block_ptr->output_buffer_allocator(
-            i, SBufferToken(new SBufferDeleter(deleter)), bytes
-        );
+        this->output_buffer_tokens[i] = block_ptr->output_buffer_allocator(i, token, bytes);
+
+        InputAllocatorMessage message;
+        message.token = token;
+        message.recommend_length = bytes;
+        task_iface.post_downstream(i, message);
     }
 }
 
 SBufferToken Block::output_buffer_allocator(
-    const size_t which_output,
+    const size_t,
     const SBufferToken &token,
     const size_t recommend_length
 ){
@@ -110,3 +114,11 @@ SBufferToken Block::output_buffer_allocator(
     }
     return token;
 }
+
+SBufferToken Block::input_buffer_allocator(
+    const size_t,
+    const SBufferToken &,
+    const size_t
+){
+    return SBufferToken(); //null
+}
diff --git a/lib/gras_impl/messages.hpp b/lib/gras_impl/messages.hpp
index 8c6664f..912e3d1 100644
--- a/lib/gras_impl/messages.hpp
+++ b/lib/gras_impl/messages.hpp
@@ -64,6 +64,12 @@ struct UpdateInputsMessage
     //empty
 };
 
+struct InputAllocatorMessage
+{
+    SBufferToken token;
+    size_t recommend_length;
+};
+
 } //namespace gnuradio
 
 #endif /*INCLUDED_LIBGRAS_IMPL_MESSAGES_HPP*/
diff --git a/lib/port_handlers.cpp b/lib/port_handlers.cpp
index d0d5896..848d9ef 100644
--- a/lib/port_handlers.cpp
+++ b/lib/port_handlers.cpp
@@ -60,6 +60,19 @@ void ElementImpl::handle_input_msg(
         return;
     }
 
+    //handle the upstream block allocation request
+    if (msg.type() == typeid(InputAllocatorMessage))
+    {
+        InputAllocatorMessage message;
+        message.token = block_ptr->input_buffer_allocator(
+            index,
+            msg.cast<InputAllocatorMessage>().token,
+            msg.cast<InputAllocatorMessage>().recommend_length
+        );
+        if (message.token) handle.post_upstream(index, message);
+        return;
+    }
+
     ASSERT(false);
 }
 
@@ -77,7 +90,6 @@ void ElementImpl::handle_output_msg(
         return;
     }
 
-
     //a downstream block has declared itself done, recheck the token
     if (msg.type() == typeid(CheckTokensMessage))
     {
@@ -111,5 +123,15 @@ void ElementImpl::handle_output_msg(
         return;
     }
 
+    //return of a positive downstream allocation
+    //reset the token, and clear old output buffers
+    //the new token from the downstream is installed
+    if (msg.type() == typeid(InputAllocatorMessage))
+    {
+        this->output_buffer_tokens[index].reset();
+        this->output_queues.flush(index);
+        this->output_buffer_tokens[index] = msg.cast<InputAllocatorMessage>().token;
+    }
+
     ASSERT(false);
 }