Adding PDU to tagged stream and tagged stream to PDU blocks along with QA python

non-stream-connected blocks still need a new thread context

22 files changed, 873 insertions, 33 deletions

diff --git a/gnuradio-core/src/lib/io/CMakeLists.txt b/gnuradio-core/src/lib/io/CMakeLists.txt
index 3dea13396..7041f2820 100644
--- a/gnuradio-core/src/lib/io/CMakeLists.txt
+++ b/gnuradio-core/src/lib/io/CMakeLists.txt
@@ -38,6 +38,7 @@ list(APPEND gnuradio_core_sources
     ${CMAKE_CURRENT_SOURCE_DIR}/microtune_xxxx.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/ppio_ppdev.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gri_wavfile.cc
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_pdu.cc
 )
 
 ########################################################################
@@ -59,6 +60,7 @@ install(FILES
     ${CMAKE_CURRENT_SOURCE_DIR}/microtune_xxxx.h
     ${CMAKE_CURRENT_SOURCE_DIR}/ppio_ppdev.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gri_wavfile.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_pdu.h
     DESTINATION ${GR_INCLUDE_DIR}/gnuradio
     COMPONENT "core_devel"
 )
@@ -85,9 +87,11 @@ set(gr_core_io_triple_threats
     gr_file_source
     gr_file_descriptor_sink
     gr_file_descriptor_source
+    gr_message_debug
     gr_message_sink
     gr_message_source
     gr_message_burst_source
+    gr_pdu_to_tagged_stream
     microtune_xxxx_eval_board
     microtune_4702_eval_board
     microtune_4937_eval_board
@@ -98,6 +102,7 @@ set(gr_core_io_triple_threats
     gr_wavfile_source
     gr_wavfile_sink
     gr_tagged_file_sink
+    gr_tagged_stream_to_pdu
 )
 
 foreach(file_tt ${gr_core_io_triple_threats})
diff --git a/gnuradio-core/src/lib/io/gr_message_debug.cc b/gnuradio-core/src/lib/io/gr_message_debug.cc
new file mode 100644
index 000000000..84c11c46e
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.cc
@@ -0,0 +1,71 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005,2010 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_message_debug.h>
+#include <gr_io_signature.h>
+#include <cstdio>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdexcept>
+#include <string.h>
+#include <iostream>
+
+// public constructor that returns a shared_ptr
+
+gr_message_debug_sptr
+gr_make_message_debug ()
+{
+  return gnuradio::get_initial_sptr(new gr_message_debug());
+}
+
+void gr_message_debug::print(pmt::pmt_t msg){
+    std::cout << "******* DEBUG PRINT ********\n";
+    pmt::pmt_print(msg);
+}
+
+
+gr_message_debug::gr_message_debug ()
+  : gr_sync_block("message_debug",
+		  gr_make_io_signature(0, 0, 0),
+		  gr_make_io_signature(0, 0, 0))
+{
+    message_port_register_in(pmt::mp("print"));
+    set_msg_handler(pmt::mp("print"), boost::bind(&gr_message_debug::print, this, _1));
+}
+
+gr_message_debug::~gr_message_debug()
+{
+}
+
+int
+gr_message_debug::work(int noutput_items,
+		      gr_vector_const_void_star &input_items,
+		      gr_vector_void_star &output_items)
+{
+  return 0; // FIXME: replace with default NOP work function in gr_block
+}
diff --git a/gnuradio-core/src/lib/io/gr_message_debug.h b/gnuradio-core/src/lib/io/gr_message_debug.h
new file mode 100644
index 000000000..37d8b7a4d
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.h
@@ -0,0 +1,59 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_GR_MESSAGE_DEBUG_H
+#define INCLUDED_GR_MESSAGE_DEBUG_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+
+class gr_message_debug;
+typedef boost::shared_ptr<gr_message_debug> gr_message_debug_sptr;
+
+GR_CORE_API gr_message_debug_sptr gr_make_message_debug ();
+
+/*!
+ * \brief Gather received items into messages and insert into msgq
+ * \ingroup sink_blk
+ */
+class GR_CORE_API gr_message_debug : public gr_sync_block
+{
+ private:
+  friend GR_CORE_API gr_message_debug_sptr
+  gr_make_message_debug();
+
+  void print(pmt::pmt_t msg);
+
+ protected:
+  gr_message_debug ();
+
+ public:
+  ~gr_message_debug ();
+
+  int work (int noutput_items,
+	    gr_vector_const_void_star &input_items,
+	    gr_vector_void_star &output_items);
+};
+
+#endif /* INCLUDED_GR_MESSAGE_DEBUG_H */
diff --git a/gnuradio-core/src/lib/io/gr_message_debug.i b/gnuradio-core/src/lib/io/gr_message_debug.i
new file mode 100644
index 000000000..7bd547c91
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.i
@@ -0,0 +1,30 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+GR_SWIG_BLOCK_MAGIC(gr,message_debug);
+
+%{
+#include <gr_message_debug.h>
+%}
+
+%include "gr_message_debug.h"
+
diff --git a/gnuradio-core/src/lib/io/gr_pdu.cc b/gnuradio-core/src/lib/io/gr_pdu.cc
new file mode 100644
index 000000000..c5290b610
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu.cc
@@ -0,0 +1,43 @@
+
+#include <gr_pdu.h>
+
+size_t gr_pdu_itemsize(gr_pdu_vector_type type){
+    switch(type){
+        case BYTE:
+            return 1;
+        case FLOAT:
+            return sizeof(float);
+        case COMPLEX:
+            return sizeof(gr_complex);
+        default:
+            throw std::runtime_error("bad type!");
+    }
+}
+
+bool gr_pdu_type_matches(gr_pdu_vector_type type, pmt::pmt_t v){
+    switch(type){
+        case BYTE:
+            return pmt::pmt_is_u8vector(v);
+        case FLOAT:
+            return pmt::pmt_is_f32vector(v);
+        case COMPLEX:
+            return pmt::pmt_is_c32vector(v);
+        default:
+            throw std::runtime_error("bad type!");
+    }
+}
+
+pmt::pmt_t gr_pdu_make_vector(gr_pdu_vector_type type, const uint8_t* buf, size_t items){
+    switch(type){
+        case BYTE:
+            return pmt::pmt_init_u8vector(items, buf);
+        case FLOAT:
+            return pmt::pmt_init_f32vector(items, (const float*)buf);
+        case COMPLEX:
+            return pmt::pmt_init_c32vector(items, (const gr_complex*)buf);
+        default:
+            throw std::runtime_error("bad type!");
+    }
+}
+
+
diff --git a/gnuradio-core/src/lib/io/gr_pdu.h b/gnuradio-core/src/lib/io/gr_pdu.h
new file mode 100644
index 000000000..3bea83740
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu.h
@@ -0,0 +1,38 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef GR_PDU_H
+#define GR_PDU_H
+
+#include <gr_complex.h>
+#include <gruel/pmt.h>
+
+#define pdu_port_id     pmt::mp("pdus")
+#define pdu_length_tag  pmt::mp("pdu_length")
+
+enum gr_pdu_vector_type { BYTE, FLOAT, COMPLEX };
+
+size_t gr_pdu_itemsize(gr_pdu_vector_type type);
+bool gr_pdu_type_matches(gr_pdu_vector_type type, pmt::pmt_t v);
+pmt::pmt_t gr_pdu_make_vector(gr_pdu_vector_type type, const uint8_t* buf, size_t items);
+
+#endif
diff --git a/gnuradio-core/src/lib/io/gr_pdu.i b/gnuradio-core/src/lib/io/gr_pdu.i
new file mode 100644
index 000000000..7cb3c62c7
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu.i
@@ -0,0 +1,30 @@
+/* -*- c++ -*- */

+/*

+ * Copyright 2005 Free Software Foundation, Inc.

+ *

+ * This file is part of GNU Radio

+ *

+ * GNU Radio is free software; you can redistribute it and/or modify

+ * it under the terms of the GNU General Public License as published by

+ * the Free Software Foundation; either version 3, or (at your option)

+ * any later version.

+ *

+ * GNU Radio is distributed in the hope that it will be useful,

+ * but WITHOUT ANY WARRANTY; without even the implied warranty of

+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

+ * GNU General Public License for more details.

+ *

+ * You should have received a copy of the GNU General Public License

+ * along with GNU Radio; see the file COPYING.  If not, write to

+ * the Free Software Foundation, Inc., 51 Franklin Street,

+ * Boston, MA 02110-1301, USA.

+ */

+

+%{

+#include <gr_pdu.h>

+%}

+

+enum gr_pdu_vector_type { BYTE, FLOAT, COMPLEX };

+

+

+

diff --git a/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.cc b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.cc
new file mode 100644
index 000000000..26c1babd6
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.cc
@@ -0,0 +1,131 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005,2010 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_pdu_to_tagged_stream.h>
+#include <gr_io_signature.h>
+#include <cstdio>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdexcept>
+#include <string.h>
+
+
+// public constructor that returns a shared_ptr
+
+gr_pdu_to_tagged_stream_sptr
+gr_make_pdu_to_tagged_stream(gr_pdu_vector_type t)
+{
+  return gnuradio::get_initial_sptr(new gr_pdu_to_tagged_stream(t));
+}
+
+gr_pdu_to_tagged_stream::gr_pdu_to_tagged_stream (gr_pdu_vector_type t)
+  : gr_sync_block("pdu_to_tagged_stream",
+		  gr_make_io_signature(0, 0, 0),
+		  gr_make_io_signature(1, 1, gr_pdu_itemsize(t))),
+    d_vectortype(t), d_itemsize(gr_pdu_itemsize(t))
+{
+    message_port_register_in(pdu_port_id);
+}
+
+gr_pdu_to_tagged_stream::~gr_pdu_to_tagged_stream()
+{
+}
+
+int
+gr_pdu_to_tagged_stream::work(int noutput_items,
+			gr_vector_const_void_star &input_items,
+			gr_vector_void_star &output_items)
+{   
+  char *out = (char *) output_items[0];
+  int nout = 0;
+
+  // if we have remaining output, send it
+  if(d_remain.size() > 0){
+    nout = std::min((size_t)d_remain.size()/d_itemsize, (size_t)noutput_items);
+    memcpy(out, &d_remain[0], nout*d_itemsize);
+    d_remain.erase( d_remain.begin(), d_remain.begin()+nout);
+    noutput_items -= nout;
+    out += nout*d_itemsize;
+    }
+
+  // if we have space for at least one item output as much as we can
+  if(noutput_items > 0){
+
+    // grab a message if one exists
+    pmt::pmt_t msg( delete_head_nowait( pdu_port_id ) );
+    if(msg.get() == NULL ){
+        return nout;
+        }
+
+    // make sure type is valid
+    if(!pmt::pmt_is_pair(msg)){
+        throw std::runtime_error("received a malformed pdu message!");
+        }
+   
+    printf("got a msg\n");
+    pmt::pmt_print(msg);
+ 
+    // grab the components of the pdu message
+    pmt::pmt_t meta(pmt::pmt_car(msg)); // make sure this is NIL || Dict ?
+    pmt::pmt_t vect(pmt::pmt_cdr(msg)); // make sure this is a vector?
+
+    // compute offset for output tag
+    uint64_t offset = nitems_written(0) + nout;
+
+    // add a tag for pdu length
+    add_item_tag(0, offset, pdu_length_tag, pmt::pmt_from_long( pmt::pmt_length(vect) ), pmt::mp(alias()));
+
+    // if we recieved metadata add it as tags
+    if( !pmt_eq(meta, pmt::PMT_NIL) ){
+        pmt::pmt_t pair(pmt::pmt_dict_keys( meta ));
+        while( !pmt_eq(pair, pmt::PMT_NIL) ){
+            pmt::pmt_t k(pmt::pmt_cdr(pair));
+            pmt::pmt_t v(pmt::pmt_dict_ref(meta, k, pmt::PMT_NIL));
+            add_item_tag(0, offset, k, v, pmt::mp(alias()));
+            }
+        }
+
+    // copy vector output
+    size_t ncopy = std::min((size_t)noutput_items, (size_t)pmt::pmt_length(vect));
+    size_t nsave = pmt::pmt_length(vect) - ncopy;
+ 
+    // copy output
+    size_t io(0);
+    nout += ncopy;
+    memcpy(out, pmt_uniform_vector_elements(vect,io), ncopy*d_itemsize);
+
+    // save leftover items if needed for next work call
+    if(nsave > 0){
+        d_remain.resize(nsave*d_itemsize, 0);
+        memcpy(&d_remain[0], pmt_uniform_vector_elements(vect,ncopy), nsave*d_itemsize);
+        }
+    
+  }
+
+  return nout;
+}
diff --git a/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.h b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.h
new file mode 100644
index 000000000..3eb2a078b
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.h
@@ -0,0 +1,63 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_GR_PDU_TO_TAGGED_STREAM_H
+#define INCLUDED_GR_PDU_TO_TAGGED_STREAM_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+#include <gr_pdu.h>
+
+class gr_pdu_to_tagged_stream;
+typedef boost::shared_ptr<gr_pdu_to_tagged_stream> gr_pdu_to_tagged_stream_sptr;
+  
+GR_CORE_API gr_pdu_to_tagged_stream_sptr gr_make_pdu_to_tagged_stream (gr_pdu_vector_type t);
+
+/*!
+ * \brief Turn received messages into a stream
+ * \ingroup source_blk
+ */
+class GR_CORE_API gr_pdu_to_tagged_stream : public gr_sync_block
+{
+ private:
+  gr_pdu_vector_type    d_vectortype;
+  size_t	 	        d_itemsize;
+  std::vector<uint8_t>  d_remain;
+
+  friend GR_CORE_API gr_pdu_to_tagged_stream_sptr
+  gr_make_pdu_to_tagged_stream(gr_pdu_vector_type t);
+
+ protected:
+  gr_pdu_to_tagged_stream (gr_pdu_vector_type t);
+
+ public:
+  ~gr_pdu_to_tagged_stream ();
+
+  int work (int noutput_items,
+	    gr_vector_const_void_star &input_items,
+	    gr_vector_void_star &output_items);
+
+};
+
+#endif /* INCLUDED_GR_PDU_TO_TAGGED_SOURCE_H */
diff --git a/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.i b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.i
new file mode 100644
index 000000000..ddf1d8c05
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.i
@@ -0,0 +1,31 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+GR_SWIG_BLOCK_MAGIC(gr,pdu_to_tagged_stream);
+
+%{
+#include <gr_pdu_to_tagged_stream.h>
+%}
+
+%include <gr_pdu_to_tagged_stream.h>
+
+
diff --git a/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.cc b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.cc
new file mode 100644
index 000000000..05ed8adc0
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.cc
@@ -0,0 +1,139 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005,2010 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_tagged_stream_to_pdu.h>
+#include <gr_io_signature.h>
+#include <cstdio>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdexcept>
+#include <string.h>
+
+
+
+// public constructor that returns a shared_ptr
+
+gr_tagged_stream_to_pdu_sptr
+gr_make_tagged_stream_to_pdu(gr_pdu_vector_type t)
+{
+  return gnuradio::get_initial_sptr(new gr_tagged_stream_to_pdu(t));
+}
+
+gr_tagged_stream_to_pdu::gr_tagged_stream_to_pdu (gr_pdu_vector_type t)
+  : gr_sync_block("tagged_stream_to_pdu",
+		  gr_make_io_signature(1, 1, gr_pdu_itemsize(t)),
+		  gr_make_io_signature(0, 0, 0)),
+    d_vectortype(t), d_itemsize(gr_pdu_itemsize(t)), d_inpdu(false),
+    d_pdu_meta(pmt::PMT_NIL), d_pdu_vector(pmt::PMT_NIL)
+{
+    message_port_register_out(pdu_port_id);
+}
+
+gr_tagged_stream_to_pdu::~gr_tagged_stream_to_pdu()
+{
+    printf("destructor running\n");
+}
+
+int
+gr_tagged_stream_to_pdu::work(int noutput_items,
+			gr_vector_const_void_star &input_items,
+			gr_vector_void_star &output_items)
+{
+  const uint8_t *in = (const uint8_t*) input_items[0];
+  uint64_t abs_N = nitems_read(0);
+
+  // if we are not in a pdu already, start a new one 
+  if(!d_inpdu){
+    get_tags_in_range(d_tags, 0, abs_N, abs_N+1);
+    bool found_length_tag(false);
+    for(d_tags_itr = d_tags.begin(); (d_tags_itr != d_tags.end()) && (!found_length_tag); d_tags_itr++){
+      if( pmt::pmt_equal( (*d_tags_itr).key, pdu_length_tag ) ){
+          if( (*d_tags_itr).offset != abs_N ){
+              throw std::runtime_error("expected next pdu length tag on a different item...");
+              }
+              found_length_tag = true;
+              d_pdu_length = pmt::pmt_to_long( (*d_tags_itr).value );
+              d_pdu_remain = d_pdu_length;
+              d_pdu_meta = pmt::pmt_make_dict();
+              break;
+          } // if have length tag
+      } // iter over tags
+    if(!found_length_tag){
+        throw std::runtime_error("tagged stream does not contain a pdu_length tag!");
+    }
+  }
+
+  size_t ncopy = std::min((size_t)noutput_items, d_pdu_remain);
+
+  // copy any tags in this range into our meta object
+  get_tags_in_range(d_tags, 0, abs_N, abs_N+ncopy);
+  for(d_tags_itr = d_tags.begin(); d_tags_itr != d_tags.end(); d_tags_itr++){
+    if( ! pmt_equal( (*d_tags_itr).key, pdu_length_tag ) ){
+        d_pdu_meta = pmt_dict_add(d_pdu_meta, (*d_tags_itr).key, (*d_tags_itr).value);
+        }
+    }
+
+  // copy samples for this vector into either a pmt or our save buffer
+  if(ncopy == d_pdu_remain){ // we will send this pdu
+    if(d_save.size() == 0){
+        d_pdu_vector = gr_pdu_make_vector(d_vectortype, in, ncopy);
+        send_message();
+      } else {
+        size_t oldsize = d_save.size();
+        d_save.resize((oldsize + ncopy)*d_itemsize, 0);
+        memcpy( &d_save[oldsize*d_itemsize], in, ncopy*d_itemsize );
+        d_pdu_vector = gr_pdu_make_vector(d_vectortype, &d_save[0], d_pdu_length);
+        send_message();
+        d_save.clear();
+        }
+    } else {
+        d_inpdu = true;
+        size_t oldsize = d_save.size();
+        d_save.resize( (oldsize+ncopy)*d_itemsize );
+        memcpy( &d_save[oldsize*d_itemsize], in, ncopy*d_itemsize );
+        d_pdu_remain -= ncopy;
+    }
+
+  return ncopy;
+}
+
+void gr_tagged_stream_to_pdu::send_message(){
+
+    if(pmt::pmt_length(d_pdu_vector) != d_pdu_length){
+        throw std::runtime_error("msg length not correct");
+    }
+
+    pmt::pmt_t msg = pmt::pmt_cons( d_pdu_meta, d_pdu_vector );
+    message_port_pub( pdu_port_id, msg );
+
+    d_pdu_meta = pmt::PMT_NIL;
+    d_pdu_vector = pmt::PMT_NIL;
+    d_pdu_length = 0;
+    d_pdu_remain = 0;
+    d_inpdu = false;
+}
diff --git a/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.h b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.h
new file mode 100644
index 000000000..6a40e1c0c
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.h
@@ -0,0 +1,76 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef INCLUDED_GR_TAGGED_STREAM_TO_PDU_H
+#define INCLUDED_GR_TAGGED_STREAM_TO_PDU_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+#include <gr_pdu.h>
+
+class gr_tagged_stream_to_pdu;
+typedef boost::shared_ptr<gr_tagged_stream_to_pdu> gr_tagged_stream_to_pdu_sptr;
+  
+GR_CORE_API gr_tagged_stream_to_pdu_sptr gr_make_tagged_stream_to_pdu (gr_pdu_vector_type t);
+
+/*!
+ * \brief Turn received messages into a stream
+ * \ingroup source_blk
+ */
+class GR_CORE_API gr_tagged_stream_to_pdu : public gr_sync_block
+{
+ private:
+  gr_pdu_vector_type    d_vectortype;
+  size_t	 	        d_itemsize;
+
+  std::vector<uint8_t>  d_save;
+
+  std::vector<gr_tag_t> d_tags;
+  std::vector<gr_tag_t>::iterator d_tags_itr;
+
+  bool d_inpdu;
+
+  size_t                d_pdu_length;
+  size_t                d_pdu_remain;
+  pmt::pmt_t            d_pdu_meta;
+  pmt::pmt_t            d_pdu_vector;
+
+  friend GR_CORE_API gr_tagged_stream_to_pdu_sptr
+  gr_make_tagged_stream_to_pdu(gr_pdu_vector_type t);
+
+ protected:
+  gr_tagged_stream_to_pdu (gr_pdu_vector_type t);
+
+ public:
+  ~gr_tagged_stream_to_pdu ();
+
+  int work (int noutput_items,
+	    gr_vector_const_void_star &input_items,
+	    gr_vector_void_star &output_items);
+
+  void send_message();
+
+};
+
+#endif /* INCLUDED_GR_PDU_TO_TAGGED_SOURCE_H */
diff --git a/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.i b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.i
new file mode 100644
index 000000000..66bd875aa
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.i
@@ -0,0 +1,31 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 Free Software Foundation, Inc.
+ *
+ * This file is part of GNU Radio
+ *
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ *
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Radio; see the file COPYING.  If not, write to
+ * the Free Software Foundation, Inc., 51 Franklin Street,
+ * Boston, MA 02110-1301, USA.
+ */
+
+GR_SWIG_BLOCK_MAGIC(gr,tagged_stream_to_pdu);
+
+%{
+#include <gr_tagged_stream_to_pdu.h>
+%}
+
+%include <gr_tagged_stream_to_pdu.h>
+
+
diff --git a/gnuradio-core/src/lib/io/io.i b/gnuradio-core/src/lib/io/io.i
index 5cd352905..871ce1356 100644
--- a/gnuradio-core/src/lib/io/io.i
+++ b/gnuradio-core/src/lib/io/io.i
@@ -45,6 +45,10 @@
 #include <gr_wavfile_sink.h>
 #include <gr_wavfile_source.h>
 #include <gr_tagged_file_sink.h>
+#include <gr_pdu_to_tagged_stream.h>
+#include <gr_tagged_stream_to_pdu.h>
+#include <gr_message_debug.h>
+#include <gr_pdu.h>
 %}
 
 %include "gr_file_sink_base.i"
@@ -67,4 +71,8 @@
 %include "gr_wavfile_sink.i"
 %include "gr_wavfile_source.i"
 %include "gr_tagged_file_sink.i"
+%include "gr_pdu_to_tagged_stream.i"
+%include "gr_tagged_stream_to_pdu.i"
+%include "gr_message_debug.i"
+%include "gr_pdu.i"
 
diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.cc b/gnuradio-core/src/lib/runtime/gr_basic_block.cc
index 3d08b63d1..7d2f275e8 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.cc
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.cc
@@ -52,6 +52,7 @@ gr_basic_block::gr_basic_block(const std::string &name,
     d_color(WHITE),
     message_subscribers(pmt::pmt_make_dict())
 {
+    mutex.unlock();
     s_ncurrently_allocated++;
 }
 
@@ -130,7 +131,7 @@ void
 gr_basic_block::_post(pmt_t which_port, pmt_t msg)
 {
   insert_tail(which_port, msg);
-  //notify_msg();
+  global_block_registry.notify_blk(alias());
 }
 
 void
@@ -151,8 +152,9 @@ gr_basic_block::delete_head_nowait(pmt::pmt_t which_port)
 {
   gruel::scoped_lock guard(mutex);
 
-  if (empty_p(which_port))
-    return pmt_t();
+  if (empty_p(which_port)){
+    return pmt::pmt_t();
+    }
 
   pmt_t m(msg_queue[which_port].front());
   msg_queue[which_port].pop_front();
@@ -160,18 +162,4 @@ gr_basic_block::delete_head_nowait(pmt::pmt_t which_port)
   return m;
 }
 
-/*
- * Caller must already be holding the mutex
- */
-pmt_t
-gr_basic_block::delete_head_nowait_already_holding_mutex(pmt::pmt_t which_port)
-{
-  if (empty_p(which_port))
-    return pmt_t();
-
-  pmt_t m(msg_queue[which_port].front());
-  msg_queue[which_port].pop_front();
-
-  return m;
-}
 
diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.h b/gnuradio-core/src/lib/runtime/gr_basic_block.h
index 31026a2e4..2ee8161c1 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.h
@@ -115,7 +115,6 @@ protected:
     vcolor color() const { return d_color; }
 
     // Message passing interface
-    std::vector<pmt::pmt_t> message_inputs;
     pmt::pmt_t message_subscribers;
 
 public:
@@ -139,15 +138,18 @@ public:
     void message_port_sub(pmt::pmt_t port_id, pmt::pmt_t target);
     void message_port_unsub(pmt::pmt_t port_id, pmt::pmt_t target);
 
-  /*!
-   * Accept msg, place in queue, arrange for thread to be awakened if it's not already.
-   */
-  void _post(pmt::pmt_t which_port, pmt::pmt_t msg);
-
+    /*!
+     * Accept msg, place in queue, arrange for thread to be awakened if it's not already.
+     */
+    void _post(pmt::pmt_t which_port, pmt::pmt_t msg);
 
     //! is the queue empty?
     //bool empty_p(const pmt::pmt_t &which_port) const { return msg_queue[which_port].empty(); }
-    bool empty_p(pmt::pmt_t which_port) { return msg_queue[which_port].empty(); }
+    bool empty_p(pmt::pmt_t which_port) { 
+        if(msg_queue.find(which_port) == msg_queue.end())
+            throw std::runtime_error("port does not exist!");
+        return msg_queue[which_port].empty(); 
+        }
     bool empty_p() { 
         bool rv = true;
         BOOST_FOREACH(msg_queue_map_t::value_type &i, msg_queue){ rv &= msg_queue[i.first].empty(); }
@@ -160,11 +162,6 @@ public:
      * \returns returns pmt at head of queue or pmt_t() if empty.
      */
     pmt::pmt_t delete_head_nowait( pmt::pmt_t which_port);
-    /*!
-     * \returns returns pmt at head of queue or pmt_t() if empty.
-     * Caller must already be holding the mutex
-     */
-    pmt::pmt_t delete_head_nowait_already_holding_mutex( pmt::pmt_t which_port);
 
     msg_queue_t::iterator get_iterator(pmt::pmt_t which_port){
         return msg_queue[which_port].begin();
diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.i b/gnuradio-core/src/lib/runtime/gr_basic_block.i
index 0a8473ba2..d6d6c3d16 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.i
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.i
@@ -23,6 +23,8 @@
 class gr_basic_block;
 typedef boost::shared_ptr<gr_basic_block> gr_basic_block_sptr;
 %template(gr_basic_block_sptr) boost::shared_ptr<gr_basic_block>;
+%include "pmt_swig.i"
+using namespace pmt;
 
 // support vectors of these...
 namespace std {
@@ -45,6 +47,7 @@ public:
     bool check_topology (int ninputs, int noutputs);
     std::string alias();
     void set_block_alias(std::string name);
+    void _post(pmt_t which_port, pmt_t msg);
 };
 
 %rename(block_ncurrently_allocated) gr_basic_block_ncurrently_allocated;
diff --git a/gnuradio-core/src/lib/runtime/gr_block.cc b/gnuradio-core/src/lib/runtime/gr_block.cc
index a88052ee0..dc77128a3 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block.cc
@@ -28,6 +28,7 @@
 #include <gr_block_detail.h>
 #include <stdexcept>
 #include <iostream>
+#include <gr_block_registry.h>
 
 gr_block::gr_block (const std::string &name,
 		    gr_io_signature_sptr input_signature,
@@ -46,10 +47,12 @@ gr_block::gr_block (const std::string &name,
     d_max_output_buffer(std::max(output_signature->max_streams(),1), -1),
     d_min_output_buffer(std::max(output_signature->max_streams(),1), -1)
 {
+    global_block_registry.register_primitive(alias(), this);
 }
 
 gr_block::~gr_block ()
 {
+    global_block_registry.unregister_primitive(alias());
 }
 
 // stub implementation:  1:1
diff --git a/gnuradio-core/src/lib/runtime/gr_block_registry.cc b/gnuradio-core/src/lib/runtime/gr_block_registry.cc
index 2478e0019..ff23d97eb 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_registry.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_registry.cc
@@ -1,5 +1,9 @@
 #include <gr_basic_block.h>
 #include <gr_block_registry.h>
+#include <gr_tpb_detail.h>
+#include <gr_block_detail.h>
+#include <gr_block.h>
+#include <stdio.h>
 
 gr_block_registry global_block_registry;
 
@@ -56,3 +60,17 @@ gr_basic_block_sptr gr_block_registry::block_lookup(pmt::pmt_t symbol){
 }
 
 
+void gr_block_registry::register_primitive(std::string blk, gr_block* ref){
+    primitive_map[blk] = ref;
+}
+
+void gr_block_registry::unregister_primitive(std::string blk){
+    primitive_map.erase(primitive_map.find(blk));
+}
+
+void gr_block_registry::notify_blk(std::string blk){
+    if(primitive_map.find(blk) == primitive_map.end()){ return; }
+    if(primitive_map[blk]->detail().get())
+        primitive_map[blk]->detail()->d_tpb.notify_msg();
+}
+
diff --git a/gnuradio-core/src/lib/runtime/gr_block_registry.h b/gnuradio-core/src/lib/runtime/gr_block_registry.h
index 8f1982984..6a2d939e5 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_registry.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_registry.h
@@ -5,6 +5,7 @@
 
 #ifndef GR_BASIC_BLOCK_H
 class gr_basic_block;
+class gr_block;
 #endif
 
 class gr_block_registry {
@@ -18,7 +19,11 @@ class gr_block_registry {
         void register_symbolic_name(gr_basic_block* block, std::string name);
 
         gr_basic_block_sptr block_lookup(pmt::pmt_t symbol);
-    
+ 
+        void register_primitive(std::string blk, gr_block* ref);
+        void unregister_primitive(std::string blk);
+        void notify_blk(std::string blk);
+
     private:
      
         //typedef std::map< long, gr_basic_block_sptr >   blocksubmap_t;
@@ -27,7 +32,8 @@ class gr_block_registry {
 
         blockmap_t d_map;
         pmt::pmt_t d_ref_map;
-
+        std::map< std::string, gr_block*> primitive_map;
+ 
 };
 
 extern gr_block_registry global_block_registry;
diff --git a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
index 1bd3014ad..ff2afca10 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
@@ -80,7 +80,7 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
 	  // handle all pending messages
       BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
       {
-	    while ((msg = block->delete_head_nowait_already_holding_mutex(i.first))){
+	    while ((msg = block->delete_head_nowait(i.first))){
 	      guard.unlock();			// release lock while processing msg
 	      block->dispatch_msg(i.first, msg);
 	      guard.lock();
@@ -103,7 +103,7 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
 	  // handle all pending messages
       BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
       {
-	    while ((msg = block->delete_head_nowait_already_holding_mutex(i.first))){
+	    while ((msg = block->delete_head_nowait(i.first))){
 	      guard.unlock();			// release lock while processing msg
 	      block->dispatch_msg(i.first,msg);
 	      guard.lock();
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
new file mode 100755
index 000000000..64eb80a8f
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
@@ -0,0 +1,70 @@
+#!/usr/bin/env python
+#
+# Copyright 2012 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# GNU Radio is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+from gnuradio import gr, gr_unittest
+from gruel import pmt
+import time;
+class test_pdu(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gr.top_block()
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_000(self):
+        # Just run some data through and make sure it doesn't puke.
+        src_data = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+
+        src = gr.pdu_to_tagged_stream(gr.BYTE);
+        snk3 = gr.tagged_stream_to_pdu(gr.BYTE);
+        snk2 = gr.vector_sink_b();
+        snk = gr.tag_debug(1, "test");
+
+        dbg = gr.message_debug();
+
+        self.tb.connect(src, snk)
+        self.tb.connect(src, snk2)
+        self.tb.connect(src, snk3)
+
+        self.tb.msg_connect(snk3, "pdus", dbg, "print");
+        self.tb.start()
+
+        # make our reference and message pmts
+        port = pmt.pmt_intern("pdus");
+        msg = pmt.pmt_cons( pmt.PMT_NIL, pmt.pmt_make_u8vector(16, 0xFF) );
+        
+        print "printing port & msg"
+        pmt.pmt_print(port);
+        pmt.pmt_print(msg);
+
+        # post the message
+        src.to_basic_block()._post( port, msg );
+
+        time.sleep(1);
+        self.tb.stop();
+        self.tb.wait();
+
+        print snk2.data();
+
+if __name__ == '__main__':
+    gr_unittest.run(test_pdu, "test_pdu.xml")