Merge branch 'master' into file_meta_data

Conflicts:
	gr-utils/src/python/CMakeLists.txt

81 files changed, 4897 insertions, 354 deletions

diff --git a/gnuradio-core/CMakeLists.txt b/gnuradio-core/CMakeLists.txt
index 4e76b3c5a..2f06791ca 100644
--- a/gnuradio-core/CMakeLists.txt
+++ b/gnuradio-core/CMakeLists.txt
@@ -1,4 +1,4 @@
-# Copyright 2010-2011 Free Software Foundation, Inc.
+# Copyright 2010-2012 Free Software Foundation, Inc.
 #
 # This file is part of GNU Radio
 #
@@ -45,6 +45,7 @@ GR_REGISTER_COMPONENT("gnuradio-core" ENABLE_GR_CORE
 
 include(GrMiscUtils)
 GR_SET_GLOBAL(GNURADIO_CORE_INCLUDE_DIRS
+    ${Boost_INCLUDE_DIRS}
     ${GRUEL_INCLUDE_DIRS} #headers depend on gruel
     ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/runtime
     ${CMAKE_CURRENT_BINARY_DIR}/src/lib/general
@@ -60,11 +61,12 @@ GR_SET_GLOBAL(GNURADIO_CORE_INCLUDE_DIRS
     ${CMAKE_CURRENT_BINARY_DIR}/src/lib/swig
     ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/swig
     ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/hier
+    ${Boost_INCLUDE_DIRS}
 )
 
 GR_SET_GLOBAL(GNURADIO_CORE_SWIG_INCLUDE_DIRS
     ${CMAKE_SOURCE_DIR}/gruel/src/swig
-    ${CMAKE_BINARY_DIR}/gruel/src/swig/
+    ${CMAKE_BINARY_DIR}/gruel/src/swig
     ${CMAKE_CURRENT_SOURCE_DIR}/src/lib/swig
     ${GNURADIO_CORE_INCLUDE_DIRS}
 )
diff --git a/gnuradio-core/src/examples/CMakeLists.txt b/gnuradio-core/src/examples/CMakeLists.txt
index c2b847c91..01d9eb83e 100644
--- a/gnuradio-core/src/examples/CMakeLists.txt
+++ b/gnuradio-core/src/examples/CMakeLists.txt
@@ -18,6 +18,7 @@
 # Boston, MA 02110-1301, USA.
 
 add_subdirectory(mp-sched)
+add_subdirectory(msg_passing)
 add_subdirectory(network)
 add_subdirectory(pfb)
 add_subdirectory(tags)
diff --git a/gnuradio-core/src/examples/msg_passing/CMakeLists.txt b/gnuradio-core/src/examples/msg_passing/CMakeLists.txt
new file mode 100644
index 000000000..c4b207a1e
--- /dev/null
+++ b/gnuradio-core/src/examples/msg_passing/CMakeLists.txt
@@ -0,0 +1,27 @@
+# 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.
+
+include(GrPython)
+
+install(
+  FILES
+  strobe.grc
+  DESTINATION ${GR_PKG_DATA_DIR}/examples/msg_passing
+  COMPONENT "core_python"
+)
diff --git a/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier.grc b/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier.grc
new file mode 100644
index 000000000..0faed49bc
--- /dev/null
+++ b/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier.grc
@@ -0,0 +1,287 @@
+<?xml version='1.0' encoding='ASCII'?>
+<flow_graph>
+  <timestamp>Mon Dec 10 19:56:24 2012</timestamp>
+  <block>
+    <key>options</key>
+    <param>
+      <key>id</key>
+      <value>test_msg_hier</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>title</key>
+      <value></value>
+    </param>
+    <param>
+      <key>author</key>
+      <value></value>
+    </param>
+    <param>
+      <key>description</key>
+      <value></value>
+    </param>
+    <param>
+      <key>window_size</key>
+      <value>1280, 1024</value>
+    </param>
+    <param>
+      <key>generate_options</key>
+      <value>hb</value>
+    </param>
+    <param>
+      <key>category</key>
+      <value>Custom</value>
+    </param>
+    <param>
+      <key>run_options</key>
+      <value>prompt</value>
+    </param>
+    <param>
+      <key>run</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>max_nouts</key>
+      <value>0</value>
+    </param>
+    <param>
+      <key>realtime_scheduling</key>
+      <value></value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(10, 10)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>pad_source</key>
+    <param>
+      <key>id</key>
+      <value>pad_source_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>label</key>
+      <value>TEST_PORT</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>message</value>
+    </param>
+    <param>
+      <key>vlen</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>optional</key>
+      <value>False</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(234, 145)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>pad_source</key>
+    <param>
+      <key>id</key>
+      <value>pad_source_0_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>label</key>
+      <value>TEST_PORT2</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>message</value>
+    </param>
+    <param>
+      <key>vlen</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>optional</key>
+      <value>False</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(167, 54)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_cons(pmt.PMT_NIL, pmt.pmt_make_u8vector(16,0x77))</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>200</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(362, 81)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>pad_sink</key>
+    <param>
+      <key>id</key>
+      <value>pad_sink_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>label</key>
+      <value>TEST_PORT</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>message</value>
+    </param>
+    <param>
+      <key>vlen</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>optional</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(618, 87)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0_1</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_intern("OUTPUT2")</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>100</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(400, 156)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>pad_sink</key>
+    <param>
+      <key>id</key>
+      <value>pad_sink_0_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>label</key>
+      <value>TEST_PORT3</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>message</value>
+    </param>
+    <param>
+      <key>vlen</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>optional</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(695, 172)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <connection>
+    <source_block_id>pad_source_0_0</source_block_id>
+    <sink_block_id>gr_message_strobe_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>pad_source_0</source_block_id>
+    <sink_block_id>gr_message_strobe_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0</source_block_id>
+    <sink_block_id>pad_sink_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0_1</source_block_id>
+    <sink_block_id>pad_sink_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0_1</source_block_id>
+    <sink_block_id>pad_sink_0_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+</flow_graph>
diff --git a/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier_topblock.grc b/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier_topblock.grc
new file mode 100644
index 000000000..f440b06b3
--- /dev/null
+++ b/gnuradio-core/src/examples/msg_passing/hier/test_msg_hier_topblock.grc
@@ -0,0 +1,185 @@
+<?xml version='1.0' encoding='ASCII'?>
+<flow_graph>
+  <timestamp>Mon Dec 10 19:56:42 2012</timestamp>
+  <block>
+    <key>options</key>
+    <param>
+      <key>id</key>
+      <value>test_msg_hier_topblock</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>title</key>
+      <value></value>
+    </param>
+    <param>
+      <key>author</key>
+      <value></value>
+    </param>
+    <param>
+      <key>description</key>
+      <value></value>
+    </param>
+    <param>
+      <key>window_size</key>
+      <value>1280, 1024</value>
+    </param>
+    <param>
+      <key>generate_options</key>
+      <value>no_gui</value>
+    </param>
+    <param>
+      <key>category</key>
+      <value>Custom</value>
+    </param>
+    <param>
+      <key>run_options</key>
+      <value>prompt</value>
+    </param>
+    <param>
+      <key>run</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>max_nouts</key>
+      <value>0</value>
+    </param>
+    <param>
+      <key>realtime_scheduling</key>
+      <value></value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(10, 10)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_intern("UPDATED2")</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>3000</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(51, 88)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_intern("UPDATED")</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>2000</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(211, 168)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_debug</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_debug_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(758, 71)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>TEST_MSG_HIER</key>
+    <param>
+      <key>id</key>
+      <value>TEST_MSG_HIER_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(402, 52)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <connection>
+    <source_block_id>TEST_MSG_HIER_0</source_block_id>
+    <sink_block_id>gr_message_debug_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0</source_block_id>
+    <sink_block_id>TEST_MSG_HIER_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0_0</source_block_id>
+    <sink_block_id>TEST_MSG_HIER_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>1</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>TEST_MSG_HIER_0</source_block_id>
+    <sink_block_id>gr_message_debug_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>1</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>TEST_MSG_HIER_0</source_block_id>
+    <sink_block_id>gr_message_debug_0</sink_block_id>
+    <source_key>1</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+</flow_graph>
diff --git a/gnuradio-core/src/examples/msg_passing/strobe.grc b/gnuradio-core/src/examples/msg_passing/strobe.grc
new file mode 100644
index 000000000..7e7e8c345
--- /dev/null
+++ b/gnuradio-core/src/examples/msg_passing/strobe.grc
@@ -0,0 +1,266 @@
+<?xml version='1.0' encoding='ASCII'?>
+<flow_graph>
+  <timestamp>Thu Dec  6 11:33:08 2012</timestamp>
+  <block>
+    <key>variable</key>
+    <param>
+      <key>id</key>
+      <value>samp_rate</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>value</key>
+      <value>32000</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(10, 170)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_pdu_to_tagged_stream</key>
+    <param>
+      <key>id</key>
+      <value>gr_pdu_to_tagged_stream_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>byte</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(443, 89)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_debug</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_debug_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(1049, 176)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_tagged_stream_to_pdu</key>
+    <param>
+      <key>id</key>
+      <value>gr_tagged_stream_to_pdu_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>byte</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(870, 89)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_kludge_copy</key>
+    <param>
+      <key>id</key>
+      <value>gr_kludge_copy_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>type</key>
+      <value>byte</value>
+    </param>
+    <param>
+      <key>num_ports</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>vlen</key>
+      <value>1</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(686, 89)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_intern("TEST")</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>1000</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(423, 177)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>gr_message_strobe</key>
+    <param>
+      <key>id</key>
+      <value>gr_message_strobe_0_0</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>msg</key>
+      <value>pmt.pmt_cons( pmt.PMT_NIL, pmt.pmt_make_u8vector(512,0) )</value>
+    </param>
+    <param>
+      <key>period</key>
+      <value>750</value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(99, 85)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <block>
+    <key>options</key>
+    <param>
+      <key>id</key>
+      <value>strobe</value>
+    </param>
+    <param>
+      <key>_enabled</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>title</key>
+      <value></value>
+    </param>
+    <param>
+      <key>author</key>
+      <value></value>
+    </param>
+    <param>
+      <key>description</key>
+      <value></value>
+    </param>
+    <param>
+      <key>window_size</key>
+      <value>1280, 1024</value>
+    </param>
+    <param>
+      <key>generate_options</key>
+      <value>no_gui</value>
+    </param>
+    <param>
+      <key>category</key>
+      <value>Custom</value>
+    </param>
+    <param>
+      <key>run_options</key>
+      <value>prompt</value>
+    </param>
+    <param>
+      <key>run</key>
+      <value>True</value>
+    </param>
+    <param>
+      <key>max_nouts</key>
+      <value>0</value>
+    </param>
+    <param>
+      <key>realtime_scheduling</key>
+      <value></value>
+    </param>
+    <param>
+      <key>_coordinate</key>
+      <value>(10, 10)</value>
+    </param>
+    <param>
+      <key>_rotation</key>
+      <value>0</value>
+    </param>
+  </block>
+  <connection>
+    <source_block_id>gr_message_strobe_0_0</source_block_id>
+    <sink_block_id>gr_pdu_to_tagged_stream_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_message_strobe_0</source_block_id>
+    <sink_block_id>gr_message_debug_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_tagged_stream_to_pdu_0</source_block_id>
+    <sink_block_id>gr_message_debug_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_kludge_copy_0</source_block_id>
+    <sink_block_id>gr_tagged_stream_to_pdu_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+  <connection>
+    <source_block_id>gr_pdu_to_tagged_stream_0</source_block_id>
+    <sink_block_id>gr_kludge_copy_0</sink_block_id>
+    <source_key>0</source_key>
+    <sink_key>0</sink_key>
+  </connection>
+</flow_graph>
diff --git a/gnuradio-core/src/examples/pfb/CMakeLists.txt b/gnuradio-core/src/examples/pfb/CMakeLists.txt
index 6bb25568c..090b2401d 100644
--- a/gnuradio-core/src/examples/pfb/CMakeLists.txt
+++ b/gnuradio-core/src/examples/pfb/CMakeLists.txt
@@ -25,7 +25,6 @@ GR_PYTHON_INSTALL(PROGRAMS
   decimate.py
   fmtest.py
   interpolate.py
-  resampler_demo.grc
   resampler.py
   synth_filter.py
   synth_to_chan.py
@@ -34,3 +33,9 @@ GR_PYTHON_INSTALL(PROGRAMS
   COMPONENT "core_python"
 )
 
+install(
+  FILES
+  resampler_demo.grc
+  DESTINATION ${GR_PKG_DATA_DIR}/examples/pfb
+  COMPONENT "core_python"
+)
diff --git a/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_ccf.cc b/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_ccf.cc
index 06e98447e..ee09fef44 100644
--- a/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_ccf.cc
+++ b/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_ccf.cc
@@ -86,6 +86,7 @@ gr_pfb_arb_resampler_ccf::~gr_pfb_arb_resampler_ccf ()
 {
   for(unsigned int i = 0; i < d_int_rate; i++) {
     delete d_filters[i];
+    delete d_diff_filters[i];
   }
 }
 
diff --git a/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_fff.cc b/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_fff.cc
index 9a9b86992..f01af2e6f 100644
--- a/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_fff.cc
+++ b/gnuradio-core/src/lib/filter/gr_pfb_arb_resampler_fff.cc
@@ -86,6 +86,7 @@ gr_pfb_arb_resampler_fff::~gr_pfb_arb_resampler_fff ()
 {
   for(unsigned int i = 0; i < d_int_rate; i++) {
     delete d_filters[i];
+    delete d_diff_filters[i];
   }
 }
 
diff --git a/gnuradio-core/src/lib/general/CMakeLists.txt b/gnuradio-core/src/lib/general/CMakeLists.txt
index 3cf7f74e4..4c99acfc3 100644
--- a/gnuradio-core/src/lib/general/CMakeLists.txt
+++ b/gnuradio-core/src/lib/general/CMakeLists.txt
@@ -187,6 +187,7 @@ set(gr_core_general_triple_threats
     gr_agc2_ff
     gr_align_on_samplenumbers_ss
     gr_bin_statistics_f
+    gr_block_gateway
     gr_bytes_to_syms
     gr_char_to_float
     gr_char_to_short
@@ -298,6 +299,7 @@ set(gr_core_general_triple_threats
     gr_burst_tagger
     gr_correlate_access_code_tag_bb
     gr_tag_debug
+    gr_message_strobe
 )
 
 foreach(file_tt ${gr_core_general_triple_threats})
diff --git a/gnuradio-core/src/lib/general/general.i b/gnuradio-core/src/lib/general/general.i
index 790549c4d..1446088a2 100644
--- a/gnuradio-core/src/lib/general/general.i
+++ b/gnuradio-core/src/lib/general/general.i
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2004,2005,2006,2007,2008,2009,2010 Free Software Foundation, Inc.
+ * Copyright 2004-2012 Free Software Foundation, Inc.
  *
  * This file is part of GNU Radio
  *
@@ -143,6 +143,7 @@
 #include <gr_add_ff.h>
 #include <gr_vector_map.h>
 #include <gr_tag_debug.h>
+#include <gr_message_strobe.h>
 %}
 
 %include "gri_control_loop.i"
@@ -266,3 +267,5 @@
 %include "gr_add_ff.i"
 %include "gr_vector_map.i"
 %include "gr_tag_debug.i"
+%include "gr_block_gateway.i"
+%include "gr_message_strobe.i"
diff --git a/gnuradio-core/src/lib/general/gr_block_gateway.cc b/gnuradio-core/src/lib/general/gr_block_gateway.cc
new file mode 100644
index 000000000..79b42803a
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_block_gateway.cc
@@ -0,0 +1,184 @@
+/*
+ * Copyright 2011-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.
+ */
+
+#include <gr_block_gateway.h>
+#include <gr_io_signature.h>
+#include <iostream>
+#include <boost/bind.hpp>
+
+/***********************************************************************
+ * Helper routines
+ **********************************************************************/
+template <typename OutType, typename InType>
+void copy_pointers(OutType &out, const InType &in){
+    out.resize(in.size());
+    for (size_t i = 0; i < in.size(); i++){
+        out[i] = (void *)(in[i]);
+    }
+}
+
+/***********************************************************************
+ * The gr_block gateway implementation class
+ **********************************************************************/
+class gr_block_gateway_impl : public gr_block_gateway{
+public:
+    gr_block_gateway_impl(
+        gr_feval_ll *handler,
+        const std::string &name,
+        gr_io_signature_sptr in_sig,
+        gr_io_signature_sptr out_sig,
+        const gr_block_gw_work_type work_type,
+        const unsigned factor
+    ):
+        gr_block(name, in_sig, out_sig),
+        _handler(handler),
+        _work_type(work_type)
+    {
+        switch(_work_type){
+        case GR_BLOCK_GW_WORK_GENERAL:
+            _decim = 1; //not relevant, but set anyway
+            _interp = 1; //not relevant, but set anyway
+            break;
+
+        case GR_BLOCK_GW_WORK_SYNC:
+            _decim = 1;
+            _interp = 1;
+            this->set_fixed_rate(true);
+            break;
+
+        case GR_BLOCK_GW_WORK_DECIM:
+            _decim = factor;
+            _interp = 1;
+            break;
+
+        case GR_BLOCK_GW_WORK_INTERP:
+            _decim = 1;
+            _interp = factor;
+            this->set_output_multiple(_interp);
+            break;
+        }
+    }
+
+    /*******************************************************************
+     * Overloads for various scheduler-called functions
+     ******************************************************************/
+    void forecast(
+        int noutput_items,
+        gr_vector_int &ninput_items_required
+    ){
+        switch(_work_type){
+        case GR_BLOCK_GW_WORK_GENERAL:
+            _message.action = gr_block_gw_message_type::ACTION_FORECAST;
+            _message.forecast_args_noutput_items = noutput_items;
+            _message.forecast_args_ninput_items_required = ninput_items_required;
+            _handler->calleval(0);
+            ninput_items_required = _message.forecast_args_ninput_items_required;
+            return;
+
+        default:
+            unsigned ninputs = ninput_items_required.size();
+            for (unsigned i = 0; i < ninputs; i++)
+                ninput_items_required[i] = fixed_rate_noutput_to_ninput(noutput_items);
+            return;
+        }
+    }
+
+    int general_work(
+        int noutput_items,
+        gr_vector_int &ninput_items,
+        gr_vector_const_void_star &input_items,
+        gr_vector_void_star &output_items
+    ){
+        switch(_work_type){
+        case GR_BLOCK_GW_WORK_GENERAL:
+            _message.action = gr_block_gw_message_type::ACTION_GENERAL_WORK;
+            _message.general_work_args_noutput_items = noutput_items;
+            _message.general_work_args_ninput_items = ninput_items;
+            copy_pointers(_message.general_work_args_input_items, input_items);
+            _message.general_work_args_output_items = output_items;
+            _handler->calleval(0);
+            return _message.general_work_args_return_value;
+
+        default:
+            int r = work (noutput_items, input_items, output_items);
+            if (r > 0) consume_each(r*_decim/_interp);
+            return r;
+        }
+    }
+
+    int work(
+        int noutput_items,
+        gr_vector_const_void_star &input_items,
+        gr_vector_void_star &output_items
+    ){
+        _message.action = gr_block_gw_message_type::ACTION_WORK;
+        _message.work_args_ninput_items = fixed_rate_noutput_to_ninput(noutput_items);
+        if (_message.work_args_ninput_items == 0) return -1;
+        _message.work_args_noutput_items = noutput_items;
+        copy_pointers(_message.work_args_input_items, input_items);
+        _message.work_args_output_items = output_items;
+        _handler->calleval(0);
+        return _message.work_args_return_value;
+    }
+
+    int fixed_rate_noutput_to_ninput(int noutput_items){
+        return (noutput_items*_decim/_interp) + history() - 1;
+    }
+
+    int fixed_rate_ninput_to_noutput(int ninput_items){
+        return std::max(0, ninput_items - (int)history() + 1)*_interp/_decim;
+    }
+
+    bool start(void){
+        _message.action = gr_block_gw_message_type::ACTION_START;
+        _handler->calleval(0);
+        return _message.start_args_return_value;
+    }
+
+    bool stop(void){
+        _message.action = gr_block_gw_message_type::ACTION_STOP;
+        _handler->calleval(0);
+        return _message.stop_args_return_value;
+    }
+
+    gr_block_gw_message_type &gr_block_message(void){
+        return _message;
+    }
+
+private:
+    gr_feval_ll *_handler;
+    gr_block_gw_message_type _message;
+    const gr_block_gw_work_type _work_type;
+    unsigned _decim, _interp;
+};
+
+boost::shared_ptr<gr_block_gateway> gr_make_block_gateway(
+    gr_feval_ll *handler,
+    const std::string &name,
+    gr_io_signature_sptr in_sig,
+    gr_io_signature_sptr out_sig,
+    const gr_block_gw_work_type work_type,
+    const unsigned factor
+){
+    return boost::shared_ptr<gr_block_gateway>(
+        new gr_block_gateway_impl(handler, name, in_sig, out_sig, work_type, factor)
+    );
+}
diff --git a/gnuradio-core/src/lib/general/gr_block_gateway.h b/gnuradio-core/src/lib/general/gr_block_gateway.h
new file mode 100644
index 000000000..ae91d41b5
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_block_gateway.h
@@ -0,0 +1,212 @@
+/*
+ * Copyright 2011-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.
+ */
+
+#ifndef INCLUDED_GRBLOCK_GATEWAY_H
+#define INCLUDED_GRBLOCK_GATEWAY_H
+
+#include <gr_core_api.h>
+#include <gr_block.h>
+#include <gr_feval.h>
+
+/*!
+ * The work type enum tells the gateway what kind of block to implement.
+ * The choices are familiar gnuradio block overloads (sync, decim, interp).
+ */
+enum gr_block_gw_work_type{
+    GR_BLOCK_GW_WORK_GENERAL,
+    GR_BLOCK_GW_WORK_SYNC,
+    GR_BLOCK_GW_WORK_DECIM,
+    GR_BLOCK_GW_WORK_INTERP,
+};
+
+/*!
+ * Shared message structure between python and gateway.
+ * Each action type represents a scheduler-called function.
+ */
+struct gr_block_gw_message_type{
+    enum action_type{
+        ACTION_GENERAL_WORK, //dispatch work
+        ACTION_WORK, //dispatch work
+        ACTION_FORECAST, //dispatch forecast
+        ACTION_START, //dispatch start
+        ACTION_STOP, //dispatch stop
+    };
+
+    action_type action;
+
+    int general_work_args_noutput_items;
+    std::vector<int> general_work_args_ninput_items;
+    std::vector<void *> general_work_args_input_items; //TODO this should be const void*, but swig cant int cast it right
+    std::vector<void *> general_work_args_output_items;
+    int general_work_args_return_value;
+
+    int work_args_ninput_items;
+    int work_args_noutput_items;
+    std::vector<void *> work_args_input_items; //TODO this should be const void*, but swig cant int cast it right
+    std::vector<void *> work_args_output_items;
+    int work_args_return_value;
+
+    int forecast_args_noutput_items;
+    std::vector<int> forecast_args_ninput_items_required;
+
+    bool start_args_return_value;
+
+    bool stop_args_return_value;
+};
+
+/*!
+ * The gateway block which performs all the magic.
+ *
+ * The gateway provides access to all the gr_block routines.
+ * The methods prefixed with gr_block__ are renamed
+ * to class methods without the prefix in python.
+ */
+class GR_CORE_API gr_block_gateway : virtual public gr_block{
+public:
+    //! Provide access to the shared message object
+    virtual gr_block_gw_message_type &gr_block_message(void) = 0;
+
+    long gr_block__unique_id(void) const{
+        return gr_block::unique_id();
+    }
+
+    std::string gr_block__name(void) const{
+        return gr_block::name();
+    }
+
+    unsigned gr_block__history(void) const{
+        return gr_block::history();
+    }
+
+    void gr_block__set_history(unsigned history){
+        return gr_block::set_history(history);
+    }
+
+    void gr_block__set_fixed_rate(bool fixed_rate){
+        return gr_block::set_fixed_rate(fixed_rate);
+    }
+
+    bool gr_block__fixed_rate(void) const{
+        return gr_block::fixed_rate();
+    }
+
+    void gr_block__set_output_multiple(int multiple){
+        return gr_block::set_output_multiple(multiple);
+    }
+
+    int gr_block__output_multiple(void) const{
+        return gr_block::output_multiple();
+    }
+
+    void gr_block__consume(int which_input, int how_many_items){
+        return gr_block::consume(which_input, how_many_items);
+    }
+
+    void gr_block__consume_each(int how_many_items){
+        return gr_block::consume_each(how_many_items);
+    }
+
+    void gr_block__produce(int which_output, int how_many_items){
+        return gr_block::produce(which_output, how_many_items);
+    }
+
+    void gr_block__set_relative_rate(double relative_rate){
+        return gr_block::set_relative_rate(relative_rate);
+    }
+
+    double gr_block__relative_rate(void) const{
+        return gr_block::relative_rate();
+    }
+
+    uint64_t gr_block__nitems_read(unsigned int which_input){
+        return gr_block::nitems_read(which_input);
+    }
+
+    uint64_t gr_block__nitems_written(unsigned int which_output){
+        return gr_block::nitems_written(which_output);
+    }
+
+    gr_block::tag_propagation_policy_t gr_block__tag_propagation_policy(void){
+        return gr_block::tag_propagation_policy();
+    }
+
+    void gr_block__set_tag_propagation_policy(gr_block::tag_propagation_policy_t p){
+        return gr_block::set_tag_propagation_policy(p);
+    }
+
+    void gr_block__add_item_tag(
+        unsigned int which_output, const gr_tag_t &tag
+    ){
+        return gr_block::add_item_tag(which_output, tag);
+    }
+
+    void gr_block__add_item_tag(
+        unsigned int which_output,
+        uint64_t abs_offset,
+        const pmt::pmt_t &key,
+        const pmt::pmt_t &value,
+        const pmt::pmt_t &srcid=pmt::PMT_F
+    ){
+        return gr_block::add_item_tag(which_output, abs_offset, key, value, srcid);
+    }
+
+    std::vector<gr_tag_t> gr_block__get_tags_in_range(
+        unsigned int which_input,
+        uint64_t abs_start,
+        uint64_t abs_end
+    ){
+        std::vector<gr_tag_t> tags;
+        gr_block::get_tags_in_range(tags, which_input, abs_start, abs_end);
+        return tags;
+    }
+
+    std::vector<gr_tag_t> gr_block__get_tags_in_range(
+        unsigned int which_input,
+        uint64_t abs_start,
+        uint64_t abs_end,
+        const pmt::pmt_t &key
+    ){
+        std::vector<gr_tag_t> tags;
+        gr_block::get_tags_in_range(tags, which_input, abs_start, abs_end, key);
+        return tags;
+    }
+};
+
+/*!
+ * Make a new gateway block.
+ * \param handler the swig director object with callback
+ * \param name the name of the block (Ex: "Shirley")
+ * \param in_sig the input signature for this block
+ * \param out_sig the output signature for this block
+ * \param work_type the type of block overload to implement
+ * \param factor the decimation or interpolation factor
+ * \return a new gateway block
+ */
+GR_CORE_API boost::shared_ptr<gr_block_gateway> gr_make_block_gateway(
+    gr_feval_ll *handler,
+    const std::string &name,
+    gr_io_signature_sptr in_sig,
+    gr_io_signature_sptr out_sig,
+    const gr_block_gw_work_type work_type,
+    const unsigned factor
+);
+
+#endif /* INCLUDED_GRBLOCK_GATEWAY_H */
diff --git a/gnuradio-core/src/lib/general/gr_block_gateway.i b/gnuradio-core/src/lib/general/gr_block_gateway.i
new file mode 100644
index 000000000..8adafdfea
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_block_gateway.i
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2011-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.
+ */
+
+////////////////////////////////////////////////////////////////////////
+// standard includes
+////////////////////////////////////////////////////////////////////////
+%include <gnuradio.i>
+%include <gr_tags.i>
+%include <gr_feval.i>
+
+////////////////////////////////////////////////////////////////////////
+// block headers
+////////////////////////////////////////////////////////////////////////
+%{
+#include <gr_block_gateway.h>
+%}
+
+////////////////////////////////////////////////////////////////////////
+// data type support
+////////////////////////////////////////////////////////////////////////
+%template(int_vector_t) std::vector<int>;
+%template(void_star_vector_t) std::vector<void *>;
+
+////////////////////////////////////////////////////////////////////////
+// block magic
+////////////////////////////////////////////////////////////////////////
+GR_SWIG_BLOCK_MAGIC(gr,block_gateway);
+%include <gr_block_gateway.h>
diff --git a/gnuradio-core/src/lib/general/gr_message_strobe.cc b/gnuradio-core/src/lib/general/gr_message_strobe.cc
new file mode 100644
index 000000000..6a9f807d1
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_message_strobe.cc
@@ -0,0 +1,75 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_message_strobe.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_strobe_sptr
+gr_make_message_strobe (pmt::pmt_t msg, float period_ms)
+{
+  return gnuradio::get_initial_sptr(new gr_message_strobe(msg, period_ms));
+}
+
+gr_message_strobe::gr_message_strobe (pmt::pmt_t msg, float period_ms)
+  : gr_block("message_strobe",
+	     gr_make_io_signature(0, 0, 0),
+	     gr_make_io_signature(0, 0, 0)),
+    d_finished(false),
+    d_period_ms(period_ms),
+    d_msg(msg)
+{
+    message_port_register_out(pmt::mp("strobe"));
+    d_thread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&gr_message_strobe::run, this)));
+
+    message_port_register_in(pmt::mp("set_msg"));
+    set_msg_handler(pmt::mp("set_msg"), boost::bind(&gr_message_strobe::set_msg, this, _1));
+}
+
+gr_message_strobe::~gr_message_strobe()
+{
+    d_finished = true;
+    d_thread->interrupt();
+    d_thread->join();
+}
+
+void gr_message_strobe::run(){
+    while(!d_finished) {
+        boost::this_thread::sleep(boost::posix_time::milliseconds(d_period_ms)); 
+        if(d_finished){ return; }
+
+        message_port_pub( pmt::mp("strobe"), d_msg );
+    } 
+}
diff --git a/gnuradio-core/src/lib/general/gr_message_strobe.h b/gnuradio-core/src/lib/general/gr_message_strobe.h
new file mode 100644
index 000000000..89046ffc0
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_message_strobe.h
@@ -0,0 +1,62 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifndef INCLUDED_GR_MESSAGE_STROBE_H
+#define INCLUDED_GR_MESSAGE_STROBE_H
+
+#include <gr_core_api.h>
+#include <gr_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+
+class gr_message_strobe;
+typedef boost::shared_ptr<gr_message_strobe> gr_message_strobe_sptr;
+
+GR_CORE_API gr_message_strobe_sptr gr_make_message_strobe (pmt::pmt_t msg, float period_ms);
+
+/*!
+ * \brief Send message at defined interval
+ * \ingroup msg_blk
+ */
+class GR_CORE_API gr_message_strobe : public gr_block
+{
+ private:
+  friend GR_CORE_API gr_message_strobe_sptr
+  gr_make_message_strobe(pmt::pmt_t msg, float period_ms);
+
+  boost::shared_ptr<boost::thread> d_thread;
+  bool d_finished;
+  float d_period_ms;
+  pmt::pmt_t d_msg;
+
+  void run();
+
+ protected:
+  gr_message_strobe (pmt::pmt_t msg, float period_ms);
+
+ public:
+  ~gr_message_strobe ();
+
+  void set_msg(pmt::pmt_t msg){ d_msg = msg; }
+};
+
+#endif /* INCLUDED_GR_MESSAGE_STROBE_H */
diff --git a/gnuradio-core/src/lib/general/gr_message_strobe.i b/gnuradio-core/src/lib/general/gr_message_strobe.i
new file mode 100644
index 000000000..490aa8e8a
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_message_strobe.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_strobe);
+
+%{
+#include <gr_message_strobe.h>
+%}
+
+%include "gr_message_strobe.h"
+
diff --git a/gnuradio-core/src/lib/general/gr_nop.cc b/gnuradio-core/src/lib/general/gr_nop.cc
index ca5983c39..edfe1d76d 100644
--- a/gnuradio-core/src/lib/general/gr_nop.cc
+++ b/gnuradio-core/src/lib/general/gr_nop.cc
@@ -40,7 +40,8 @@ gr_nop::gr_nop (size_t sizeof_stream_item)
     d_nmsgs_recvd(0)
 {
   // Arrange to have count_received_msgs called when messages are received.
-  set_msg_handler(boost::bind(&gr_nop::count_received_msgs, this, _1));
+  message_port_register_in(pmt::mp("port"));
+  set_msg_handler(pmt::mp("port"), boost::bind(&gr_nop::count_received_msgs, this, _1));
 }
 
 // Trivial message handler that just counts them.
diff --git a/gnuradio-core/src/lib/general/gr_simple_correlator.cc b/gnuradio-core/src/lib/general/gr_simple_correlator.cc
index 9c2ebef15..b9209e74f 100644
--- a/gnuradio-core/src/lib/general/gr_simple_correlator.cc
+++ b/gnuradio-core/src/lib/general/gr_simple_correlator.cc
@@ -158,11 +158,13 @@ gr_simple_correlator::general_work (int noutput_items,
   int decision;
   int hamming_dist;
 
+#ifdef DEBUG_SIMPLE_CORRELATOR
   struct debug_data {
     float	raw_data;
     float	sampled;
     float 	enter_locked;
   } debug_data;
+#endif
 
   while (n < nin){
 
@@ -212,7 +214,9 @@ gr_simple_correlator::general_work (int noutput_items,
       else if (d_state == ST_UNDER_THRESHOLD && hamming_dist > THRESHOLD){
 	// no longer seeing good PN code, compute center of goodness
 	enter_locked ();
+#ifdef DEBUG_SIMPLE_CORRELATOR
 	debug_data.enter_locked = 1.0;
+#endif
       }
       break;
 
diff --git a/gnuradio-core/src/lib/gengen/gr_vector_insert_X.cc.t b/gnuradio-core/src/lib/gengen/gr_vector_insert_X.cc.t
index 37963cdfe..20968afe2 100644
--- a/gnuradio-core/src/lib/gengen/gr_vector_insert_X.cc.t
+++ b/gnuradio-core/src/lib/gengen/gr_vector_insert_X.cc.t
@@ -44,7 +44,7 @@
     // some sanity checks
     assert(offset < periodicity);
     assert(offset >= 0);
-    assert(periodicity > data.size());
+    assert((size_t)periodicity > data.size());
 }
 
 int
diff --git a/gnuradio-core/src/lib/io/CMakeLists.txt b/gnuradio-core/src/lib/io/CMakeLists.txt
index ddd43a4a3..c5b85d304 100644
--- a/gnuradio-core/src/lib/io/CMakeLists.txt
+++ b/gnuradio-core/src/lib/io/CMakeLists.txt
@@ -38,6 +38,8 @@ 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
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_stream_pdu_base.cc
 )
 
 ########################################################################
@@ -59,6 +61,8 @@ 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
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_stream_pdu_base.h
     DESTINATION ${GR_INCLUDE_DIR}/gnuradio
     COMPONENT "core_devel"
 )
@@ -86,9 +90,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
@@ -99,6 +105,9 @@ set(gr_core_io_triple_threats
     gr_wavfile_source
     gr_wavfile_sink
     gr_tagged_file_sink
+    gr_tagged_stream_to_pdu
+    gr_tuntap_pdu
+    gr_socket_pdu
 )
 
 foreach(file_tt ${gr_core_io_triple_threats})
diff --git a/gnuradio-core/src/lib/io/gr_file_sink.cc b/gnuradio-core/src/lib/io/gr_file_sink.cc
index 5d147fcfe..10c8360cb 100644
--- a/gnuradio-core/src/lib/io/gr_file_sink.cc
+++ b/gnuradio-core/src/lib/io/gr_file_sink.cc
@@ -42,8 +42,6 @@ gr_file_sink::gr_file_sink(size_t itemsize, const char *filename)
     gr_file_sink_base(filename, true),
     d_itemsize(itemsize)
 {
-  if (!open(filename))
-    throw std::runtime_error ("can't open file");
 }
 
 gr_file_sink::~gr_file_sink ()
@@ -55,23 +53,32 @@ gr_file_sink::work (int noutput_items,
 		    gr_vector_const_void_star &input_items,
 		    gr_vector_void_star &output_items)
 {
-  char *inbuf = (char *) input_items[0];
+  char *inbuf = (char*)input_items[0];
   int  nwritten = 0;
 
   do_update();				// update d_fp is reqd
 
-  if (!d_fp)
+  if(!d_fp)
     return noutput_items;		// drop output on the floor
 
-  while (nwritten < noutput_items){
-    int count = fwrite (inbuf, d_itemsize, noutput_items - nwritten, d_fp);
-    if (count == 0)	// FIXME add error handling
-      break;
+  while(nwritten < noutput_items) {
+    int count = fwrite(inbuf, d_itemsize, noutput_items - nwritten, d_fp);
+    if(count == 0) {
+      if(ferror(d_fp)) {
+	std::stringstream s;
+	s << "file_sink write failed with error " << fileno(d_fp) << std::endl;
+	throw std::runtime_error(s.str());
+      }
+      else { // is EOF
+	break;
+      }
+    }
     nwritten += count;
     inbuf += count * d_itemsize;
   }
-  if (d_unbuffered)
-	  fflush (d_fp);
+
+  if(d_unbuffered)
+    fflush (d_fp);
 
   return nwritten;
 }
diff --git a/gnuradio-core/src/lib/io/gr_file_sink_base.cc b/gnuradio-core/src/lib/io/gr_file_sink_base.cc
index cb67bbb6c..4eecf928e 100644
--- a/gnuradio-core/src/lib/io/gr_file_sink_base.cc
+++ b/gnuradio-core/src/lib/io/gr_file_sink_base.cc
@@ -126,5 +126,5 @@ gr_file_sink_base::do_update()
 void
 gr_file_sink_base::set_unbuffered(bool unbuffered)
 {
-	d_unbuffered = unbuffered;
+  d_unbuffered = unbuffered;
 }
diff --git a/gnuradio-core/src/lib/io/gr_file_source.cc b/gnuradio-core/src/lib/io/gr_file_source.cc
index 3f06a8244..09f3986cd 100644
--- a/gnuradio-core/src/lib/io/gr_file_source.cc
+++ b/gnuradio-core/src/lib/io/gr_file_source.cc
@@ -49,24 +49,14 @@
 #define	OUR_O_LARGEFILE 0
 #endif
 
-gr_file_source::gr_file_source (size_t itemsize, const char *filename, bool repeat)
-  : gr_sync_block ("file_source",
-		   gr_make_io_signature (0, 0, 0),
-		   gr_make_io_signature (1, 1, itemsize)),
-    d_itemsize (itemsize), d_fp (0), d_repeat (repeat)
+gr_file_source::gr_file_source(size_t itemsize, const char *filename, bool repeat)
+  : gr_sync_block("file_source",
+		  gr_make_io_signature (0, 0, 0),
+		  gr_make_io_signature (1, 1, itemsize)),
+    d_itemsize(itemsize), d_fp(0), d_new_fp (0), d_repeat(repeat),
+    d_updated(false)
 {
-  // we use "open" to use to the O_LARGEFILE flag
-
-  int fd;
-  if ((fd = open (filename, O_RDONLY | OUR_O_LARGEFILE | OUR_O_BINARY)) < 0){
-    perror (filename);
-    throw std::runtime_error ("can't open file");
-  }
-
-  if ((d_fp = fdopen (fd, "rb")) == NULL){
-    perror (filename);
-    throw std::runtime_error ("can't open file");
-  }
+  open(filename, repeat);
 }
 
 // public constructor that returns a shared_ptr
@@ -79,7 +69,11 @@ gr_make_file_source (size_t itemsize, const char *filename, bool repeat)
 
 gr_file_source::~gr_file_source ()
 {
-  fclose ((FILE *) d_fp);
+  close();
+  if(d_fp) {
+    fclose(d_fp);
+    d_fp = 0;
+  }
 }
 
 int
@@ -91,6 +85,11 @@ gr_file_source::work (int noutput_items,
   int i;
   int size = noutput_items;
 
+  do_update();       // update d_fp is reqd
+  if(d_fp == NULL)
+    throw std::runtime_error("work with file not open");
+
+  boost::mutex::scoped_lock lock(fp_mutex); // hold for the rest of this function
   while (size) {
     i = fread(o, d_itemsize, size, (FILE *) d_fp);
 
@@ -111,8 +110,9 @@ gr_file_source::work (int noutput_items,
       break;
 
     if (fseek ((FILE *) d_fp, 0, SEEK_SET) == -1) {
-      fprintf(stderr, "[%s] fseek failed\n", __FILE__);
-      exit(-1);
+      std::stringstream s;
+      s << "[" << __FILE__ << "]" << " fseek failed" << std::endl;
+      throw std::runtime_error(s.str());
     }
   }
 
@@ -128,5 +128,64 @@ gr_file_source::work (int noutput_items,
 bool
 gr_file_source::seek (long seek_point, int whence)
 {
-   return fseek ((FILE *) d_fp, seek_point * d_itemsize, whence) == 0;
+  // obtain exclusive access for duration of this function
+  boost::mutex::scoped_lock lock(fp_mutex);
+  return fseek((FILE *) d_fp, seek_point * d_itemsize, whence) == 0;
+}
+
+void
+gr_file_source::open(const char *filename, bool repeat)
+{
+  // obtain exclusive access for duration of this function
+  boost::mutex::scoped_lock     lock(fp_mutex);
+
+  int fd;
+
+  // we use "open" to use to the O_LARGEFILE flag
+  if((fd = ::open(filename, O_RDONLY | OUR_O_LARGEFILE | OUR_O_BINARY)) < 0) {
+    perror(filename);
+    throw std::runtime_error("can't open file");
+  }
+
+  if(d_new_fp) {
+    fclose(d_new_fp);
+    d_new_fp = 0;
+  }
+
+  if((d_new_fp = fdopen (fd, "rb")) == NULL) {
+    perror(filename);
+    ::close(fd);	// don't leak file descriptor if fdopen fails
+    throw std::runtime_error("can't open file");
+  }
+
+  d_updated = true;
+  d_repeat = repeat;
+}
+
+void
+gr_file_source::close()
+{
+  // obtain exclusive access for duration of this function
+  boost::mutex::scoped_lock lock(fp_mutex);
+
+  if(d_new_fp != NULL) {
+    fclose(d_new_fp);
+    d_new_fp = NULL;
+  }
+  d_updated = true;
+}
+
+void
+gr_file_source::do_update()
+{
+  if(d_updated) {
+    boost::mutex::scoped_lock lock(fp_mutex); // hold while in scope
+
+    if(d_fp)
+      fclose(d_fp);
+
+    d_fp = d_new_fp;                    // install new file pointer
+    d_new_fp = 0;
+    d_updated = false;
+  }
 }
diff --git a/gnuradio-core/src/lib/io/gr_file_source.h b/gnuradio-core/src/lib/io/gr_file_source.h
index 1cc44a3b1..0478fba04 100644
--- a/gnuradio-core/src/lib/io/gr_file_source.h
+++ b/gnuradio-core/src/lib/io/gr_file_source.h
@@ -25,6 +25,7 @@
 
 #include <gr_core_api.h>
 #include <gr_sync_block.h>
+#include <boost/thread/mutex.hpp>
 
 class gr_file_source;
 typedef boost::shared_ptr<gr_file_source> gr_file_source_sptr;
@@ -39,31 +40,68 @@ gr_make_file_source (size_t itemsize, const char *filename, bool repeat = false)
 
 class GR_CORE_API gr_file_source : public gr_sync_block
 {
-  friend GR_CORE_API gr_file_source_sptr gr_make_file_source (size_t itemsize,
-						  const char *filename,
-						  bool repeat);
  private:
-  size_t	d_itemsize;
-  void	       *d_fp;
-  bool		d_repeat;
+  size_t d_itemsize;
+  FILE *d_fp;
+  FILE *d_new_fp;
+  bool d_repeat;
+  bool d_updated;
 
  protected:
-  gr_file_source (size_t itemsize, const char *filename, bool repeat);
+  gr_file_source(size_t itemsize, const char *filename, bool repeat);
+
+  void do_update();
+
+  boost::mutex fp_mutex;
 
  public:
-  ~gr_file_source ();
+  /*!
+   * \brief Create a file source.
+   *
+   * Opens \p filename as a source of items into a flowgraph. The data
+   * is expected to be in binary format, item after item. The \p
+   * itemsize of the block determines the conversion from bits to
+   * items.
+   *
+   * If \p repeat is turned on, the file will repeat the file after
+   * it's reached the end.
+   *
+   * \param itemsize	the size of each item in the file, in bytes
+   * \param filename	name of the file to source from
+   * \param repeat	repeat file from start
+   */
+  friend GR_CORE_API gr_file_source_sptr
+    gr_make_file_source(size_t itemsize,
+			const char *filename,
+			bool repeat);
+
+  ~gr_file_source();
 
-  int work (int noutput_items,
-	    gr_vector_const_void_star &input_items,
-	    gr_vector_void_star &output_items);
+  int work(int noutput_items,
+	   gr_vector_const_void_star &input_items,
+	   gr_vector_void_star &output_items);
 
   /*!
-   * \brief seek file to \p seek_point relative to \p whence
+   * \brief Seek file to \p seek_point relative to \p whence
    *
    * \param seek_point	sample offset in file
    * \param whence	one of SEEK_SET, SEEK_CUR, SEEK_END (man fseek)
    */
-  bool seek (long seek_point, int whence);
+  bool seek(long seek_point, int whence);
+
+  /*!
+   * \brief Opens a new file.
+   *
+   * \param filename	name of the file to source from
+   * \param repeat	repeat file from start
+   */
+  void open(const char *filename, bool repeat);
+
+  /*!
+   * \brief Close the file handle.
+   */
+  void close();
+
 };
 
 #endif /* INCLUDED_GR_FILE_SOURCE_H */
diff --git a/gnuradio-core/src/lib/io/gr_file_source.i b/gnuradio-core/src/lib/io/gr_file_source.i
index 9bf44691d..e71cef0d1 100644
--- a/gnuradio-core/src/lib/io/gr_file_source.i
+++ b/gnuradio-core/src/lib/io/gr_file_source.i
@@ -40,4 +40,6 @@ class gr_file_source : public gr_sync_block
   ~gr_file_source ();
 
   bool seek (long seek_point, int whence);
+  void open (const char *filename, bool repeat);
+  void close();
 };
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..27f4c65fd
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.cc
@@ -0,0 +1,117 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005,2010,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.
+ */
+
+#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 << "******* MESSAGE DEBUG PRINT ********\n";
+  pmt::pmt_print(msg);
+  std::cout << "************************************\n";
+}
+
+void
+gr_message_debug::store(pmt::pmt_t msg)
+{
+  gruel::scoped_lock guard(d_mutex);
+  d_messages.push_back(msg);
+}
+
+void
+gr_message_debug::print_verbose(pmt::pmt_t msg)
+{
+  pmt::pmt_t meta = pmt::pmt_car(msg);
+  pmt::pmt_t vector = pmt::pmt_cdr(msg);
+  std::cout << "* MESSAGE DEBUG PRINT PDU VERBOSE *\n";
+  pmt::pmt_print(meta);
+  size_t len = pmt::pmt_length(vector);
+  std::cout << "pdu_length = " << len << std::endl;
+  std::cout << "contents = " << std::endl;
+  size_t offset(0);
+  const uint8_t* d = (const uint8_t*) pmt_uniform_vector_elements(vector, offset);
+  for(size_t i=0; i<len; i+=16){
+    printf("%04x: ", i);
+    for(size_t j=i; j<std::min(i+16,len); j++){
+      printf("%02x ",d[j] );
+    }
+
+    std::cout << std::endl;
+  }
+
+  std::cout << "***********************************\n";
+}
+
+int
+gr_message_debug::num_messages()
+{
+  return (int)d_messages.size();
+}
+
+pmt::pmt_t
+gr_message_debug::get_message(int i)
+{
+  gruel::scoped_lock guard(d_mutex);
+
+  if((size_t)i >= d_messages.size()) {
+    throw std::runtime_error("gr_message_debug: index for message out of bounds.\n");
+  }
+
+  return d_messages[i];
+}
+
+gr_message_debug::gr_message_debug()
+  : gr_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));
+  
+  message_port_register_in(pmt::mp("store"));
+  set_msg_handler(pmt::mp("store"), boost::bind(&gr_message_debug::store, this, _1));
+}
+
+gr_message_debug::~gr_message_debug()
+{
+}
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..6e6e5103c
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.h
@@ -0,0 +1,103 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005,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.
+ */
+
+#ifndef INCLUDED_GR_MESSAGE_DEBUG_H
+#define INCLUDED_GR_MESSAGE_DEBUG_H
+
+#include <gr_core_api.h>
+#include <gr_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+#include <gruel/thread.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 Print received messages to stdout
+ * \ingroup sink_blk
+ */
+class GR_CORE_API gr_message_debug : public gr_block
+{
+ private:
+  friend GR_CORE_API gr_message_debug_sptr
+  gr_make_message_debug();
+
+  /*!
+   * \brief Messages received in this port are printed to stdout.
+   *
+   * This port receives messages from the scheduler's message handling
+   * mechanism and prints it to stdout. This message handler function
+   * is only meant to be used by the scheduler to handle messages
+   * posted to port 'print'.
+   *
+   * \param msg A pmt message passed from the scheduler's message handling.
+   */
+  void print(pmt::pmt_t msg);
+  void print_verbose(pmt::pmt_t msg);
+
+  /*!
+   * \brief Messages received in this port are stored in a vector.
+   *
+   * This port receives messages from the scheduler's message handling
+   * mechanism and stores it in a vector. Messages can be retrieved
+   * later using the 'get_message' function. This message handler
+   * function is only meant to be used by the scheduler to handle
+   * messages posted to port 'store'.
+   *
+   * \param msg A pmt message passed from the scheduler's message handling.
+   */
+  void store(pmt::pmt_t msg);
+
+  gruel::mutex d_mutex;
+  std::vector<pmt::pmt_t> d_messages;
+
+ protected:
+  gr_message_debug ();
+
+ public:
+  ~gr_message_debug ();
+
+  /*! 
+   * \brief Reports the number of messages received by this block.
+   */
+  int num_messages();
+
+  /*!
+   * \brief Get a message (as a PMT) from the message vector at index \p i.
+   *
+   * Messages passed to the 'store' port will be stored in a
+   * vector. This function retrieves those messages by index. They are
+   * index in order of when they were received (all messages are just
+   * pushed onto the back of a vector). This is mostly useful in
+   * debugging message passing graphs and in QA code.
+   *
+   * \param i The index in the vector for the message to retrieve.
+   *
+   * \return a message at index \p i as a pmt_t.
+   */
+  pmt::pmt_t get_message(int i);
+};
+
+#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..65d3bfc4a
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_message_debug.i
@@ -0,0 +1,30 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+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..b2757c307
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu.cc
@@ -0,0 +1,79 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#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!");
+  }
+}
+
+gr_pdu_vector_type type_from_pmt(pmt::pmt_t vector){
+  if(pmt_is_u8vector(vector))
+    return BYTE;
+  if(pmt_is_f32vector(vector))
+    return FLOAT;
+  if(pmt_is_c32vector(vector))
+    return COMPLEX;
+  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..5ed9cdded
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu.h
@@ -0,0 +1,39 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#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);
+gr_pdu_vector_type type_from_pmt(pmt::pmt_t vector);
+
+#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..5c319dc39
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.cc
@@ -0,0 +1,132 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#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 ) );
+    pmt::pmt_t msg( delete_head_blocking( 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..3105a3d38
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.h
@@ -0,0 +1,63 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#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..ec760b309
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_pdu_to_tagged_stream.i
@@ -0,0 +1,31 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+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_socket_pdu.cc b/gnuradio-core/src/lib/io/gr_socket_pdu.cc
new file mode 100644
index 000000000..bb374b300
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_socket_pdu.cc
@@ -0,0 +1,157 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_socket_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>
+#include <iostream>
+#include <gr_pdu.h>
+#include <boost/format.hpp>
+
+// public constructor that returns a shared_ptr
+gr_socket_pdu_sptr
+gr_make_socket_pdu (std::string type, std::string addr, std::string port, int MTU)
+{
+  return gnuradio::get_initial_sptr(new gr_socket_pdu(type,addr,port,MTU));
+}
+
+gr_socket_pdu::gr_socket_pdu (std::string type, std::string addr, std::string port, int MTU)
+    : gr_stream_pdu_base(MTU)
+{
+
+    if( (type == "TCP_SERVER") || (type == "TCP_CLIENT")){
+        boost::asio::ip::tcp::resolver resolver(_io_service);
+        boost::asio::ip::tcp::resolver::query query(boost::asio::ip::tcp::v4(), addr, port);
+        _tcp_endpoint = *resolver.resolve(query);
+    }
+    if( (type == "UDP_SERVER") || (type == "UDP_CLIENT")){
+        boost::asio::ip::udp::resolver resolver(_io_service);
+        boost::asio::ip::udp::resolver::query query(boost::asio::ip::udp::v4(), addr, port);
+        if( (type == "UDP_SERVER") ){
+            _udp_endpoint = *resolver.resolve(query);
+        } else {
+            _udp_endpoint_other = *resolver.resolve(query);
+        }
+    }
+
+    // register ports
+    message_port_register_out(pmt::mp("pdus"));
+    message_port_register_in(pmt::mp("pdus"));
+
+    // set up socketry
+    if (type == "TCP_SERVER"){
+        _acceptor_tcp.reset(new boost::asio::ip::tcp::acceptor(_io_service, _tcp_endpoint));
+        _acceptor_tcp->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
+        start_tcp_accept();
+        // bind tcp server send handler
+        set_msg_handler(pmt::mp("pdus"), boost::bind(&gr_socket_pdu::tcp_server_send, this, _1));
+    } else if(type =="TCP_CLIENT"){
+        boost::system::error_code error = boost::asio::error::host_not_found;
+        _tcp_socket.reset(new boost::asio::ip::tcp::socket(_io_service));
+        _tcp_socket->connect(_tcp_endpoint, error);
+        if(error){
+            throw boost::system::system_error(error);
+        }
+        set_msg_handler(pmt::mp("pdus"), boost::bind(&gr_socket_pdu::tcp_client_send, this, _1));
+        _tcp_socket->async_read_some(
+            boost::asio::buffer(rxbuf),
+            boost::bind(&gr_socket_pdu::handle_tcp_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+
+    } else if(type =="UDP_SERVER"){
+        _udp_socket.reset(new boost::asio::ip::udp::socket(_io_service, _udp_endpoint));
+        _udp_socket->async_receive_from( boost::asio::buffer(rxbuf), _udp_endpoint_other, 
+            boost::bind(&gr_socket_pdu::handle_udp_read, this,
+                boost::asio::placeholders::error,
+                boost::asio::placeholders::bytes_transferred)); 
+        set_msg_handler(pmt::mp("pdus"), boost::bind(&gr_socket_pdu::udp_send, this, _1));
+    } else if(type =="UDP_CLIENT"){
+        _udp_socket.reset(new boost::asio::ip::udp::socket(_io_service, _udp_endpoint));
+        _udp_socket->async_receive_from( boost::asio::buffer(rxbuf), _udp_endpoint_other, 
+            boost::bind(&gr_socket_pdu::handle_udp_read, this,
+                boost::asio::placeholders::error,
+                boost::asio::placeholders::bytes_transferred)); 
+        set_msg_handler(pmt::mp("pdus"), boost::bind(&gr_socket_pdu::udp_send, this, _1));
+    } else {
+        throw std::runtime_error("unknown socket type!");
+    }
+    
+    // start thread for io_service
+    d_thread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&gr_socket_pdu::run_io_service, this)));
+    d_started = true;
+}
+
+void tcp_connection::handle_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred)
+  {
+    if(!error)
+    {
+        pmt::pmt_t vector = pmt::pmt_init_u8vector(bytes_transferred, (const uint8_t*)&buf[0]);
+        pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector);
+
+        d_block->message_port_pub( pmt::mp("pdus"), pdu );
+
+        socket_.async_read_some(
+            boost::asio::buffer(buf),
+            boost::bind(&tcp_connection::handle_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+
+    } else {
+        std::cout << "error occurred\n";
+    }
+
+  }
+
+
+void gr_socket_pdu::tcp_server_send(pmt::pmt_t msg){
+    pmt::pmt_t vector = pmt::pmt_cdr(msg);
+    for(size_t i=0; i<d_tcp_connections.size(); i++){
+        d_tcp_connections[i]->send(vector);
+    }
+}
+
+void gr_socket_pdu::tcp_client_send(pmt::pmt_t msg){
+    pmt::pmt_t vector = pmt::pmt_cdr(msg);
+    size_t len = pmt::pmt_length(vector);
+    size_t offset(0);
+    boost::array<char, 10000> txbuf;
+    memcpy(&txbuf[0], pmt::pmt_uniform_vector_elements(vector, offset), len);
+    _tcp_socket->send(boost::asio::buffer(txbuf,len));
+}
+
+void gr_socket_pdu::udp_send(pmt::pmt_t msg){
+    pmt::pmt_t vector = pmt::pmt_cdr(msg);
+    size_t len = pmt::pmt_length(vector);
+    size_t offset(0);
+    boost::array<char, 10000> txbuf;
+    memcpy(&txbuf[0], pmt::pmt_uniform_vector_elements(vector, offset), len);
+    if(_udp_endpoint_other.address().to_string() != "0.0.0.0")
+        _udp_socket->send_to(boost::asio::buffer(txbuf,len), _udp_endpoint_other);
+}
diff --git a/gnuradio-core/src/lib/io/gr_socket_pdu.h b/gnuradio-core/src/lib/io/gr_socket_pdu.h
new file mode 100644
index 000000000..f554febdc
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_socket_pdu.h
@@ -0,0 +1,201 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifndef INCLUDED_GR_SOCKET_PDU_H
+#define INCLUDED_GR_SOCKET_PDU_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+#include <gr_stream_pdu_base.h>
+#include <boost/asio.hpp>
+
+class gr_socket_pdu;
+typedef boost::shared_ptr<gr_socket_pdu> gr_socket_pdu_sptr;
+
+GR_CORE_API gr_socket_pdu_sptr gr_make_socket_pdu (std::string type, std::string addr, std::string port, int MTU=10000);
+
+class tcp_connection
+  : public boost::enable_shared_from_this<tcp_connection>
+{
+public:
+  typedef boost::shared_ptr<tcp_connection> pointer;
+  gr_socket_pdu *d_block;
+  boost::array<char, 10000> buf;
+
+  static pointer create(boost::asio::io_service& io_service)
+  {
+    return pointer(new tcp_connection(io_service));
+  }
+
+  boost::asio::ip::tcp::socket& socket()
+  {
+    return socket_;
+  }
+
+  void start(gr_socket_pdu* parent)
+  {
+    d_block = parent;
+//    message_ = "connected to gr_socket_pdu\n";
+//    boost::asio::async_write(socket_, boost::asio::buffer(message_),
+//        boost::bind(&tcp_connection::handle_write, shared_from_this(),
+//          boost::asio::placeholders::error,
+//          boost::asio::placeholders::bytes_transferred));
+
+    socket_.async_read_some(
+        boost::asio::buffer(buf),
+         boost::bind(&tcp_connection::handle_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+  }
+  void send(pmt::pmt_t vector){
+    size_t len = pmt::pmt_length(vector);
+    size_t offset(0);
+    boost::array<char, 10000> txbuf;
+    memcpy(&txbuf[0], pmt::pmt_uniform_vector_elements(vector, offset), len);
+    boost::asio::async_write(socket_, boost::asio::buffer(txbuf, len),
+        boost::bind(&tcp_connection::handle_write, shared_from_this(),
+          boost::asio::placeholders::error,
+          boost::asio::placeholders::bytes_transferred));
+    }
+
+  ~tcp_connection(){
+//    std::cout << "tcp_connection destroyed\n";
+    }
+
+private:
+  tcp_connection(boost::asio::io_service& io_service)
+    : socket_(io_service)
+  {
+  }
+
+  void handle_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred);
+
+  void handle_write(const boost::system::error_code& /*error*/,
+      size_t /*bytes_transferred*/)
+  {
+  }
+
+  boost::asio::ip::tcp::socket socket_;
+  std::string message_;
+};
+
+
+/*!
+ * \brief Gather received items into messages and insert into msgq
+ * \ingroup sink_blk
+ */
+class GR_CORE_API gr_socket_pdu : public gr_stream_pdu_base
+{
+ private:
+  friend GR_CORE_API gr_socket_pdu_sptr
+  gr_make_socket_pdu(std::string type, std::string addr, std::string port, int MTU);
+
+  boost::asio::io_service _io_service;
+ 
+  boost::array<char, 10000> rxbuf;
+
+  // tcp specific
+  boost::asio::ip::tcp::endpoint _tcp_endpoint;
+
+  // specific to tcp server
+  boost::shared_ptr<boost::asio::ip::tcp::acceptor> _acceptor_tcp;
+  std::vector<tcp_connection::pointer> d_tcp_connections;
+  void tcp_server_send(pmt::pmt_t msg);
+  void tcp_client_send(pmt::pmt_t msg);
+  void udp_send(pmt::pmt_t msg);
+
+  // specific to tcp client
+  boost::shared_ptr<boost::asio::ip::tcp::socket> _tcp_socket;
+
+  // specific to udp client/server
+  boost::asio::ip::udp::endpoint _udp_endpoint;
+  boost::asio::ip::udp::endpoint _udp_endpoint_other;
+  boost::shared_ptr<boost::asio::ip::udp::socket> _udp_socket;
+
+  void handle_receive(const boost::system::error_code& error, std::size_t ){
+    }
+ 
+  void start_tcp_accept(){
+    tcp_connection::pointer new_connection =
+        tcp_connection::create(_acceptor_tcp->get_io_service());
+
+    _acceptor_tcp->async_accept(new_connection->socket(),
+        boost::bind(&gr_socket_pdu::handle_tcp_accept, this, new_connection,
+            boost::asio::placeholders::error));
+    }
+
+  void handle_tcp_accept(tcp_connection::pointer new_connection, const boost::system::error_code& error){
+        if (!error)
+        {
+            new_connection->start(this);
+            d_tcp_connections.push_back(new_connection);
+            start_tcp_accept();
+        } else {
+            std::cout << error << std::endl;
+        }
+    }
+ 
+  void run_io_service(){
+    _io_service.run();
+    } 
+
+  void handle_udp_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred){
+    if(!error){
+        pmt::pmt_t vector = pmt::pmt_init_u8vector(bytes_transferred, (const uint8_t*)&rxbuf[0]);
+        pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector);
+        
+        message_port_pub( pmt::mp("pdus"), pdu );
+    
+        _udp_socket->async_receive_from( boost::asio::buffer(rxbuf), _udp_endpoint_other,
+            boost::bind(&gr_socket_pdu::handle_udp_read, this,
+                boost::asio::placeholders::error,
+                boost::asio::placeholders::bytes_transferred));
+    } else {
+        throw boost::system::system_error(error);
+//        std::cout << "error occurred\n";
+    }
+  }
+  void handle_tcp_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred){
+    if(!error)
+    {
+        pmt::pmt_t vector = pmt::pmt_init_u8vector(bytes_transferred, (const uint8_t*)&rxbuf[0]);
+        pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector);
+
+        message_port_pub( pmt::mp("pdus"), pdu );
+
+        _tcp_socket->async_read_some(
+            boost::asio::buffer(rxbuf),
+            boost::bind(&gr_socket_pdu::handle_tcp_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+
+    } else {
+        //std::cout << "error occurred\n";
+        throw boost::system::system_error(error);
+    }
+  }
+
+ protected:
+  gr_socket_pdu (std::string type, std::string addr, std::string port, int MTU=10000);
+ public:
+  ~gr_socket_pdu () {}
+};
+
+#endif /* INCLUDED_GR_TUNTAP_PDU_H */
diff --git a/gnuradio-core/src/lib/io/gr_socket_pdu.i b/gnuradio-core/src/lib/io/gr_socket_pdu.i
new file mode 100644
index 000000000..3e20b63e2
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_socket_pdu.i
@@ -0,0 +1,33 @@
+/* -*- 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,socket_pdu);
+
+%ignore tcp_connection;
+
+%{
+#include <gr_socket_pdu.h>
+%}
+
+%include "gr_stream_pdu_base.h"
+%include "gr_socket_pdu.h"
+
diff --git a/gnuradio-core/src/lib/io/gr_stream_pdu_base.cc b/gnuradio-core/src/lib/io/gr_stream_pdu_base.cc
new file mode 100644
index 000000000..cff7296cb
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_stream_pdu_base.cc
@@ -0,0 +1,117 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_stream_pdu_base.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>
+#include <gr_pdu.h>
+#include <boost/format.hpp>
+
+static const long timeout_us = 100*1000; //100ms
+
+gr_stream_pdu_base::gr_stream_pdu_base (int MTU)
+  : gr_sync_block("stream_pdu_base",
+		  gr_make_io_signature(0, 0, 0),
+		  gr_make_io_signature(0, 0, 0)),
+    d_finished(false), d_started(false), d_fd(-1)
+{
+    // reserve space for rx buffer 
+    d_rxbuf.resize(MTU,0);
+}
+
+gr_stream_pdu_base::~gr_stream_pdu_base()
+{
+    stop_rxthread();
+}
+
+void gr_stream_pdu_base::stop_rxthread(){
+    d_finished = true;
+    if(d_started){
+        d_thread->interrupt();
+        d_thread->join();
+        }
+    }
+
+void gr_stream_pdu_base::start_rxthread(pmt::pmt_t _rxport){
+    rxport = _rxport;
+    d_thread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&gr_stream_pdu_base::run, this)));
+    d_started = true;
+    }
+
+void gr_stream_pdu_base::run(){
+    while(!d_finished) {
+        if(not wait_ready()){ continue; }   
+        const int result = read( d_fd, &d_rxbuf[0], d_rxbuf.size() );
+        if(result <= 0){ throw std::runtime_error("gr_stream_pdu_base, bad socket read!"); }
+        pmt::pmt_t vector = pmt::pmt_init_u8vector(result, &d_rxbuf[0]);       
+        pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector);
+        message_port_pub(rxport, pdu);
+    } 
+}
+
+void gr_stream_pdu_base::send(pmt::pmt_t msg){
+    pmt::pmt_t vector = pmt::pmt_cdr(msg);
+    size_t offset(0);
+    size_t itemsize(gr_pdu_itemsize(type_from_pmt(vector)));
+    int len( pmt::pmt_length(vector)*itemsize );
+    
+    const int rv = write(d_fd, pmt::pmt_uniform_vector_elements(vector, offset), len);
+    if(rv != len){
+        std::cerr << boost::format("WARNING: gr_stream_pdu_base::send(pdu) write failed! (d_fd=%d, len=%d, rv=%d)")
+                            % d_fd % len % rv << std::endl;
+        }
+}
+
+int
+gr_stream_pdu_base::work(int noutput_items,
+		      gr_vector_const_void_star &input_items,
+		      gr_vector_void_star &output_items)
+{
+  throw std::runtime_error("should not be called.\n");
+  return 0; 
+}
+
+bool gr_stream_pdu_base::wait_ready(){
+    //setup timeval for timeout
+    timeval tv;
+    tv.tv_sec = 0;
+    tv.tv_usec = timeout_us;
+    
+    //setup rset for timeout
+    fd_set rset;
+    FD_ZERO(&rset);
+    FD_SET(d_fd, &rset);
+
+    //call select with timeout on receive socket
+    return ::select(d_fd+1, &rset, NULL, NULL, &tv) > 0;
+}
diff --git a/gnuradio-core/src/lib/io/gr_stream_pdu_base.h b/gnuradio-core/src/lib/io/gr_stream_pdu_base.h
new file mode 100644
index 000000000..35bacf523
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_stream_pdu_base.h
@@ -0,0 +1,59 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifndef INCLUDED_GR_STREAM_PDU_BASE_H
+#define INCLUDED_GR_STREAM_PDU_BASE_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+
+/*!
+ * \brief Gather received items into messages and insert into msgq
+ * \ingroup sink_blk
+ */
+class GR_CORE_API gr_stream_pdu_base : public gr_sync_block
+{
+ public:
+  boost::shared_ptr<boost::thread> d_thread;
+  bool d_finished;
+  bool d_started;
+  std::vector<uint8_t> d_rxbuf;
+  void run();
+  int d_fd;
+  gr_stream_pdu_base (int MTU=10000);
+  ~gr_stream_pdu_base ();
+  void send(pmt::pmt_t msg);
+  bool wait_ready();
+  int work (int noutput_items,
+	    gr_vector_const_void_star &input_items,
+	    gr_vector_void_star &output_items);
+  void start_rxthread(pmt::pmt_t _rxport);
+  void stop_rxthread();
+ private:
+  pmt::pmt_t rxport;
+};
+
+typedef boost::shared_ptr<gr_stream_pdu_base> gr_stream_pdu_base_sptr;
+
+#endif /* INCLUDED_GR_TUNTAP_PDU_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..8211b7672
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.cc
@@ -0,0 +1,137 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#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..c3fff3581
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.h
@@ -0,0 +1,76 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#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..f12987b74
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tagged_stream_to_pdu.i
@@ -0,0 +1,31 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+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/gr_tuntap_pdu.cc b/gnuradio-core/src/lib/io/gr_tuntap_pdu.cc
new file mode 100644
index 000000000..8dd4b18a1
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tuntap_pdu.cc
@@ -0,0 +1,145 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_tuntap_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>
+#include <iostream>
+#include <gr_pdu.h>
+#include <boost/format.hpp>
+
+#if (defined(linux) || defined(__linux) || defined(__linux__))
+
+#include <sys/ioctl.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include <arpa/inet.h>
+#include <linux/if.h>
+
+
+// public constructor that returns a shared_ptr
+
+gr_tuntap_pdu_sptr
+gr_make_tuntap_pdu (std::string dev, int MTU)
+{
+  return gnuradio::get_initial_sptr(new gr_tuntap_pdu(dev, MTU));
+}
+
+gr_tuntap_pdu::gr_tuntap_pdu (std::string dev, int MTU)
+    : gr_stream_pdu_base(MTU)
+{
+
+    // make the tuntap
+    char dev_cstr[1024];
+    memset(dev_cstr, 0x00, 1024);
+    strncpy(dev_cstr, dev.c_str(), std::min(sizeof(dev_cstr), dev.size()));
+    d_fd = tun_alloc(dev_cstr);
+    if(d_fd <= 0){
+        throw std::runtime_error("TunTap make: tun_alloc failed (are you running as root?)");
+        }
+
+    std::cout << boost::format(
+        "Allocated virtual ethernet interface: %s\n"
+        "You must now use ifconfig to set its IP address. E.g.,\n"
+        "  $ sudo ifconfig %s 192.168.200.1\n"
+        "Be sure to use a different address in the same subnet for each machine.\n"
+        ) % dev % dev << std::endl;
+
+    // set up output message port
+    message_port_register_out(pmt::mp("pdus"));
+    start_rxthread(pmt::mp("pdus"));
+    
+    // set up input message port
+    message_port_register_in(pmt::mp("pdus"));
+    set_msg_handler(pmt::mp("pdus"), boost::bind(&gr_tuntap_pdu::send, this, _1));
+}
+
+
+int gr_tuntap_pdu::tun_alloc(char *dev, int flags) {
+  struct ifreq ifr;
+  int fd, err;
+  const char *clonedev = "/dev/net/tun";
+
+  /* Arguments taken by the function:
+   *
+   * char *dev: the name of an interface (or '\0'). MUST have enough
+   *   space to hold the interface name if '\0' is passed
+   * int flags: interface flags (eg, IFF_TUN etc.)
+   */
+
+   /* open the clone device */
+   if( (fd = open(clonedev, O_RDWR)) < 0 ) {
+     return fd;
+   }
+
+   /* preparation of the struct ifr, of type "struct ifreq" */
+   memset(&ifr, 0, sizeof(ifr));
+
+   ifr.ifr_flags = flags;   /* IFF_TUN or IFF_TAP, plus maybe IFF_NO_PI */
+
+   if (*dev) {
+     /* if a device name was specified, put it in the structure; otherwise,
+      * the kernel will try to allocate the "next" device of the
+      * specified type */
+     strncpy(ifr.ifr_name, dev, IFNAMSIZ);
+   }
+
+   /* try to create the device */
+   if( (err = ioctl(fd, TUNSETIFF, (void *) &ifr)) < 0 ) {
+     close(fd);
+     return err;
+   }
+
+  /* if the operation was successful, write back the name of the
+   * interface to the variable "dev", so the caller can know
+   * it. Note that the caller MUST reserve space in *dev (see calling
+   * code below) */
+  strcpy(dev, ifr.ifr_name);
+
+  /* this is the special file descriptor that the caller will use to talk
+   * with the virtual interface */
+  return fd;
+}
+
+#else //if not linux
+
+gr_block_sptr
+gr_make_tuntap_pdu(std::string dev, int MTU)
+{
+    gr_block_sptr rv;
+    throw std::runtime_error("tuntap only implemented on linux");
+    return rv;
+}
+
+#endif
diff --git a/gnuradio-core/src/lib/io/gr_tuntap_pdu.h b/gnuradio-core/src/lib/io/gr_tuntap_pdu.h
new file mode 100644
index 000000000..18c83f42b
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tuntap_pdu.h
@@ -0,0 +1,74 @@
+/* -*- c++ -*- */
+/*
+ * 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.
+ */
+
+#ifndef INCLUDED_GR_TUNTAP_PDU_H
+#define INCLUDED_GR_TUNTAP_PDU_H
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+#include <gr_stream_pdu_base.h>
+
+#if (defined(linux) || defined(__linux) || defined(__linux__))
+
+#include <linux/if_tun.h>
+
+class gr_tuntap_pdu;
+typedef boost::shared_ptr<gr_tuntap_pdu> gr_tuntap_pdu_sptr;
+
+GR_CORE_API gr_tuntap_pdu_sptr gr_make_tuntap_pdu (std::string dev, int MTU=10000);
+
+/*!
+ * \brief Gather received items into messages and insert into msgq
+ * \ingroup sink_blk
+ */
+class GR_CORE_API gr_tuntap_pdu : public gr_stream_pdu_base
+{
+ private:
+  friend GR_CORE_API gr_tuntap_pdu_sptr
+  gr_make_tuntap_pdu(std::string dev, int MTU);
+  int tun_alloc(char* dev, int flags = IFF_TAP | IFF_NO_PI);
+  std::string d_dev;
+ protected:
+  gr_tuntap_pdu (std::string dev, int MTU=10000);
+
+ public:
+  ~gr_tuntap_pdu () {}
+
+};
+
+#else // if not linux
+
+class gr_tuntap_pdu
+{
+private:
+  gr_tuntap_pdu() {};
+public:
+  ~gr_tuntap_pdu() {};
+};
+
+GR_CORE_API gr_block_sptr gr_make_tuntap_pdu (std::string dev, int MTU=0);
+
+#endif
+
+#endif /* INCLUDED_GR_TUNTAP_PDU_H */
diff --git a/gnuradio-core/src/lib/io/gr_tuntap_pdu.i b/gnuradio-core/src/lib/io/gr_tuntap_pdu.i
new file mode 100644
index 000000000..589bbc385
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tuntap_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.
+ */
+
+GR_SWIG_BLOCK_MAGIC(gr,tuntap_pdu);
+
+%{
+#include <gr_tuntap_pdu.h>
+%}
+
+%include "gr_tuntap_pdu.h"
+
diff --git a/gnuradio-core/src/lib/io/io.i b/gnuradio-core/src/lib/io/io.i
index 07f2e9195..055f28847 100644
--- a/gnuradio-core/src/lib/io/io.i
+++ b/gnuradio-core/src/lib/io/io.i
@@ -46,6 +46,12 @@
 #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_tuntap_pdu.h>
+#include <gr_socket_pdu.h>
 %}
 
 %include "gr_file_sink_base.i"
@@ -69,4 +75,11 @@
 %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"
+%include "gr_tuntap_pdu.i"
+%include "gr_socket_pdu.i"
+
 
diff --git a/gnuradio-core/src/lib/runtime/CMakeLists.txt b/gnuradio-core/src/lib/runtime/CMakeLists.txt
index 5f3672dde..70938a0f1 100644
--- a/gnuradio-core/src/lib/runtime/CMakeLists.txt
+++ b/gnuradio-core/src/lib/runtime/CMakeLists.txt
@@ -54,6 +54,7 @@ list(APPEND gnuradio_core_sources
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_detail.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_executor.cc
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_registry.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_hier_block2.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_hier_block2_detail.cc
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_buffer.cc
@@ -116,6 +117,7 @@ install(FILES
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_detail.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_executor.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/gr_block_registry.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_hier_block2.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_hier_block2_detail.h
     ${CMAKE_CURRENT_SOURCE_DIR}/gr_buffer.h
diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.cc b/gnuradio-core/src/lib/runtime/gr_basic_block.cc
index 3d809aa8b..6ff57a1d6 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.cc
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2006,2012 Free Software Foundation, Inc.
  *
  * This file is part of GNU Radio
  *
@@ -25,7 +25,10 @@
 #endif
 
 #include <gr_basic_block.h>
+#include <gr_block_registry.h>
 #include <stdexcept>
+#include <sstream>
+#include <iostream>
 
 using namespace pmt;
 
@@ -35,7 +38,7 @@ static long s_ncurrently_allocated = 0;
 long
 gr_basic_block_ncurrently_allocated()
 {
-    return s_ncurrently_allocated;
+  return s_ncurrently_allocated;
 }
 
 gr_basic_block::gr_basic_block(const std::string &name,
@@ -45,20 +48,180 @@ gr_basic_block::gr_basic_block(const std::string &name,
     d_input_signature(input_signature),
     d_output_signature(output_signature),
     d_unique_id(s_next_id++),
+    d_symbolic_id(global_block_registry.block_register(this)),
+    d_symbol_name(global_block_registry.register_symbolic_name(this)),
     d_color(WHITE),
-    d_max_output_buffer(std::max(output_signature->max_streams(),1), -1),
-    d_min_output_buffer(std::max(output_signature->max_streams(),1), -1)
+    message_subscribers(pmt::pmt_make_dict())
 {
-    s_ncurrently_allocated++;
+  s_ncurrently_allocated++;
 }
 
 gr_basic_block::~gr_basic_block()
 {
-    s_ncurrently_allocated--;
+  s_ncurrently_allocated--;
+  global_block_registry.block_unregister(this);
 }
 
 gr_basic_block_sptr
 gr_basic_block::to_basic_block()
 {
-    return shared_from_this();
+  return shared_from_this();
+}
+
+void
+gr_basic_block::set_block_alias(std::string name)
+{ 
+  global_block_registry.register_symbolic_name(this, name); 
+}
+
+// ** Message passing interface **
+
+//  - register a new input message port
+void
+gr_basic_block::message_port_register_in(pmt::pmt_t port_id)
+{
+  if(!pmt::pmt_is_symbol(port_id)) {
+    throw std::runtime_error("message_port_register_in: bad port id");
+  }
+  msg_queue[port_id] = msg_queue_t();
+  msg_queue_ready[port_id] = boost::shared_ptr<boost::condition_variable>(new boost::condition_variable());
+}
+
+pmt::pmt_t
+gr_basic_block::message_ports_in()
+{
+  pmt::pmt_t port_names = pmt::pmt_make_vector(msg_queue.size(), pmt::PMT_NIL);
+  msg_queue_map_itr itr = msg_queue.begin();
+  for(size_t i = 0; i < msg_queue.size(); i++) {
+    pmt::pmt_vector_set(port_names, i, (*itr).first);
+    itr++;
+  }
+  return port_names;
+}
+
+//  - register a new output message port
+void
+gr_basic_block::message_port_register_out(pmt::pmt_t port_id)
+{
+  if(!pmt::pmt_is_symbol(port_id)) {
+    throw std::runtime_error("message_port_register_out: bad port id");
+  }
+  if(pmt::pmt_dict_has_key(message_subscribers, port_id)) {
+    throw std::runtime_error("message_port_register_out: port already in use");
+  }
+  message_subscribers = pmt::pmt_dict_add(message_subscribers, port_id, pmt::PMT_NIL);
+}
+
+pmt::pmt_t
+gr_basic_block::message_ports_out()
+{
+  size_t len = pmt::pmt_length(message_subscribers);
+  pmt::pmt_t port_names = pmt::pmt_make_vector(len, pmt::PMT_NIL);
+  pmt::pmt_t keys = pmt::pmt_dict_keys(message_subscribers);
+  for(size_t i = 0; i < len; i++) {
+    pmt::pmt_vector_set(port_names, i, pmt::pmt_nth(i, keys));
+  }
+  return port_names;
+}
+
+//  - publish a message on a message port
+void gr_basic_block::message_port_pub(pmt::pmt_t port_id, pmt::pmt_t msg)
+{
+  if(!pmt::pmt_dict_has_key(message_subscribers, port_id)) {
+    throw std::runtime_error("port does not exist");
+  }
+  
+  pmt::pmt_t currlist = pmt::pmt_dict_ref(message_subscribers, port_id, pmt::PMT_NIL);
+  // iterate through subscribers on port
+  while(pmt::pmt_is_pair(currlist)) {
+    pmt::pmt_t target = pmt::pmt_car(currlist);
+
+    pmt::pmt_t block = pmt::pmt_car(target);
+    pmt::pmt_t port = pmt::pmt_cdr(target);
+    
+    currlist = pmt::pmt_cdr(currlist);
+    gr_basic_block_sptr blk = global_block_registry.block_lookup(block);
+    //blk->post(msg);
+    blk->post(port, msg);
+  }
+}
+
+//  - subscribe to a message port
+void
+gr_basic_block::message_port_sub(pmt::pmt_t port_id, pmt::pmt_t target){
+  if(!pmt::pmt_dict_has_key(message_subscribers, port_id)){ 
+    std::stringstream ss;
+    ss << "Port does not exist: \"" << pmt::pmt_write_string(port_id) << "\" on block: " << pmt::pmt_write_string(target) << std::endl;
+    throw std::runtime_error(ss.str());
+  }
+  pmt::pmt_t currlist = pmt::pmt_dict_ref(message_subscribers,port_id,pmt::PMT_NIL);
+  
+  // ignore re-adds of the same target
+  if(!pmt::pmt_list_has(currlist, target))
+    message_subscribers = pmt::pmt_dict_add(message_subscribers,port_id,pmt::pmt_list_add(currlist,target));
+}
+
+void
+gr_basic_block::message_port_unsub(pmt::pmt_t port_id, pmt::pmt_t target){
+  if(!pmt::pmt_dict_has_key(message_subscribers, port_id)){ 
+    std::stringstream ss;
+    ss << "Port does not exist: \"" << pmt::pmt_write_string(port_id) << "\" on block: " << pmt::pmt_write_string(target) << std::endl;
+    throw std::runtime_error(ss.str());
+  }
+  
+  // ignore unsubs of unknown targets
+  pmt::pmt_t currlist = pmt::pmt_dict_ref(message_subscribers,port_id,pmt::PMT_NIL);
+  message_subscribers = pmt::pmt_dict_add(message_subscribers,port_id,pmt::pmt_list_rm(currlist,target));
+}
+
+void
+gr_basic_block::_post(pmt_t which_port, pmt_t msg)
+{
+  insert_tail(which_port, msg);
+}
+
+void
+gr_basic_block::insert_tail(pmt::pmt_t which_port, pmt::pmt_t msg)
+{
+  gruel::scoped_lock guard(mutex);
+
+    if( (msg_queue.find(which_port) == msg_queue.end()) || (msg_queue_ready.find(which_port) == msg_queue_ready.end())){
+        std::cout << "target port = " << pmt::pmt_symbol_to_string(which_port) << std::endl;
+        throw std::runtime_error("attempted to insert_tail on invalid queue!");
+    }
+
+  msg_queue[which_port].push_back(msg);
+  msg_queue_ready[which_port]->notify_one();
+
+  // wake up thread if BLKD_IN or BLKD_OUT
+  global_block_registry.notify_blk(alias());
+}
+
+pmt_t
+gr_basic_block::delete_head_nowait(pmt::pmt_t which_port)
+{
+  gruel::scoped_lock guard(mutex);
+
+  if (empty_p(which_port)){
+    return pmt::pmt_t();
+    }
+
+  pmt_t m(msg_queue[which_port].front());
+  msg_queue[which_port].pop_front();
+
+  return m;
+}
+
+pmt_t
+gr_basic_block::delete_head_blocking(pmt::pmt_t which_port)
+{
+  gruel::scoped_lock guard(mutex);
+
+  while (empty_p(which_port)){
+    msg_queue_ready[which_port]->wait(guard);
+    }
+
+  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 b8e8148b2..f3b7b835b 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.h
@@ -29,8 +29,13 @@
 #include <boost/enable_shared_from_this.hpp>
 #include <boost/function.hpp>
 #include <gr_msg_accepter.h>
-#include <gr_io_signature.h>
 #include <string>
+#include <deque>
+#include <map>
+#include <gr_io_signature.h>
+#include <gruel/thread.h>
+#include <boost/foreach.hpp>
+#include <boost/thread/condition_variable.hpp>
 #include <iostream>
 
 /*!
@@ -46,183 +51,215 @@
 
 class GR_CORE_API gr_basic_block : public gr_msg_accepter, public boost::enable_shared_from_this<gr_basic_block>
 {
-    typedef boost::function<void(pmt::pmt_t)> msg_handler_t;
-
-private:
-    /*
-     * This function is called by the runtime system to dispatch messages.
-     *
-     * The thread-safety guarantees mentioned in set_msg_handler are implemented
-     * by the callers of this method.
-     */
-    void dispatch_msg(pmt::pmt_t msg)
-    {
-      if (d_msg_handler)	// Is there a handler?
-	d_msg_handler(msg);	// Yes, invoke it.
-    };
-
-    msg_handler_t	 d_msg_handler;
-
-    /*
-     * Used to expand the vectors that hold the min/max buffer sizes.
-     *
-     * Specifically, when -1 is used, the vectors are just initialized
-     * with 1 value; this is used by the flat_flowgraph to expand when
-     * required to add a new value for new ports on these blocks.
-     */
-    void expand_minmax_buffer(int port) {
-      if((size_t)port >= d_max_output_buffer.size())
-	set_max_output_buffer(port, -1);
-      if((size_t)port >= d_min_output_buffer.size())
-	set_min_output_buffer(port, -1);
-    }
-    
-protected:
-    friend class gr_flowgraph;
-    friend class gr_flat_flowgraph; // TODO: will be redundant
-    friend class gr_tpb_thread_body;
-
-    enum vcolor { WHITE, GREY, BLACK };
-
-    std::string          d_name;
-    gr_io_signature_sptr d_input_signature;
-    gr_io_signature_sptr d_output_signature;
-    long                 d_unique_id;
-    vcolor               d_color;
-    std::vector<long>    d_max_output_buffer;
-    std::vector<long>    d_min_output_buffer;
-
-    gr_basic_block(void){} //allows pure virtual interface sub-classes
-
-    //! Protected constructor prevents instantiation by non-derived classes
-    gr_basic_block(const std::string &name,
-                   gr_io_signature_sptr input_signature,
-                   gr_io_signature_sptr output_signature);
-
-    //! may only be called during constructor
-    void set_input_signature(gr_io_signature_sptr iosig) {
-        d_input_signature = iosig;
-    }
-
-    //! may only be called during constructor
-    void set_output_signature(gr_io_signature_sptr iosig) {
-        d_output_signature = iosig;
-    }
+  typedef boost::function<void(pmt::pmt_t)> msg_handler_t;
+  
+ private:
+  /*
+   * This function is called by the runtime system to dispatch messages.
+   *
+   * The thread-safety guarantees mentioned in set_msg_handler are implemented
+   * by the callers of this method.
+   */
+  void dispatch_msg(pmt::pmt_t which_port, pmt::pmt_t msg)
+  {
+    // AA Update this
+    if (d_msg_handlers.find(which_port) != d_msg_handlers.end()) // Is there a handler?
+      d_msg_handlers[which_port](msg); // Yes, invoke it.
+  };
+  
+  //msg_handler_t	 d_msg_handler;
+  typedef std::map<pmt::pmt_t , msg_handler_t, pmt::pmt_comperator> d_msg_handlers_t;
+  d_msg_handlers_t d_msg_handlers;
+  
+  typedef std::deque<pmt::pmt_t>    msg_queue_t;
+  typedef std::map<pmt::pmt_t, msg_queue_t, pmt::pmt_comperator>    msg_queue_map_t;
+  typedef std::map<pmt::pmt_t, msg_queue_t, pmt::pmt_comperator>::iterator msg_queue_map_itr;
+  std::map<pmt::pmt_t, boost::shared_ptr<boost::condition_variable>, pmt::pmt_comperator> msg_queue_ready;
+  
+  gruel::mutex          mutex;          //< protects all vars
+  
+ protected:
+  friend class gr_flowgraph;
+  friend class gr_flat_flowgraph; // TODO: will be redundant
+  friend class gr_tpb_thread_body;
+  
+  enum vcolor { WHITE, GREY, BLACK };
+  
+  std::string          d_name;
+  gr_io_signature_sptr d_input_signature;
+  gr_io_signature_sptr d_output_signature;
+  long                 d_unique_id;
+  long                 d_symbolic_id;
+  std::string          d_symbol_name;
+  std::string          d_symbol_alias;
+  vcolor               d_color;
+  msg_queue_map_t msg_queue;
+  
+  gr_basic_block(void){} //allows pure virtual interface sub-classes
+  
+  //! Protected constructor prevents instantiation by non-derived classes
+  gr_basic_block(const std::string &name,
+		 gr_io_signature_sptr input_signature,
+		 gr_io_signature_sptr output_signature);
+  
+  //! may only be called during constructor
+  void set_input_signature(gr_io_signature_sptr iosig) {
+    d_input_signature = iosig;
+  }
+  
+  //! may only be called during constructor
+  void set_output_signature(gr_io_signature_sptr iosig) {
+    d_output_signature = iosig;
+  }
+  
+  /*!
+   * \brief Allow the flowgraph to set for sorting and partitioning
+   */
+  void set_color(vcolor color) { d_color = color; }
+  vcolor color() const { return d_color; }
+  
+  // Message passing interface
+  pmt::pmt_t message_subscribers;
+  
+ public:
+  virtual ~gr_basic_block();
+  long unique_id() const { return d_unique_id; }
+  long symbolic_id() const { return d_symbolic_id; }
+  std::string name() const { return d_name; }
+  std::string symbol_name() const { return d_symbol_name; }
+  gr_io_signature_sptr input_signature() const  { return d_input_signature; }
+  gr_io_signature_sptr output_signature() const { return d_output_signature; }
+  gr_basic_block_sptr to_basic_block(); // Needed for Python type coercion
+  bool alias_set() { return !d_symbol_alias.empty(); }
+  std::string alias(){ return alias_set()?d_symbol_alias:symbol_name(); }
+  pmt::pmt_t alias_pmt(){ return pmt::pmt_intern(alias()); }
+  void set_block_alias(std::string name);
+  
+  // ** Message passing interface **
+  void message_port_register_in(pmt::pmt_t port_id);
+  void message_port_register_out(pmt::pmt_t port_id);
+  void message_port_pub(pmt::pmt_t port_id, pmt::pmt_t msg);
+  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);
+  
+  virtual bool message_port_is_hier(pmt::pmt_t port_id) { std::cout << "is_hier\n"; return false; }
+  virtual bool message_port_is_hier_in(pmt::pmt_t port_id) { std::cout << "is_hier_in\n"; return false; }
+  virtual bool message_port_is_hier_out(pmt::pmt_t port_id) { std::cout << "is_hier_out\n"; return false; }
+  
+  /*!
+   * \brief Get input message port names.
+   *
+   * Returns the available input message ports for a block. The
+   * return object is a PMT vector that is filled with PMT symbols.
+   */
+  pmt::pmt_t message_ports_in();
+  
+  /*!
+   * \brief Get output message port names.
+   *
+   * Returns the available output message ports for a block. The
+   * return object is a PMT vector that is filled with PMT symbols.
+   */
+  pmt::pmt_t message_ports_out();
+  
+  /*!
+   * 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) { 
+    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(); }
+    return rv;
+  }
+  
+  //| Acquires and release the mutex
+  void insert_tail( pmt::pmt_t which_port, pmt::pmt_t msg);
+  /*!
+   * \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.
+   */
+  pmt::pmt_t delete_head_blocking( pmt::pmt_t which_port);
+  
+  msg_queue_t::iterator get_iterator(pmt::pmt_t which_port){
+    return msg_queue[which_port].begin();
+  }
 
-    /*!
-     * \brief Allow the flowgraph to set for sorting and partitioning
-     */
-    void set_color(vcolor color) { d_color = color; }
-    vcolor color() const { return d_color; }
-
-public:
-    virtual ~gr_basic_block();
-    long unique_id() const { return d_unique_id; }
-    std::string name() const { return d_name; }
-    gr_io_signature_sptr input_signature() const  { return d_input_signature; }
-    gr_io_signature_sptr output_signature() const { return d_output_signature; }
-    gr_basic_block_sptr to_basic_block(); // Needed for Python type coercion
-
-    /*!
-     * \brief Returns max buffer size on output port \p i.
-     */
-    long max_output_buffer(size_t i) {
-      if(i >= d_max_output_buffer.size())
-	throw std::invalid_argument("gr_basic_block::max_output_buffer: port out of range.");
-      return d_max_output_buffer[i];
-    }
-
-    /*!
-     * \brief Sets max buffer size on all output ports.
-     */
-    void set_max_output_buffer(long max_output_buffer) { 
-      for(int i=0; i<output_signature()->max_streams(); i++) {
-	set_max_output_buffer(i, max_output_buffer);
-      }
-    }
-
-    /*!
-     * \brief Sets max buffer size on output port \p port.
-     */
-    void set_max_output_buffer(int port, long max_output_buffer) {
-      if((size_t)port >= d_max_output_buffer.size())
-	d_max_output_buffer.push_back(max_output_buffer);
-      else
-	d_max_output_buffer[port] = max_output_buffer; 
-    }
-
-    /*!
-     * \brief Returns min buffer size on output port \p i.
-     */
-    long min_output_buffer(size_t i) {
-      if(i >= d_min_output_buffer.size())
-	throw std::invalid_argument("gr_basic_block::min_output_buffer: port out of range.");
-      return d_min_output_buffer[i];
-    }
-
-    /*!
-     * \brief Sets min buffer size on all output ports.
-     */
-    void set_min_output_buffer(long min_output_buffer) {
-      for(int i=0; i<output_signature()->max_streams(); i++) {
-	set_min_output_buffer(i, min_output_buffer);
-      }
+  void erase_msg(pmt::pmt_t which_port, msg_queue_t::iterator it){
+    msg_queue[which_port].erase(it);
+  }
+  
+  virtual bool has_msg_port(pmt::pmt_t which_port){
+    if(msg_queue.find(which_port) != msg_queue.end()){
+      return true;
     }
-
-    /*!
-     * \brief Sets min buffer size on output port \p port.
-     */
-    void set_min_output_buffer(int port, long min_output_buffer) {
-      if((size_t)port >= d_min_output_buffer.size())
-	d_min_output_buffer.push_back(min_output_buffer);
-      else
-	d_min_output_buffer[port] = min_output_buffer; 
-    }
-
-    /*!
-     * \brief Confirm that ninputs and noutputs is an acceptable combination.
-     *
-     * \param ninputs	number of input streams connected
-     * \param noutputs	number of output streams connected
-     *
-     * \returns true if this is a valid configuration for this block.
-     *
-     * This function is called by the runtime system whenever the
-     * topology changes.  Most classes do not need to override this.
-     * This check is in addition to the constraints specified by the input
-     * and output gr_io_signatures.
-     */
-    virtual bool check_topology(int ninputs, int noutputs) { return true; }
-
-    /*!
-     * \brief Set the callback that is fired when messages are available.
-     *
-     * \p msg_handler can be any kind of function pointer or function object
-     * that has the signature:
-     * <pre>
-     *    void msg_handler(pmt::pmt msg);
-     * </pre>
-     *
-     * (You may want to use boost::bind to massage your callable into the
-     * correct form.  See gr_nop.{h,cc} for an example that sets up a class
-     * method as the callback.)
-     *
-     * Blocks that desire to handle messages must call this method in their
-     * constructors to register the handler that will be invoked when messages
-     * are available.
-     *
-     * If the block inherits from gr_block, the runtime system will ensure that
-     * msg_handler is called in a thread-safe manner, such that work and
-     * msg_handler will never be called concurrently.  This allows msg_handler
-     * to update state variables without having to worry about thread-safety
-     * issues with work, general_work or another invocation of msg_handler.
-     *
-     * If the block inherits from gr_hier_block2, the runtime system will
-     * ensure that no reentrant calls are made to msg_handler.
-     */
-    template <typename T> void set_msg_handler(T msg_handler){
-      d_msg_handler = msg_handler_t(msg_handler);
+    if(pmt::pmt_dict_has_key(message_subscribers, which_port)){
+      return true;
     }
+    return false;
+  }
+  
+  
+  /*!
+   * \brief Confirm that ninputs and noutputs is an acceptable combination.
+   *
+   * \param ninputs	number of input streams connected
+   * \param noutputs	number of output streams connected
+   *
+   * \returns true if this is a valid configuration for this block.
+   *
+   * This function is called by the runtime system whenever the
+   * topology changes.  Most classes do not need to override this.
+   * This check is in addition to the constraints specified by the input
+   * and output gr_io_signatures.
+   */
+  virtual bool check_topology(int ninputs, int noutputs) { return true; }
+  
+  /*!
+   * \brief Set the callback that is fired when messages are available.
+   *
+   * \p msg_handler can be any kind of function pointer or function object
+   * that has the signature:
+   * <pre>
+   *    void msg_handler(pmt::pmt msg);
+   * </pre>
+   *
+   * (You may want to use boost::bind to massage your callable into the
+   * correct form.  See gr_nop.{h,cc} for an example that sets up a class
+   * method as the callback.)
+   *
+   * Blocks that desire to handle messages must call this method in their
+   * constructors to register the handler that will be invoked when messages
+   * are available.
+   *
+   * If the block inherits from gr_block, the runtime system will ensure that
+   * msg_handler is called in a thread-safe manner, such that work and
+   * msg_handler will never be called concurrently.  This allows msg_handler
+   * to update state variables without having to worry about thread-safety
+   * issues with work, general_work or another invocation of msg_handler.
+   *
+   * If the block inherits from gr_hier_block2, the runtime system will
+   * ensure that no reentrant calls are made to msg_handler.
+   */
+  //template <typename T> void set_msg_handler(T msg_handler){
+  //  d_msg_handler = msg_handler_t(msg_handler);
+  //}
+  template <typename T> void set_msg_handler(pmt::pmt_t which_port, T msg_handler){
+    if(msg_queue.find(which_port) == msg_queue.end()){ 
+      throw std::runtime_error("attempt to set_msg_handler() on bad input message port!"); }
+    d_msg_handlers[which_port] = msg_handler_t(msg_handler);
+  }
 };
 
 inline bool operator<(gr_basic_block_sptr lhs, gr_basic_block_sptr rhs)
@@ -237,8 +274,8 @@ GR_CORE_API long gr_basic_block_ncurrently_allocated();
 
 inline std::ostream &operator << (std::ostream &os, gr_basic_block_sptr basic_block)
 {
-    os << basic_block->name() << "(" << basic_block->unique_id() << ")";
-    return os;
+  os << basic_block->name() << "(" << basic_block->unique_id() << ")";
+  return os;
 }
 
 #endif /* INCLUDED_GR_BASIC_BLOCK_H */
diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.i b/gnuradio-core/src/lib/runtime/gr_basic_block.i
index 7713f2fe1..62f16462d 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 {
@@ -37,17 +39,17 @@ protected:
 public:
     virtual ~gr_basic_block();
     std::string name() const;
+    std::string symbol_name() const;
     gr_io_signature_sptr input_signature() const;
     gr_io_signature_sptr output_signature() const;
     long unique_id() const;
     gr_basic_block_sptr to_basic_block();
     bool check_topology (int ninputs, int noutputs);
-    long max_output_buffer(int i);
-    void set_max_output_buffer(long max_output_buffer);
-    void set_max_output_buffer(int port, long max_output_buffer);
-    long min_output_buffer(int i);
-    void set_min_output_buffer(long min_output_buffer);
-    void set_min_output_buffer(int port, long min_output_buffer);
+    std::string alias();
+    void set_block_alias(std::string name);
+    void _post(pmt_t which_port, pmt_t msg);
+    pmt_t message_ports_in();
+    pmt_t message_ports_out();
 };
 
 %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 7e01c0ba8..43aebf0bf 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,
@@ -42,12 +43,16 @@ gr_block::gr_block (const std::string &name,
     d_fixed_rate(false),
     d_max_noutput_items_set(false),
     d_max_noutput_items(0),
-    d_tag_propagation_policy(TPP_ALL_TO_ALL)
+    d_tag_propagation_policy(TPP_ALL_TO_ALL),
+    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
@@ -246,3 +251,12 @@ operator << (std::ostream& os, const gr_block *m)
   return os;
 }
 
+int
+gr_block::general_work(int noutput_items,
+		       gr_vector_int &ninput_items,
+		       gr_vector_const_void_star &input_items,
+		       gr_vector_void_star &output_items)
+{
+  throw std::runtime_error("gr_block::general_work() not implemented");
+  return 0;
+}
diff --git a/gnuradio-core/src/lib/runtime/gr_block.h b/gnuradio-core/src/lib/runtime/gr_block.h
index c89138bb3..57e3fda90 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_block.h
@@ -124,7 +124,7 @@ class GR_CORE_API gr_block : public gr_basic_block {
   virtual int general_work (int noutput_items,
 			    gr_vector_int &ninput_items,
 			    gr_vector_const_void_star &input_items,
-			    gr_vector_void_star &output_items) = 0;
+			    gr_vector_void_star &output_items);
 
   /*!
    * \brief Called to enable drivers, etc for i/o devices.
@@ -288,6 +288,76 @@ class GR_CORE_API gr_block : public gr_basic_block {
    */
   bool is_set_max_noutput_items();
 
+  /*
+   * Used to expand the vectors that hold the min/max buffer sizes.
+   *
+   * Specifically, when -1 is used, the vectors are just initialized
+   * with 1 value; this is used by the flat_flowgraph to expand when
+   * required to add a new value for new ports on these blocks.
+   */
+  void expand_minmax_buffer(int port) {
+    if((size_t)port >= d_max_output_buffer.size())
+      set_max_output_buffer(port, -1);
+    if((size_t)port >= d_min_output_buffer.size())
+      set_min_output_buffer(port, -1);
+  }
+
+  /*!
+   * \brief Returns max buffer size on output port \p i.
+   */
+  long max_output_buffer(size_t i) {
+    if(i >= d_max_output_buffer.size())
+      throw std::invalid_argument("gr_basic_block::max_output_buffer: port out of range.");
+    return d_max_output_buffer[i];
+  }
+
+  /*!
+   * \brief Sets max buffer size on all output ports.
+   */
+  void set_max_output_buffer(long max_output_buffer) { 
+    for(int i = 0; i < output_signature()->max_streams(); i++) {
+      set_max_output_buffer(i, max_output_buffer);
+    }
+  }
+
+  /*!
+   * \brief Sets max buffer size on output port \p port.
+   */
+  void set_max_output_buffer(int port, long max_output_buffer) {
+    if((size_t)port >= d_max_output_buffer.size())
+      d_max_output_buffer.push_back(max_output_buffer);
+    else
+      d_max_output_buffer[port] = max_output_buffer; 
+  }
+
+  /*!
+   * \brief Returns min buffer size on output port \p i.
+   */
+  long min_output_buffer(size_t i) {
+    if(i >= d_min_output_buffer.size())
+      throw std::invalid_argument("gr_basic_block::min_output_buffer: port out of range.");
+    return d_min_output_buffer[i];
+  }
+
+  /*!
+   * \brief Sets min buffer size on all output ports.
+   */
+  void set_min_output_buffer(long min_output_buffer) {
+    for(int i=0; i<output_signature()->max_streams(); i++) {
+      set_min_output_buffer(i, min_output_buffer);
+    }
+  }
+
+  /*!
+   * \brief Sets min buffer size on output port \p port.
+   */
+  void set_min_output_buffer(int port, long min_output_buffer) {
+    if((size_t)port >= d_min_output_buffer.size())
+      d_min_output_buffer.push_back(min_output_buffer);
+    else
+      d_min_output_buffer[port] = min_output_buffer; 
+  }
+
   // ----------------------------------------------------------------------------
 
  private:
@@ -384,6 +454,10 @@ class GR_CORE_API gr_block : public gr_basic_block {
 			 uint64_t abs_end,
 			 const pmt::pmt_t &key);
 
+  std::vector<long>    d_max_output_buffer;
+  std::vector<long>    d_min_output_buffer;
+
+
   // These are really only for internal use, but leaving them public avoids
   // having to work up an ever-varying list of friend GR_CORE_APIs
 
diff --git a/gnuradio-core/src/lib/runtime/gr_block.i b/gnuradio-core/src/lib/runtime/gr_block.i
index e9341e8cb..db6c1d04a 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.i
+++ b/gnuradio-core/src/lib/runtime/gr_block.i
@@ -58,6 +58,14 @@ class gr_block : public gr_basic_block {
   void unset_max_noutput_items();
   bool is_set_max_noutput_items();
 
+  // Methods to manage block's min/max buffer sizes.
+  long max_output_buffer(int i);
+  void set_max_output_buffer(long max_output_buffer);
+  void set_max_output_buffer(int port, long max_output_buffer);
+  long min_output_buffer(int i);
+  void set_min_output_buffer(long min_output_buffer);
+  void set_min_output_buffer(int port, long min_output_buffer);
+
   // internal use
   gr_block_detail_sptr detail () const { return d_detail; }
   void set_detail (gr_block_detail_sptr detail) { d_detail = detail; }
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.cc b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
index 2792cd471..337c9518e 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
@@ -127,12 +127,6 @@ gr_block_detail::produce_each (int how_many_items)
 }
 
 
-void
-gr_block_detail::_post(pmt_t msg)
-{
-  d_tpb.insert_tail(msg);
-}
-
 uint64_t
 gr_block_detail::nitems_read(unsigned int which_input)
 {
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.h b/gnuradio-core/src/lib/runtime/gr_block_detail.h
index c96f00db8..16d9f4d42 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.h
@@ -85,11 +85,6 @@ class GR_CORE_API gr_block_detail {
    */
   void produce_each (int how_many_items);
 
-  /*!
-   * Accept msg, place in queue, arrange for thread to be awakened if it's not already.
-   */
-  void _post(pmt::pmt_t msg);
-
   // Return the number of items read on input stream which_input
   uint64_t nitems_read(unsigned int which_input);
 
diff --git a/gnuradio-core/src/lib/runtime/gr_block_registry.cc b/gnuradio-core/src/lib/runtime/gr_block_registry.cc
new file mode 100644
index 000000000..ff23d97eb
--- /dev/null
+++ b/gnuradio-core/src/lib/runtime/gr_block_registry.cc
@@ -0,0 +1,76 @@
+#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;
+
+gr_block_registry::gr_block_registry(){
+    d_ref_map = pmt::pmt_make_dict();
+}
+
+long gr_block_registry::block_register(gr_basic_block* block){
+    if(d_map.find(block->name()) == d_map.end()){
+        d_map[block->name()] = blocksubmap_t();
+        d_map[block->name()][0] = block;
+        return 0;
+    } else {
+        for(size_t i=0; i<=d_map[block->name()].size(); i++){
+            if(d_map[block->name()].find(i) == d_map[block->name()].end()){
+                d_map[block->name()][i] = block;
+                return i;
+            }
+        }
+    }
+    throw std::runtime_error("should not reach this");
+}
+
+void gr_block_registry::block_unregister(gr_basic_block* block){
+    d_map[block->name()].erase( d_map[block->name()].find(block->symbolic_id()));
+    d_ref_map = pmt::pmt_dict_delete(d_ref_map, pmt::pmt_intern(block->symbol_name()));
+    if(block->alias_set()){
+        d_ref_map = pmt::pmt_dict_delete(d_ref_map, pmt::pmt_intern(block->alias()));
+        }
+}
+
+std::string gr_block_registry::register_symbolic_name(gr_basic_block* block){
+    std::stringstream ss;
+    ss << block->name() << block->symbolic_id();
+    //std::cout << "register_symbolic_name: " << ss.str() << std::endl;
+    register_symbolic_name(block, ss.str());
+    return ss.str();
+}
+
+void gr_block_registry::register_symbolic_name(gr_basic_block* block, std::string name){
+    if(pmt_dict_has_key(d_ref_map, pmt::pmt_intern(name))){
+        throw std::runtime_error("symbol already exists, can not re-use!");
+        }
+    d_ref_map = pmt_dict_add(d_ref_map, pmt::pmt_intern(name), pmt::pmt_make_any(block));
+}
+
+gr_basic_block_sptr gr_block_registry::block_lookup(pmt::pmt_t symbol){
+    pmt::pmt_t ref = pmt_dict_ref(d_ref_map, symbol, pmt::PMT_NIL);
+    if(pmt::pmt_eq(ref, pmt::PMT_NIL)){
+        throw std::runtime_error("block lookup failed! block not found!");
+    }
+    gr_basic_block* blk = boost::any_cast<gr_basic_block*>( pmt::pmt_any_ref(ref) );
+    return blk->shared_from_this();
+}
+
+
+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
new file mode 100644
index 000000000..6a2d939e5
--- /dev/null
+++ b/gnuradio-core/src/lib/runtime/gr_block_registry.h
@@ -0,0 +1,42 @@
+#ifndef GR_BLOCK_REGISTRY_H
+#define GR_BLOCK_REGISTRY_H
+
+#include <map>
+
+#ifndef GR_BASIC_BLOCK_H
+class gr_basic_block;
+class gr_block;
+#endif
+
+class gr_block_registry {
+    public:
+        gr_block_registry();
+
+        long block_register(gr_basic_block* block);
+        void block_unregister(gr_basic_block* block);
+
+        std::string register_symbolic_name(gr_basic_block* block);
+        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;
+        typedef std::map< long, gr_basic_block* >   blocksubmap_t;
+        typedef std::map< std::string, blocksubmap_t >  blockmap_t;
+
+        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;
+
+#endif
+
diff --git a/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.cc b/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.cc
index b70135b85..9294a5dca 100644
--- a/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.cc
+++ b/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.cc
@@ -31,8 +31,9 @@
 #include <volk/volk.h>
 #include <iostream>
 #include <map>
+#include <boost/format.hpp>
 
-#define GR_FLAT_FLOWGRAPH_DEBUG 0
+#define GR_FLAT_FLOWGRAPH_DEBUG  0
 
 // 32Kbyte buffer size between blocks
 #define GR_FIXED_BUFFER_SIZE (32*(1L<<10))
@@ -71,6 +72,15 @@ gr_flat_flowgraph::setup_connections()
     block->set_is_unaligned(false);
   }
 
+  // Connect message ports connetions
+  for(gr_msg_edge_viter_t i = d_msg_edges.begin(); i != d_msg_edges.end(); i++){
+    if(GR_FLAT_FLOWGRAPH_DEBUG)
+        std::cout << boost::format("flat_fg connecting msg primitives: (%s, %s)->(%s, %s)\n") %
+                    i->src().block() % i->src().port() %
+                    i->dst().block() % i->dst().port();
+    i->src().block()->message_port_sub( i->src().port(), pmt::pmt_cons(i->dst().block()->alias_pmt(), i->dst().port()) );
+    }
+
 }
 
 gr_block_detail_sptr
@@ -80,18 +90,23 @@ gr_flat_flowgraph::allocate_block_detail(gr_basic_block_sptr block)
   int noutputs = calc_used_ports(block, false).size();
   gr_block_detail_sptr detail = gr_make_block_detail(ninputs, noutputs);
 
+  gr_block_sptr grblock = cast_to_block_sptr(block);
+  if(!grblock)
+    throw std::runtime_error("allocate_block_detail found non-gr_block");
+
   if (GR_FLAT_FLOWGRAPH_DEBUG)
     std::cout << "Creating block detail for " << block << std::endl;
 
   for (int i = 0; i < noutputs; i++) {
-    block->expand_minmax_buffer(i);
+    grblock->expand_minmax_buffer(i);
+
     gr_buffer_sptr buffer = allocate_buffer(block, i);
     if (GR_FLAT_FLOWGRAPH_DEBUG)
       std::cout << "Allocated buffer for output " << block << ":" << i << std::endl;
     detail->set_output(i, buffer);
 
     // Update the block's max_output_buffer based on what was actually allocated.
-    block->set_max_output_buffer(i, buffer->bufsize());
+    grblock->set_max_output_buffer(i, buffer->bufsize());
   }
 
   return detail;
@@ -119,15 +134,15 @@ gr_flat_flowgraph::allocate_buffer(gr_basic_block_sptr block, int port)
   gr_basic_block_vector_t blocks = calc_downstream_blocks(block, port);
 
   // limit buffer size if indicated
-  if(block->max_output_buffer(port) > 0) {
+  if(grblock->max_output_buffer(port) > 0) {
 //    std::cout << "constraining output items to " << block->max_output_buffer(port) << "\n";
-    nitems = std::min((long)nitems, (long)block->max_output_buffer(port));
+    nitems = std::min((long)nitems, (long)grblock->max_output_buffer(port));
     nitems -= nitems%grblock->output_multiple();
     if( nitems < 1 )
       throw std::runtime_error("problems allocating a buffer with the given max output buffer constraint!");
   }
-  else if(block->min_output_buffer(port) > 0) {
-    nitems = std::max((long)nitems, (long)block->min_output_buffer(port));
+  else if(grblock->min_output_buffer(port) > 0) {
+    nitems = std::max((long)nitems, (long)grblock->min_output_buffer(port));
     nitems -= nitems%grblock->output_multiple();
     if( nitems < 1 )
       throw std::runtime_error("problems allocating a buffer with the given min output buffer constraint!");
@@ -345,3 +360,44 @@ gr_flat_flowgraph::make_block_vector(gr_basic_block_vector_t &blocks)
 
   return result;
 }
+
+
+void gr_flat_flowgraph::clear_endpoint(const gr_msg_endpoint &e, bool is_src){
+    for(size_t i=0; i<d_msg_edges.size(); i++){
+        if(is_src){
+            if(d_msg_edges[i].src() == e){
+                d_msg_edges.erase(d_msg_edges.begin() + i);
+                i--;
+            }
+        } else {
+            if(d_msg_edges[i].dst() == e){
+                d_msg_edges.erase(d_msg_edges.begin() + i);
+                i--;
+            }
+        }
+    }
+}
+
+void gr_flat_flowgraph::replace_endpoint(const gr_msg_endpoint &e, const gr_msg_endpoint &r, bool is_src){
+    size_t n_replr(0);
+    if(GR_FLAT_FLOWGRAPH_DEBUG)
+        std::cout << boost::format("gr_flat_flowgraph::replace_endpoint( %s, %s, %d )\n") % e.block()% r.block()% is_src;
+    for(size_t i=0; i<d_msg_edges.size(); i++){
+        if(is_src){
+            if(d_msg_edges[i].src() == e){
+                if(GR_FLAT_FLOWGRAPH_DEBUG)
+                    std::cout << boost::format("gr_flat_flowgraph::replace_endpoint() flattening to ( %s, %s )\n") % r.block()% d_msg_edges[i].dst().block();
+                d_msg_edges.push_back( gr_msg_edge(r, d_msg_edges[i].dst() ) );
+                n_replr++;
+            }
+        } else {
+            if(d_msg_edges[i].dst() == e){
+                if(GR_FLAT_FLOWGRAPH_DEBUG)
+                    std::cout << boost::format("gr_flat_flowgraph::replace_endpoint() flattening to ( %s, %s )\n") % r.block()% d_msg_edges[i].dst().block();
+                d_msg_edges.push_back( gr_msg_edge(d_msg_edges[i].src(), r ) );
+                n_replr++;
+            }
+        }
+    }
+}
+
diff --git a/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.h b/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.h
index 0926bcc8f..5c8268d7d 100644
--- a/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.h
+++ b/gnuradio-core/src/lib/runtime/gr_flat_flowgraph.h
@@ -46,7 +46,7 @@ public:
 
   // Wire gr_blocks together in new flat_flowgraph
   void setup_connections();
-
+  
   // Merge applicable connections from existing flat flowgraph
   void merge_connections(gr_flat_flowgraph_sptr sfg);
 
@@ -57,6 +57,9 @@ public:
    */
   static gr_block_vector_t make_block_vector(gr_basic_block_vector_t &blocks);
 
+  void replace_endpoint(const gr_msg_endpoint &e, const gr_msg_endpoint &r, bool is_src);
+  void clear_endpoint(const gr_msg_endpoint &e, bool is_src);
+
 private:
   gr_flat_flowgraph();
 
diff --git a/gnuradio-core/src/lib/runtime/gr_flowgraph.cc b/gnuradio-core/src/lib/runtime/gr_flowgraph.cc
index 78e1bc99a..63a208480 100644
--- a/gnuradio-core/src/lib/runtime/gr_flowgraph.cc
+++ b/gnuradio-core/src/lib/runtime/gr_flowgraph.cc
@@ -149,6 +149,16 @@ gr_flowgraph::check_valid_port(gr_io_signature_sptr sig, int port)
   }
 }
 
+void 
+gr_flowgraph::check_valid_port(const gr_msg_endpoint &e)
+{
+    if (GR_FLOWGRAPH_DEBUG)
+        std::cout << "check_valid_port( " << e.block() << ", " << e.port() << ")\n";
+
+    if(!e.block()->has_msg_port(e.port()))
+        throw std::invalid_argument("invalid msg port in connect() or disconnect()");
+}
+
 void
 gr_flowgraph::check_dst_not_used(const gr_endpoint &dst)
 {
@@ -180,6 +190,13 @@ gr_flowgraph::calc_used_blocks()
 {
   gr_basic_block_vector_t tmp;
 
+  // make sure free standing message blocks are included
+  for (gr_msg_edge_viter_t p = d_msg_edges.begin(); p != d_msg_edges.end(); p++) {
+//  for now only blocks receiving messages get a thread context - uncomment to allow senders to also obtain one
+//    tmp.push_back(p->src().block());
+    tmp.push_back(p->dst().block());
+  }
+
   // Collect all blocks in the edge list
   for (gr_edge_viter_t p = d_edges.begin(); p != d_edges.end(); p++) {
     tmp.push_back(p->src().block());
@@ -472,3 +489,27 @@ gr_flowgraph::topological_dfs_visit(gr_basic_block_sptr block, gr_basic_block_ve
   output.push_back(block);
 }
 
+void gr_flowgraph::connect(const gr_msg_endpoint &src, const gr_msg_endpoint &dst){
+  check_valid_port(src);
+  check_valid_port(dst);
+  for (gr_msg_edge_viter_t p = d_msg_edges.begin(); p != d_msg_edges.end(); p++) {
+    if(p->src() == src && p->dst() == dst){ 
+        throw std::runtime_error("connect called on already connected edge!");
+        }
+    }
+  d_msg_edges.push_back(gr_msg_edge(src,dst));
+}
+
+void gr_flowgraph::disconnect(const gr_msg_endpoint &src, const gr_msg_endpoint &dst){
+  check_valid_port(src);
+  check_valid_port(dst);
+  for (gr_msg_edge_viter_t p = d_msg_edges.begin(); p != d_msg_edges.end(); p++) {
+    if(p->src() == src && p->dst() == dst){
+        d_msg_edges.erase(p);
+        return;
+        }
+  }
+  throw std::runtime_error("disconnect called on non-connected edge!");
+}
+
+
diff --git a/gnuradio-core/src/lib/runtime/gr_flowgraph.h b/gnuradio-core/src/lib/runtime/gr_flowgraph.h
index a2c1580eb..bef70f626 100644
--- a/gnuradio-core/src/lib/runtime/gr_flowgraph.h
+++ b/gnuradio-core/src/lib/runtime/gr_flowgraph.h
@@ -52,6 +52,31 @@ inline bool gr_endpoint::operator==(const gr_endpoint &other) const
 	  d_port == other.d_port);
 }
 
+class GR_CORE_API gr_msg_endpoint
+{
+private:
+  gr_basic_block_sptr d_basic_block;
+  pmt::pmt_t d_port;
+  bool d_is_hier;
+public:
+  gr_msg_endpoint() : d_basic_block(), d_port(pmt::PMT_NIL) { }
+  gr_msg_endpoint(gr_basic_block_sptr block, pmt::pmt_t port, bool is_hier=false){ d_basic_block = block; d_port = port; d_is_hier = is_hier;}
+  gr_basic_block_sptr block() const { return d_basic_block; }
+  pmt::pmt_t port() const { return d_port; }
+  bool is_hier() const { return d_is_hier; }
+  void set_hier(bool h) { d_is_hier = h; }
+
+  bool operator==(const gr_msg_endpoint &other) const;
+
+};
+
+inline bool gr_msg_endpoint::operator==(const gr_msg_endpoint &other) const
+{
+  return (d_basic_block == other.d_basic_block &&
+	  pmt::pmt_equal(d_port, other.d_port));
+}
+
+
 // Hold vectors of gr_endpoint objects
 typedef std::vector<gr_endpoint> gr_endpoint_vector_t;
 typedef std::vector<gr_endpoint>::iterator gr_endpoint_viter_t;
@@ -75,11 +100,35 @@ private:
   gr_endpoint d_dst;
 };
 
+
 // Hold vectors of gr_edge objects
 typedef std::vector<gr_edge> gr_edge_vector_t;
 typedef std::vector<gr_edge>::iterator gr_edge_viter_t;
 
 
+/*!
+ *\brief Class representing a msg connection between to graph msg endpoints
+ *
+ */
+class GR_CORE_API gr_msg_edge
+{
+public:
+  gr_msg_edge() : d_src(), d_dst() { };
+  gr_msg_edge(const gr_msg_endpoint &src, const gr_msg_endpoint &dst) : d_src(src), d_dst(dst) { }
+  ~gr_msg_edge() {}
+
+  const gr_msg_endpoint &src() const { return d_src; }
+  const gr_msg_endpoint &dst() const { return d_dst; }
+
+private:
+  gr_msg_endpoint d_src;
+  gr_msg_endpoint d_dst;
+};
+
+// Hold vectors of gr_edge objects
+typedef std::vector<gr_msg_edge> gr_msg_edge_vector_t;
+typedef std::vector<gr_msg_edge>::iterator gr_msg_edge_viter_t;
+
 // Create a shared pointer to a heap allocated flowgraph
 // (types defined in gr_runtime_types.h)
 GR_CORE_API gr_flowgraph_sptr gr_make_flowgraph();
@@ -110,6 +159,12 @@ public:
   void disconnect(gr_basic_block_sptr src_block, int src_port,
 		  gr_basic_block_sptr dst_block, int dst_port);
 
+  // Connect two msg endpoints
+  void connect(const gr_msg_endpoint &src, const gr_msg_endpoint &dst);
+  
+  // Disconnect two msg endpoints
+  void disconnect(const gr_msg_endpoint &src, const gr_msg_endpoint &dst);
+
   // Validate connectivity, raise exception if invalid
   void validate();
 
@@ -118,6 +173,9 @@ public:
 
   // Return vector of edges
   const gr_edge_vector_t &edges() const { return d_edges; }
+  
+  // Return vector of msg edges
+  const gr_msg_edge_vector_t &msg_edges() const { return d_msg_edges; }
 
   // Return vector of connected blocks
   gr_basic_block_vector_t calc_used_blocks();
@@ -132,6 +190,7 @@ public:
 protected:
   gr_basic_block_vector_t d_blocks;
   gr_edge_vector_t d_edges;
+  gr_msg_edge_vector_t d_msg_edges;
 
   gr_flowgraph();
   std::vector<int> calc_used_ports(gr_basic_block_sptr block, bool check_inputs);
@@ -143,6 +202,7 @@ protected:
 private:
 
   void check_valid_port(gr_io_signature_sptr sig, int port);
+  void check_valid_port(const gr_msg_endpoint &e);
   void check_dst_not_used(const gr_endpoint &dst);
   void check_type_match(const gr_endpoint &src, const gr_endpoint &dst);
   gr_edge_vector_t calc_connections(gr_basic_block_sptr block, bool check_inputs); // false=use outputs
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2.cc b/gnuradio-core/src/lib/runtime/gr_hier_block2.cc
index 756852df8..8c2794c63 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2.cc
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2.cc
@@ -44,7 +44,9 @@ gr_hier_block2::gr_hier_block2(const std::string &name,
                                gr_io_signature_sptr input_signature,
                                gr_io_signature_sptr output_signature)
   : gr_basic_block(name, input_signature, output_signature),
-    d_detail(new gr_hier_block2_detail(this))
+    d_detail(new gr_hier_block2_detail(this)),
+    hier_message_ports_in(pmt::PMT_NIL),
+    hier_message_ports_out(pmt::PMT_NIL)
 {
   // This bit of magic ensures that self() works in the constructors of derived classes.
   gnuradio::detail::sptr_magic::create_and_stash_initial_sptr(this);
@@ -81,6 +83,36 @@ gr_hier_block2::connect(gr_basic_block_sptr src, int src_port,
 }
 
 void
+gr_hier_block2::msg_connect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+                        gr_basic_block_sptr dst, pmt::pmt_t dstport)
+{
+  if(!pmt::pmt_is_symbol(srcport)){throw std::runtime_error("bad port id"); }
+  d_detail->msg_connect(src, srcport, dst, dstport);
+}
+
+void
+gr_hier_block2::msg_connect(gr_basic_block_sptr src, std::string srcport,
+                        gr_basic_block_sptr dst, std::string dstport)
+{
+  d_detail->msg_connect(src, pmt::mp(srcport), dst, pmt::mp(dstport));
+}
+
+void
+gr_hier_block2::msg_disconnect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+                        gr_basic_block_sptr dst, pmt::pmt_t dstport)
+{
+  if(!pmt::pmt_is_symbol(srcport)){throw std::runtime_error("bad port id"); }
+  d_detail->msg_disconnect(src, srcport, dst, dstport);
+}
+
+void
+gr_hier_block2::msg_disconnect(gr_basic_block_sptr src, std::string srcport,
+                        gr_basic_block_sptr dst, std::string dstport)
+{
+  d_detail->msg_disconnect(src, pmt::mp(srcport), dst, pmt::mp(dstport));
+}
+
+void
 gr_hier_block2::disconnect(gr_basic_block_sptr block)
 {
   d_detail->disconnect(block);
@@ -111,6 +143,7 @@ gr_hier_block2::unlock()
   d_detail->unlock();
 }
 
+
 gr_flat_flowgraph_sptr
 gr_hier_block2::flatten() const
 {
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2.h b/gnuradio-core/src/lib/runtime/gr_hier_block2.h
index 123178724..f80dd73e4 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2.h
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2.h
@@ -98,6 +98,21 @@ public:
 	       gr_basic_block_sptr dst, int dst_port);
 
   /*!
+   * \brief Add gr-blocks or hierarchical blocks to internal graph and wire together
+   *
+   * This adds (if not done earlier by another connect) a pair of gr-blocks or
+   * hierarchical blocks to the internal message port subscription
+   */
+  void msg_connect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+	       gr_basic_block_sptr dst, pmt::pmt_t dstport);
+  void msg_connect(gr_basic_block_sptr src, std::string srcport,
+	       gr_basic_block_sptr dst, std::string dstport);
+  void msg_disconnect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+	       gr_basic_block_sptr dst, pmt::pmt_t dstport);
+  void msg_disconnect(gr_basic_block_sptr src, std::string srcport,
+	       gr_basic_block_sptr dst, std::string dstport);
+
+  /*!
    * \brief Remove a gr-block or hierarchical block from the internal flowgraph.
    *
    * This removes a gr-block or hierarchical block from the internal flowgraph,
@@ -151,6 +166,39 @@ public:
   gr_flat_flowgraph_sptr flatten() const;
 
   gr_hier_block2_sptr to_hier_block2(); // Needed for Python type coercion
+
+  bool has_msg_port(pmt::pmt_t which_port){
+    return message_port_is_hier(which_port) || gr_basic_block::has_msg_port(which_port);
+    }   
+  
+  bool message_port_is_hier(pmt::pmt_t port_id){
+    return message_port_is_hier_in(port_id) || message_port_is_hier_out(port_id);
+    }
+  bool message_port_is_hier_in(pmt::pmt_t port_id){
+    return pmt::pmt_list_has(hier_message_ports_in, port_id);
+    }
+  bool message_port_is_hier_out(pmt::pmt_t port_id){
+    return pmt::pmt_list_has(hier_message_ports_out, port_id);
+    }
+
+  pmt::pmt_t hier_message_ports_in;
+  pmt::pmt_t hier_message_ports_out;
+
+  void message_port_register_hier_in(pmt::pmt_t port_id){
+    if(pmt::pmt_list_has(hier_message_ports_in, port_id))
+        throw std::invalid_argument("hier msg in port by this name already registered");
+    if(msg_queue.find(port_id) != msg_queue.end())
+        throw std::invalid_argument("block already has a primitive input port by this name");
+    hier_message_ports_in = pmt::pmt_list_add(hier_message_ports_in, port_id);
+    }
+  void message_port_register_hier_out(pmt::pmt_t port_id){
+    if(pmt::pmt_list_has(hier_message_ports_out, port_id))
+        throw std::invalid_argument("hier msg out port by this name already registered");
+    if(pmt::pmt_dict_has_key(message_subscribers, port_id))
+        throw std::invalid_argument("block already has a primitive output port by this name");
+    hier_message_ports_out = pmt::pmt_list_add(hier_message_ports_out, port_id);
+    }
+
 };
 
 inline gr_hier_block2_sptr cast_to_hier_block2_sptr(gr_basic_block_sptr block) {
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2.i b/gnuradio-core/src/lib/runtime/gr_hier_block2.i
index eefb965b4..a857394ca 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2.i
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2.i
@@ -38,6 +38,10 @@ gr_hier_block2_sptr gr_make_hier_block2(const std::string name,
 // better interface in scripting land.
 %rename(primitive_connect) gr_hier_block2::connect;
 %rename(primitive_disconnect) gr_hier_block2::disconnect;
+%rename(primitive_msg_connect) gr_hier_block2::msg_connect;
+%rename(primitive_msg_disconnect) gr_hier_block2::msg_disconnect;
+%rename(primitive_message_port_register_hier_in) gr_hier_block2::message_port_register_hier_in;
+%rename(primitive_message_port_register_hier_out) gr_hier_block2::message_port_register_hier_out;
 
 class gr_hier_block2 : public gr_basic_block
 {
@@ -54,6 +58,19 @@ public:
   void connect(gr_basic_block_sptr src, int src_port,
 	       gr_basic_block_sptr dst, int dst_port)
     throw (std::invalid_argument);
+  void msg_connect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+           gr_basic_block_sptr dst, pmt::pmt_t dstport)
+    throw (std::runtime_error);
+  void msg_connect(gr_basic_block_sptr src, std::string srcport,
+           gr_basic_block_sptr dst,  std::string dstport)
+    throw (std::runtime_error);
+  void msg_disconnect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+           gr_basic_block_sptr dst, pmt::pmt_t dstport)
+    throw (std::runtime_error);
+  void msg_disconnect(gr_basic_block_sptr src, std::string srcport,
+           gr_basic_block_sptr dst, std::string dstport)
+    throw (std::runtime_error);
+
   void disconnect(gr_basic_block_sptr block)
     throw (std::invalid_argument);
   void disconnect(gr_basic_block_sptr src, int src_port,
@@ -63,5 +80,9 @@ public:
   void lock();
   void unlock();
 
+  void message_port_register_hier_in(pmt::pmt_t port_id);
+  void message_port_register_hier_out(pmt::pmt_t port_id);
+
+
   gr_hier_block2_sptr to_hier_block2(); // Needed for Python type coercion
 };
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
index 76c5ce06f..add6da024 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.cc
@@ -27,6 +27,7 @@
 #include <gr_io_signature.h>
 #include <stdexcept>
 #include <sstream>
+#include <boost/format.hpp>
 
 #define GR_HIER_BLOCK2_DETAIL_DEBUG 0
 
@@ -53,6 +54,7 @@ gr_hier_block2_detail::gr_hier_block2_detail(gr_hier_block2 *owner) :
   d_outputs = gr_endpoint_vector_t(max_outputs);
 }
 
+
 gr_hier_block2_detail::~gr_hier_block2_detail()
 {
   d_owner = 0; // Don't use delete, we didn't allocate
@@ -144,6 +146,65 @@ gr_hier_block2_detail::connect(gr_basic_block_sptr src, int src_port,
 }
 
 void
+gr_hier_block2_detail::msg_connect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+                               gr_basic_block_sptr dst, pmt::pmt_t dstport)
+{
+  if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+    std::cout << "connecting message port..." << std::endl;
+    
+  // register the subscription
+// this is done later...
+//  src->message_port_sub(srcport, pmt::pmt_cons(dst->alias_pmt(), dstport));
+
+  // add block uniquely to list to internal blocks
+  if (std::find(d_blocks.begin(), d_blocks.end(), dst) == d_blocks.end()){
+    d_blocks.push_back(src);
+    d_blocks.push_back(dst);
+    }
+
+  bool hier_out = (d_owner == src.get()) && src->message_port_is_hier_out(srcport);;
+  bool hier_in = (d_owner == dst.get()) && dst->message_port_is_hier_in(dstport);
+
+  gr_hier_block2_sptr src_block(cast_to_hier_block2_sptr(src));
+  gr_hier_block2_sptr dst_block(cast_to_hier_block2_sptr(dst));
+
+  if (src_block && src.get() != d_owner) {
+    if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+      std::cout << "connect: src is hierarchical, setting parent to " << this << std::endl;
+    src_block->d_detail->d_parent_detail = this;
+  }
+
+  if (dst_block && dst.get() != d_owner) {
+    if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+      std::cout << "connect: dst is hierarchical, setting parent to " << this << std::endl;
+    dst_block->d_detail->d_parent_detail = this;
+  }
+
+  // add edge for this message connection
+  if(GR_HIER_BLOCK2_DETAIL_DEBUG)
+        std::cout << boost::format("connect( (%s, %s, %d), (%s, %s, %d) )\n") %
+                        src % srcport % hier_out %
+                        dst % dstport % hier_in;
+  d_fg->connect( gr_msg_endpoint(src, srcport, hier_out), gr_msg_endpoint(dst, dstport, hier_in));
+}
+
+void
+gr_hier_block2_detail::msg_disconnect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+                               gr_basic_block_sptr dst, pmt::pmt_t dstport)
+{
+  if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+    std::cout << "disconnecting message port..." << std::endl;
+    
+  // unregister the subscription - if already subscribed
+  src->message_port_unsub(srcport, pmt::pmt_cons(dst->alias_pmt(), dstport));
+
+  // remove edge for this message connection
+  bool hier_out = (d_owner == src.get()) && src->message_port_is_hier_out(srcport);;
+  bool hier_in = (d_owner == dst.get()) && dst->message_port_is_hier_in(dstport);
+  d_fg->disconnect( gr_msg_endpoint(src, srcport, hier_out), gr_msg_endpoint(dst, dstport, hier_in));
+}
+
+void
 gr_hier_block2_detail::disconnect(gr_basic_block_sptr block)
 {
   // Check on singleton list
@@ -405,11 +466,16 @@ void
 gr_hier_block2_detail::flatten_aux(gr_flat_flowgraph_sptr sfg) const
 {
   if (GR_HIER_BLOCK2_DETAIL_DEBUG)
-    std::cout << "Flattening " << d_owner->name() << std::endl;
+    std::cout << " ** Flattening " << d_owner->name() << std::endl;
 
   // Add my edges to the flow graph, resolving references to actual endpoints
   gr_edge_vector_t edges = d_fg->edges();
+  gr_msg_edge_vector_t msg_edges = d_fg->msg_edges();
   gr_edge_viter_t p;
+  gr_msg_edge_viter_t q,u;
+
+  if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+    std::cout << "Flattening stream connections: " << std::endl;
 
   for (p = edges.begin(); p != edges.end(); p++) {
     if (GR_HIER_BLOCK2_DETAIL_DEBUG)
@@ -428,6 +494,53 @@ gr_hier_block2_detail::flatten_aux(gr_flat_flowgraph_sptr sfg) const
     }
   }
 
+  // loop through flattening hierarchical connections
+  if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+    std::cout << "Flattening msg connections: " << std::endl;
+
+
+  std::vector<std::pair<gr_msg_endpoint, bool> > resolved_endpoints;
+  for(q = msg_edges.begin(); q != msg_edges.end(); q++) {
+    if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+      std::cout << boost::format(" flattening edge ( %s, %s, %d) -> ( %s, %s, %d)\n") % q->src().block() % q->src().port() % q->src().is_hier() % q->dst().block() % q->dst().port() % q->dst().is_hier();
+
+    bool normal_connection = true;
+
+    // resolve existing connections to hier ports
+    if(q->dst().is_hier()){
+        if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+          std::cout << boost::format("  resolve hier output (%s, %s)") % q->dst().block() % q->dst().port() << std::endl;
+        sfg->replace_endpoint( q->dst(), q->src(), true );
+        resolved_endpoints.push_back(std::pair<gr_msg_endpoint, bool>(q->dst(),true));
+        normal_connection = false;
+        }
+
+    if(q->src().is_hier()){
+        if (GR_HIER_BLOCK2_DETAIL_DEBUG)
+          std::cout << boost::format("  resolve hier input (%s, %s)") % q->src().block() % q->src().port() << std::endl;
+        sfg->replace_endpoint( q->src(), q->dst(), false );
+        resolved_endpoints.push_back(std::pair<gr_msg_endpoint, bool>(q->src(),false));
+        normal_connection = false;
+        } 
+
+    // propogate non hier connections through
+    if(normal_connection){
+        sfg->connect( q->src(), q->dst() );
+        } 
+    }
+    for(std::vector<std::pair<gr_msg_endpoint, bool> >::iterator it = resolved_endpoints.begin(); it != resolved_endpoints.end(); it++){
+        sfg->clear_endpoint( (*it).first, (*it).second );
+    }
+
+/*  // connect primitive edges in the new fg
+  for(q = msg_edges.begin(); q != msg_edges.end(); q++) {
+    if( (!q->src().is_hier()) && (!q->dst().is_hier()) ){
+        sfg->connect( q->src(), q->dst() );
+        } else {
+        std::cout << "not connecting hier connection!" << std::endl;
+        }
+    }*/
+
   // Construct unique list of blocks used either in edges, inputs,
   // outputs, or by themselves.  I still hate STL.
   gr_basic_block_vector_t blocks; // unique list of used blocks
@@ -468,7 +581,7 @@ gr_hier_block2_detail::flatten_aux(gr_flat_flowgraph_sptr sfg) const
   // Recurse hierarchical children
   for (gr_basic_block_viter_t p = blocks.begin(); p != blocks.end(); p++) {
     gr_hier_block2_sptr hier_block2(cast_to_hier_block2_sptr(*p));
-    if (hier_block2) {
+    if (hier_block2 && (hier_block2.get() != d_owner)) {
       if (GR_HIER_BLOCK2_DETAIL_DEBUG)
 	std::cout << "flatten_aux: recursing into hierarchical block " << hier_block2 << std::endl;
       hier_block2->d_detail->flatten_aux(sfg);
@@ -499,3 +612,4 @@ gr_hier_block2_detail::unlock()
   else
     d_owner->unlock();
 }
+
diff --git a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
index f4f950e9d..b38dae301 100644
--- a/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_hier_block2_detail.h
@@ -39,6 +39,10 @@ public:
   void connect(gr_basic_block_sptr block);
   void connect(gr_basic_block_sptr src, int src_port,
 	       gr_basic_block_sptr dst, int dst_port);
+  void msg_connect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+	       gr_basic_block_sptr dst, pmt::pmt_t dstport);
+  void msg_disconnect(gr_basic_block_sptr src, pmt::pmt_t srcport,
+	       gr_basic_block_sptr dst, pmt::pmt_t dstport);
   void disconnect(gr_basic_block_sptr block);
   void disconnect(gr_basic_block_sptr, int src_port,
 		  gr_basic_block_sptr, int dst_port);
@@ -57,6 +61,7 @@ private:
   gr_endpoint_vector_t d_outputs;             // Single internal endpoint per external output
   gr_basic_block_vector_t d_blocks;
 
+
   void connect_input(int my_port, int port, gr_basic_block_sptr block);
   void connect_output(int my_port, int port, gr_basic_block_sptr block);
   void disconnect_input(int my_port, int port, gr_basic_block_sptr block);
diff --git a/gnuradio-core/src/lib/runtime/gr_msg_accepter.cc b/gnuradio-core/src/lib/runtime/gr_msg_accepter.cc
index 5018ee9e6..93d5fb20e 100644
--- a/gnuradio-core/src/lib/runtime/gr_msg_accepter.cc
+++ b/gnuradio-core/src/lib/runtime/gr_msg_accepter.cc
@@ -41,12 +41,12 @@ gr_msg_accepter::~gr_msg_accepter()
 }
 
 void
-gr_msg_accepter::post(pmt_t msg)
+gr_msg_accepter::post(pmt_t which_port, pmt_t msg)
 {
   // Notify derived class, handled case by case
   gr_block *p = dynamic_cast<gr_block *>(this);
   if (p) {
-    p->detail()->_post(msg);
+    p->_post(which_port,msg);
     return;
   }
   gr_hier_block2 *p2 = dynamic_cast<gr_hier_block2 *>(this);
diff --git a/gnuradio-core/src/lib/runtime/gr_msg_accepter.h b/gnuradio-core/src/lib/runtime/gr_msg_accepter.h
index 3e5c97638..a497ba6e7 100644
--- a/gnuradio-core/src/lib/runtime/gr_msg_accepter.h
+++ b/gnuradio-core/src/lib/runtime/gr_msg_accepter.h
@@ -36,7 +36,7 @@ public:
   gr_msg_accepter();
   ~gr_msg_accepter();
 
-  void post(pmt::pmt_t msg);
+  void post(pmt::pmt_t which_port, pmt::pmt_t msg);
 
 };
 
diff --git a/gnuradio-core/src/lib/runtime/gr_tpb_detail.cc b/gnuradio-core/src/lib/runtime/gr_tpb_detail.cc
index 46b33d91f..46eb6bbe0 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_detail.cc
@@ -68,43 +68,3 @@ gr_tpb_detail::notify_neighbors(gr_block_detail *d)
   notify_upstream(d);
 }
 
-void
-gr_tpb_detail::insert_tail(pmt::pmt_t msg)
-{
-  gruel::scoped_lock guard(mutex);
-
-  msg_queue.push_back(msg);
-
-  // wake up thread if BLKD_IN or BLKD_OUT
-  input_cond.notify_one();
-  output_cond.notify_one();
-}
-
-pmt_t
-gr_tpb_detail::delete_head_nowait()
-{
-  gruel::scoped_lock guard(mutex);
-
-  if (empty_p())
-    return pmt_t();
-
-  pmt_t m(msg_queue.front());
-  msg_queue.pop_front();
-
-  return m;
-}
-
-/*
- * Caller must already be holding the mutex
- */
-pmt_t
-gr_tpb_detail::delete_head_nowait_already_holding_mutex()
-{
-  if (empty_p())
-    return pmt_t();
-
-  pmt_t m(msg_queue.front());
-  msg_queue.pop_front();
-
-  return m;
-}
diff --git a/gnuradio-core/src/lib/runtime/gr_tpb_detail.h b/gnuradio-core/src/lib/runtime/gr_tpb_detail.h
index b6e342dee..69feb6007 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_detail.h
@@ -39,9 +39,6 @@ struct GR_CORE_API gr_tpb_detail {
   bool				output_changed;
   gruel::condition_variable	output_cond;
 
-private:
-  std::deque<pmt::pmt_t>	msg_queue;
-
 public:
   gr_tpb_detail()
     : input_changed(false), output_changed(false) { }
@@ -55,6 +52,12 @@ public:
   //! Called by us to notify both upstream and downstream
   void notify_neighbors(gr_block_detail *d);
 
+  //! Called by pmt msg posters
+  void notify_msg(){
+    input_cond.notify_one();
+    output_cond.notify_one();
+    }
+
   //! Called by us
   void clear_changed()
   {
@@ -63,23 +66,6 @@ public:
     output_changed = false;
   }
 
-  //! is the queue empty?
-  bool empty_p() const { return msg_queue.empty(); }
-
-  //| Acquires and release the mutex
-  void insert_tail(pmt::pmt_t msg);
-
-  /*!
-   * \returns returns pmt at head of queue or pmt_t() if empty.
-   */
-  pmt::pmt_t delete_head_nowait();
-
-  /*!
-   * \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();
-
 private:
 
   //! Used by notify_downstream
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 a5aabb379..9f17a48a8 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
@@ -25,13 +25,14 @@
 #include <iostream>
 #include <boost/thread.hpp>
 #include <gruel/pmt.h>
+#include <boost/foreach.hpp>
 
 using namespace pmt;
 
 gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_items)
   : d_exec(block, max_noutput_items)
 {
-  // std::cerr << "gr_tpb_thread_body: " << block << std::endl;
+  //std::cerr << "gr_tpb_thread_body: " << block << std::endl;
 
   gr_block_detail *d = block->detail().get();
   gr_block_executor::state s;
@@ -42,11 +43,22 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
     boost::this_thread::interruption_point();
 
     // handle any queued up messages
-    while ((msg = d->d_tpb.delete_head_nowait()))
-      block->dispatch_msg(msg);
+    //BOOST_FOREACH( pmt::pmt_t port, block->msg_queue.keys() )
+    
+    BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
+    {
+        while ((msg = block->delete_head_nowait(i.first))){
+            block->dispatch_msg(i.first,msg);
+        }
+    }
 
     d->d_tpb.clear_changed();
-    s = d_exec.run_one_iteration();
+    // run one iteration if we are a connected stream block
+    if(d->noutputs() >0 || d->ninputs()>0){
+        s = d_exec.run_one_iteration();
+    } else {
+        s = gr_block_executor::BLKD_IN;
+    }
 
     switch(s){
     case gr_block_executor::READY:		// Tell neighbors we made progress.
@@ -67,15 +79,18 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
 	while (!d->d_tpb.input_changed){
 
 	  // wait for input or message
-	  while(!d->d_tpb.input_changed && d->d_tpb.empty_p())
+	  while(!d->d_tpb.input_changed && block->empty_p())
 	    d->d_tpb.input_cond.wait(guard);
 
 	  // handle all pending messages
-	  while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
-	    guard.unlock();			// release lock while processing msg
-	    block->dispatch_msg(msg);
-	    guard.lock();
-	  }
+      BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
+      {
+	    while ((msg = block->delete_head_nowait(i.first))){
+	      guard.unlock();			// release lock while processing msg
+	      block->dispatch_msg(i.first, msg);
+	      guard.lock();
+	    }
+      }
 	}
       }
       break;
@@ -87,15 +102,18 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
 	while (!d->d_tpb.output_changed){
 
 	  // wait for output room or message
-	  while(!d->d_tpb.output_changed && d->d_tpb.empty_p())
+	  while(!d->d_tpb.output_changed && block->empty_p())
 	    d->d_tpb.output_cond.wait(guard);
 
 	  // handle all pending messages
-	  while ((msg = d->d_tpb.delete_head_nowait_already_holding_mutex())){
-	    guard.unlock();			// release lock while processing msg
-	    block->dispatch_msg(msg);
-	    guard.lock();
-	  }
+      BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
+      {
+	    while ((msg = block->delete_head_nowait(i.first))){
+	      guard.unlock();			// release lock while processing msg
+	      block->dispatch_msg(i.first,msg);
+	      guard.lock();
+	    }
+      }
 	}
       }
       break;
diff --git a/gnuradio-core/src/lib/runtime/qa_set_msg_handler.cc b/gnuradio-core/src/lib/runtime/qa_set_msg_handler.cc
index 25ae0b1e1..c84a219bd 100644
--- a/gnuradio-core/src/lib/runtime/qa_set_msg_handler.cc
+++ b/gnuradio-core/src/lib/runtime/qa_set_msg_handler.cc
@@ -65,15 +65,11 @@ void qa_set_msg_handler::t0()
   tb->start();
 
   // Send them...
+  pmt_t port(pmt_intern("port"));
   for (int i = 0; i < NMSGS; i++){
-    send(nop, mp(mp("example-msg"), mp(i)));
+    send(nop, port, mp(mp("example-msg"), mp(i)));
   }
 
-  // And send a message to null_source to confirm that the default
-  // message handling action (which should be a nop) doesn't dump
-  // core.
-  send(src, mp(mp("example-msg"), mp(0)));
-
   // Give the messages a chance to be processed
   boost::this_thread::sleep(boost::posix_time::milliseconds(100));
 
diff --git a/gnuradio-core/src/lib/swig/CMakeLists.txt b/gnuradio-core/src/lib/swig/CMakeLists.txt
index 5b740d916..734547131 100644
--- a/gnuradio-core/src/lib/swig/CMakeLists.txt
+++ b/gnuradio-core/src/lib/swig/CMakeLists.txt
@@ -25,11 +25,9 @@ include_directories(${Boost_INCLUDE_DIRS})
 link_directories(${Boost_LIBRARY_DIRS})
 
 set(GR_SWIG_INCLUDE_DIRS
-    ${Boost_INCLUDE_DIRS}
-    ${GSL_INCLUDE_DIRS}
+    ${CMAKE_CURRENT_BINARY_DIR}
     ${GRUEL_INCLUDE_DIRS}
     ${GNURADIO_CORE_SWIG_INCLUDE_DIRS}
-    ${CMAKE_CURRENT_BINARY_DIR}
 )
 set(GR_SWIG_LIBRARIES gnuradio-core)
 
diff --git a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
index 6a0555021..62f3d7e46 100644
--- a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
+++ b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
@@ -23,6 +23,7 @@ include(GrPython)
 GR_PYTHON_INSTALL(FILES
     __init__.py
     exceptions.py
+    gateway.py
     gr_threading.py
     gr_threading_23.py
     gr_threading_24.py
@@ -42,6 +43,13 @@ include(GrTest)
 file(GLOB py_qa_test_files "qa_*.py")
 foreach(py_qa_test_file ${py_qa_test_files})
     get_filename_component(py_qa_test_name ${py_qa_test_file} NAME_WE)
+    set(GR_TEST_PYTHON_DIRS
+        ${CMAKE_SOURCE_DIR}/gruel/src/python
+        ${CMAKE_BINARY_DIR}/gruel/src/swig
+        ${CMAKE_BINARY_DIR}/gnuradio-core/src/python
+        ${CMAKE_BINARY_DIR}/gnuradio-core/src/lib/swig
+    )
+    set(GR_TEST_TARGET_DEPS volk gruel gnuradio-core)
     GR_ADD_TEST(${py_qa_test_name} ${PYTHON_EXECUTABLE} ${PYTHON_DASH_B} ${py_qa_test_file})
 endforeach(py_qa_test_file)
 endif(ENABLE_TESTING)
diff --git a/gnuradio-core/src/python/gnuradio/gr/__init__.py b/gnuradio-core/src/python/gnuradio/gr/__init__.py
index e5a8fdbf9..5d01ea11b 100644
--- a/gnuradio-core/src/python/gnuradio/gr/__init__.py
+++ b/gnuradio-core/src/python/gnuradio/gr/__init__.py
@@ -1,5 +1,5 @@
 #
-# Copyright 2003,2004,2006,2008,2009,2010 Free Software Foundation, Inc.
+# Copyright 2003-2012 Free Software Foundation, Inc.
 #
 # This file is part of GNU Radio
 #
@@ -28,6 +28,7 @@ from gnuradio_core import *
 from exceptions import *
 from hier_block2 import *
 from top_block import *
+from gateway import basic_block, sync_block, decim_block, interp_block
 
 # create a couple of aliases
 serial_to_parallel = stream_to_vector
diff --git a/gnuradio-core/src/python/gnuradio/gr/gateway.py b/gnuradio-core/src/python/gnuradio/gr/gateway.py
new file mode 100644
index 000000000..244b8b592
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/gateway.py
@@ -0,0 +1,215 @@
+#
+# Copyright 2011-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.
+#
+
+import gnuradio_core as gr_core
+from gnuradio_core import io_signature, io_signaturev
+from gnuradio_core import gr_block_gw_message_type
+from gnuradio_core import block_gateway
+import numpy
+
+########################################################################
+# Magic to turn pointers into numpy arrays
+# http://docs.scipy.org/doc/numpy/reference/arrays.interface.html
+########################################################################
+def pointer_to_ndarray(addr, dtype, nitems):
+    class array_like:
+        __array_interface__ = {
+            'data' : (int(addr), False),
+            'typestr' : dtype.base.str,
+            'descr' : dtype.base.descr,
+            'shape' : (nitems,) + dtype.shape,
+            'strides' : None,
+            'version' : 3
+        }
+    return numpy.asarray(array_like()).view(dtype.base)
+
+########################################################################
+# Handler that does callbacks from C++
+########################################################################
+class gateway_handler(gr_core.feval_ll):
+
+    #dont put a constructor, it wont work
+
+    def init(self, callback):
+        self._callback = callback
+
+    def eval(self, arg):
+        try: self._callback()
+        except Exception as ex:
+            print("handler caught exception: %s"%ex)
+            import traceback; traceback.print_exc()
+            raise ex
+        return 0
+
+########################################################################
+# The guts that make this into a gr block
+########################################################################
+class gateway_block(object):
+
+    def __init__(self, name, in_sig, out_sig, work_type, factor):
+
+        #ensure that the sigs are iterable dtypes
+        def sig_to_dtype_sig(sig):
+            if sig is None: sig = ()
+            return map(numpy.dtype, sig)
+        self.__in_sig = sig_to_dtype_sig(in_sig)
+        self.__out_sig = sig_to_dtype_sig(out_sig)
+
+        #cache the ranges to iterate when dispatching work
+        self.__in_indexes = range(len(self.__in_sig))
+        self.__out_indexes = range(len(self.__out_sig))
+
+        #convert the signatures into gr.io_signatures
+        def sig_to_gr_io_sigv(sig):
+            if not len(sig): return io_signature(0, 0, 0)
+            return io_signaturev(len(sig), len(sig), [s.itemsize for s in sig])
+        gr_in_sig = sig_to_gr_io_sigv(self.__in_sig)
+        gr_out_sig = sig_to_gr_io_sigv(self.__out_sig)
+
+        #create internal gateway block
+        self.__handler = gateway_handler()
+        self.__handler.init(self.__gr_block_handle)
+        self.__gateway = block_gateway(
+            self.__handler, name, gr_in_sig, gr_out_sig, work_type, factor)
+        self.__message = self.__gateway.gr_block_message()
+
+        #register gr_block functions
+        prefix = 'gr_block__'
+        for attr in [x for x in dir(self.__gateway) if x.startswith(prefix)]:
+            setattr(self, attr.replace(prefix, ''), getattr(self.__gateway, attr))
+        self.pop_msg_queue = lambda: gr_core.gr_block_gw_pop_msg_queue_safe(self.__gateway)
+
+    def to_basic_block(self):
+        """
+        Makes this block connectable by hier/top block python
+        """
+        return self.__gateway.to_basic_block()
+
+    def __gr_block_handle(self):
+        """
+        Dispatch tasks according to the action type specified in the message.
+        """
+        if self.__message.action == gr_block_gw_message_type.ACTION_GENERAL_WORK:
+            self.__message.general_work_args_return_value = self.general_work(
+
+                input_items=[pointer_to_ndarray(
+                    self.__message.general_work_args_input_items[i],
+                    self.__in_sig[i],
+                    self.__message.general_work_args_ninput_items[i]
+                ) for i in self.__in_indexes],
+
+                output_items=[pointer_to_ndarray(
+                    self.__message.general_work_args_output_items[i],
+                    self.__out_sig[i],
+                    self.__message.general_work_args_noutput_items
+                ) for i in self.__out_indexes],
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_WORK:
+            self.__message.work_args_return_value = self.work(
+
+                input_items=[pointer_to_ndarray(
+                    self.__message.work_args_input_items[i],
+                    self.__in_sig[i],
+                    self.__message.work_args_ninput_items
+                ) for i in self.__in_indexes],
+
+                output_items=[pointer_to_ndarray(
+                    self.__message.work_args_output_items[i],
+                    self.__out_sig[i],
+                    self.__message.work_args_noutput_items
+                ) for i in self.__out_indexes],
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_FORECAST:
+            self.forecast(
+                noutput_items=self.__message.forecast_args_noutput_items,
+                ninput_items_required=self.__message.forecast_args_ninput_items_required,
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_START:
+            self.__message.start_args_return_value = self.start()
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_STOP:
+            self.__message.stop_args_return_value = self.stop()
+
+    def forecast(self, noutput_items, ninput_items_required):
+        """
+        forecast is only called from a general block
+        this is the default implementation
+        """
+        for ninput_item in ninput_items_required:
+            ninput_item = noutput_items + self.history() - 1;
+        return
+
+    def general_work(self, *args, **kwargs):
+        """general work to be overloaded in a derived class"""
+        raise NotImplementedError("general work not implemented")
+
+    def work(self, *args, **kwargs):
+        """work to be overloaded in a derived class"""
+        raise NotImplementedError("work not implemented")
+
+    def start(self): return True
+    def stop(self): return True
+
+########################################################################
+# Wrappers for the user to inherit from
+########################################################################
+class basic_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_GENERAL,
+            factor=1, #not relevant factor
+        )
+
+class sync_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_SYNC,
+            factor=1,
+        )
+
+class decim_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig, decim):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_DECIM,
+            factor=decim,
+        )
+
+class interp_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig, interp):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_INTERP,
+            factor=interp,
+        )
diff --git a/gnuradio-core/src/python/gnuradio/gr/hier_block2.py b/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
index 0c45f1691..b95782238 100644
--- a/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
+++ b/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
@@ -20,6 +20,10 @@
 #
 
 from gnuradio_core import hier_block2_swig
+try:
+    import pmt
+except ImportError:
+    from gruel import pmt
 
 #
 # This hack forces a 'has-a' relationship to look like an 'is-a' one.
@@ -111,3 +115,15 @@ class hier_block2(object):
         self._hb.primitive_disconnect(src_block.to_basic_block(), src_port,
                                       dst_block.to_basic_block(), dst_port)
 
+    def msg_connect(self, src, srcport, dst, dstport):
+        self.primitive_msg_connect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
+    def msg_disconnect(self, src, srcport, dst, dstport):
+        self.primitive_msg_disconnect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
+    def message_port_register_hier_in(self, portname):
+        self.primitive_message_port_register_hier_in(pmt.pmt_intern(portname));
+
+    def message_port_register_hier_out(self, portname):
+        self.primitive_message_port_register_hier_out(pmt.pmt_intern(portname));
+
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py b/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py
new file mode 100644
index 000000000..911879f6f
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py
@@ -0,0 +1,235 @@
+#
+# Copyright 2011-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
+import pmt
+import numpy
+
+class add_2_f32_1_f32(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "add 2 f32",
+            in_sig = [numpy.float32, numpy.float32],
+            out_sig = [numpy.float32],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0] + input_items[1]
+        return len(output_items[0])
+
+class add_2_fc32_1_fc32(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "add 2 fc32",
+            in_sig = [numpy.complex64, numpy.complex64],
+            out_sig = [numpy.complex64],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0] + input_items[1]
+        return len(output_items[0])
+
+class convolve(gr.sync_block):
+    """
+    A demonstration using block history to properly perform a convolution.
+    """
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "convolve",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32]
+        )
+        self._taps = [1, 0, 0, 0]
+        self.set_history(len(self._taps))
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = numpy.convolve(input_items[0], self._taps, mode='valid')
+        return len(output_items[0])
+
+class decim2x(gr.decim_block):
+    def __init__(self):
+        gr.decim_block.__init__(
+            self,
+            name = "decim2x",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32],
+            decim = 2
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0][::2]
+        return len(output_items[0])
+
+class interp2x(gr.interp_block):
+    def __init__(self):
+        gr.interp_block.__init__(
+            self,
+            name = "interp2x",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32],
+            interp = 2
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][1::2] = input_items[0]
+        output_items[0][::2] = input_items[0]
+        return len(output_items[0])
+
+class tag_source(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "tag source",
+            in_sig = None,
+            out_sig = [numpy.float32],
+        )
+
+    def work(self, input_items, output_items):
+        num_output_items = len(output_items[0])
+
+        #put code here to fill the output items...
+
+        #make a new tag on the middle element every time work is called
+        count = self.nitems_written(0) + num_output_items/2
+        key = pmt.pmt_string_to_symbol("example_key")
+        value = pmt.pmt_string_to_symbol("example_value")
+        self.add_item_tag(0, count, key, value)
+
+        return num_output_items
+
+class tag_sink(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "tag sink",
+            in_sig = [numpy.float32],
+            out_sig = None,
+        )
+        self.key = None
+
+    def work(self, input_items, output_items):
+        num_input_items = len(input_items[0])
+
+        #put code here to process the input items...
+
+        #print all the tags received in this work call
+        nread = self.nitems_read(0)
+        tags = self.get_tags_in_range(0, nread, nread+num_input_items)
+        for tag in tags:
+            #print tag.offset
+            #print pmt.pmt_symbol_to_string(tag.key)
+            #print pmt.pmt_symbol_to_string(tag.value)
+            self.key = pmt.pmt_symbol_to_string(tag.key)
+
+        return num_input_items
+
+class fc32_to_f32_2(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "fc32_to_f32_2",
+            in_sig = [numpy.complex64],
+            out_sig = [(numpy.float32, 2)],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][::,0] = numpy.real(input_items[0])
+        output_items[0][::,1] = numpy.imag(input_items[0])
+        return len(output_items[0])
+
+class test_block_gateway(gr_unittest.TestCase):
+
+    def test_add_f32(self):
+        tb = gr.top_block()
+        src0 = gr.vector_source_f([1, 3, 5, 7, 9], False)
+        src1 = gr.vector_source_f([0, 2, 4, 6, 8], False)
+        adder = add_2_f32_1_f32()
+        sink = gr.vector_sink_f()
+        tb.connect((src0, 0), (adder, 0))
+        tb.connect((src1, 0), (adder, 1))
+        tb.connect(adder, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 5, 9, 13, 17))
+
+    def test_add_fc32(self):
+        tb = gr.top_block()
+        src0 = gr.vector_source_c([1, 3j, 5, 7j, 9], False)
+        src1 = gr.vector_source_c([0, 2j, 4, 6j, 8], False)
+        adder = add_2_fc32_1_fc32()
+        sink = gr.vector_sink_c()
+        tb.connect((src0, 0), (adder, 0))
+        tb.connect((src1, 0), (adder, 1))
+        tb.connect(adder, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 5j, 9, 13j, 17))
+
+    def test_convolve(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False)
+        cv = convolve()
+        sink = gr.vector_sink_f()
+        tb.connect(src, cv, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8))
+
+    def test_decim2x(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False)
+        d2x = decim2x()
+        sink = gr.vector_sink_f()
+        tb.connect(src, d2x, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 3, 5, 7))
+
+    def test_interp2x(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 3, 5, 7, 9], False)
+        i2x = interp2x()
+        sink = gr.vector_sink_f()
+        tb.connect(src, i2x, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 1, 3, 3, 5, 5, 7, 7, 9, 9))
+
+    def test_tags(self):
+        src = tag_source()
+        sink = tag_sink()
+        head = gr.head(gr.sizeof_float, 50000) #should be enough items to get a tag through
+        tb = gr.top_block()
+        tb.connect(src, head, sink)
+        tb.run()
+        self.assertEqual(sink.key, "example_key")
+
+    def test_fc32_to_f32_2(self):
+        tb = gr.top_block()
+        src = gr.vector_source_c([1+2j, 3+4j, 5+6j, 7+8j, 9+10j], False)
+        convert = fc32_to_f32_2()
+        v2s = gr.vector_to_stream(gr.sizeof_float, 2)
+        sink = gr.vector_sink_f()
+        tb.connect(src, convert, v2s, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+
+if __name__ == '__main__':
+    gr_unittest.run(test_block_gateway, "test_block_gateway.xml")
+
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..ebc365b61
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
@@ -0,0 +1,110 @@
+#!/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
+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()
+
+        # Test that the right number of ports exist.
+        pi = dbg.message_ports_in()
+        po = dbg.message_ports_out()
+        self.assertEqual(pmt.pmt_length(pi), 2)
+        self.assertEqual(pmt.pmt_length(po), 0)
+
+        pi = snk3.message_ports_in()
+        po = snk3.message_ports_out()
+        self.assertEqual(pmt.pmt_length(pi), 0)
+        self.assertEqual(pmt.pmt_length(po), 1)
+
+        #print "Message Debug input ports: "
+        #pmt.pmt_print(pi)
+        #print "Message Debug output ports: "
+        #pmt.pmt_print(po)
+        
+        #print "Stream to PDU input ports: "
+        #pmt.pmt_print(pi)
+        #print "Stream to PDU output ports: "
+        #pmt.pmt_print(po)
+        time.sleep(0.1)
+
+        self.tb.connect(src, snk)
+        self.tb.connect(src, snk2)
+        self.tb.connect(src, snk3)
+
+        self.tb.msg_connect(snk3, "pdus", dbg, "store")
+        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 )
+
+        while(dbg.num_messages() < 1):
+            time.sleep(0.5)
+        self.tb.stop()
+        self.tb.wait()
+
+        # Get the vector of data from the vector sink
+        result_data = snk2.data()
+
+        # Get the vector of data from the message sink
+        # Convert the message PMT as a pair into its vector
+        result_msg = dbg.get_message(0)
+        msg_vec = pmt.pmt_cdr(result_msg)
+        pmt.pmt_print(msg_vec)
+
+        # Convert the PMT vector into a Python list
+        msg_data = []
+        for i in xrange(16):
+            msg_data.append(pmt.pmt_u8vector_ref(msg_vec, i))
+
+        actual_data = 16*[0xFF,]
+        self.assertEqual(actual_data, list(result_data))
+        self.assertEqual(actual_data, msg_data)
+
+if __name__ == '__main__':
+    gr_unittest.run(test_pdu, "test_pdu.xml")
diff --git a/gnuradio-core/src/python/gnuradio/gr/top_block.py b/gnuradio-core/src/python/gnuradio/gr/top_block.py
index 43af8073b..dc1f443aa 100644
--- a/gnuradio-core/src/python/gnuradio/gr/top_block.py
+++ b/gnuradio-core/src/python/gnuradio/gr/top_block.py
@@ -123,6 +123,12 @@ class top_block(object):
 		for i in range (1, len (points)):
         	    self._connect(points[i-1], points[i])
 
+    def msg_connect(self, src, srcport, dst, dstport):
+        self.primitive_msg_connect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+    
+    def msg_disconnect(self, src, srcport, dst, dstport):
+        self.primitive_msg_disconnect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
     def _connect(self, src, dst):
         (src_block, src_port) = self._coerce_endpoint(src)
         (dst_block, dst_port) = self._coerce_endpoint(dst)