core: added gr_tuntap_pdu, gr_socket_pdu, and msg passing enhancements

14 files changed, 891 insertions, 35 deletions

diff --git a/gnuradio-core/src/lib/io/CMakeLists.txt b/gnuradio-core/src/lib/io/CMakeLists.txt
index 7041f2820..59ca06b5a 100644
--- a/gnuradio-core/src/lib/io/CMakeLists.txt
+++ b/gnuradio-core/src/lib/io/CMakeLists.txt
@@ -39,6 +39,7 @@ list(APPEND gnuradio_core_sources
     ${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
 )
 
 ########################################################################
@@ -61,6 +62,7 @@ install(FILES
     ${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"
 )
@@ -103,6 +105,8 @@ set(gr_core_io_triple_threats
     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_message_debug.cc b/gnuradio-core/src/lib/io/gr_message_debug.cc
index 7d28ff18e..27f4c65fd 100644
--- a/gnuradio-core/src/lib/io/gr_message_debug.cc
+++ b/gnuradio-core/src/lib/io/gr_message_debug.cc
@@ -58,6 +58,30 @@ gr_message_debug::store(pmt::pmt_t msg)
   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()
 {
@@ -81,11 +105,11 @@ gr_message_debug::gr_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));
+  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
index 1ffef1b02..6e6e5103c 100644
--- a/gnuradio-core/src/lib/io/gr_message_debug.h
+++ b/gnuradio-core/src/lib/io/gr_message_debug.h
@@ -55,6 +55,7 @@ class GR_CORE_API gr_message_debug : public gr_block
    * \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.
diff --git a/gnuradio-core/src/lib/io/gr_pdu.cc b/gnuradio-core/src/lib/io/gr_pdu.cc
index f33eed0a3..b2757c307 100644
--- a/gnuradio-core/src/lib/io/gr_pdu.cc
+++ b/gnuradio-core/src/lib/io/gr_pdu.cc
@@ -28,42 +28,52 @@
 
 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!");
-    }
+  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!");
-    }
+  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!");
-    }
+  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
index 67519c89d..5ed9cdded 100644
--- a/gnuradio-core/src/lib/io/gr_pdu.h
+++ b/gnuradio-core/src/lib/io/gr_pdu.h
@@ -34,5 +34,6 @@ 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_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..3a96a3f97
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_socket_pdu.h
@@ -0,0 +1,203 @@
+/* -*- 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>
+
+#include <linux/if_tun.h>
+
+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..dc5dc5c2e
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_stream_pdu_base.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_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>
+
+#include <linux/if_tun.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_tuntap_pdu.cc b/gnuradio-core/src/lib/io/gr_tuntap_pdu.cc
new file mode 100644
index 000000000..44de1a5f7
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tuntap_pdu.cc
@@ -0,0 +1,143 @@
+/* -*- 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
+
+boost::shared_ptr<gr_block> gr_make_tuntap_pdu (std::string dev, int MTU){
+    boost::shared_ptr<gr_block> 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..0e8071c30
--- /dev/null
+++ b/gnuradio-core/src/lib/io/gr_tuntap_pdu.h
@@ -0,0 +1,66 @@
+/* -*- 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
+
+GR_CORE_API boost::shared_ptr<gr_block> 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 871ce1356..e2de4eb97 100644
--- a/gnuradio-core/src/lib/io/io.i
+++ b/gnuradio-core/src/lib/io/io.i
@@ -49,6 +49,8 @@
 #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"
@@ -75,4 +77,7 @@
 %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"
+