Merge branch 'master' into next

Conflicts: gr-blocks/lib/stream_pdu_base.cc
author: Johnathan Corgan <johnathan@corganlabs.com> 2013-02-28 13:18:28 -0800
committer: Johnathan Corgan <johnathan@corganlabs.com> 2013-02-28 13:38:37 -0800
commit: 4a99e708ad0dd54bae4a4275ae2c06e44decbebf (patch)
tree: 5ec7ac397cdfce5532e101edc84f9c11d26a5569 /gr-blocks/lib
parent: 2f55d7dfc33e8d990e44c5bbb7c6d2fbdaddd563 (diff)
parent: 8a49689fbd5d55d74033ad398cab6862fcb17acc (diff)
12 files changed, 877 insertions, 4 deletions
diff --git a/gr-blocks/lib/CMakeLists.txt b/gr-blocks/lib/CMakeLists.txt
index 28158ab593..4ebaf3dcb8 100644
--- a/gr-blocks/lib/CMakeLists.txt
+++ b/gr-blocks/lib/CMakeLists.txt
@@ -166,6 +166,8 @@ list(APPEND gr_blocks_sources
     nlog10_ff_impl.cc
     pack_k_bits_bb_impl.cc
     patterned_interleaver_impl.cc
+    pdu.cc
+    pdu_to_tagged_stream_impl.cc
     peak_detector2_fb_impl.cc
     regenerate_bb_impl.cc
     repeat_impl.cc
@@ -173,6 +175,7 @@ list(APPEND gr_blocks_sources
     rms_ff_impl.cc
     short_to_char_impl.cc
     short_to_float_impl.cc
+    socket_pdu_impl.cc
     stream_mux_impl.cc
     stream_pdu_base.cc
     stream_to_streams_impl.cc
@@ -180,9 +183,11 @@ list(APPEND gr_blocks_sources
     streams_to_stream_impl.cc
     streams_to_vector_impl.cc
     stretch_ff_impl.cc
+    tagged_stream_to_pdu_impl.cc
     threshold_ff_impl.cc
     throttle_impl.cc
     transcendental_impl.cc
+    tcp_connection.cc
     tuntap_pdu_impl.cc
     uchar_array_to_float.cc
     uchar_to_float_impl.cc
diff --git a/gr-blocks/lib/pdu.cc b/gr-blocks/lib/pdu.cc
new file mode 100644
index 0000000000..ae4b17aecc
--- /dev/null
+++ b/gr-blocks/lib/pdu.cc
@@ -0,0 +1,92 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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 <blocks/pdu.h>
+
+namespace gr {
+  namespace blocks {
+    namespace pdu {
+
+      size_t 
+      itemsize(vector_type type)
+      {
+	switch(type) {
+	case byte_t:
+	  return sizeof(char);
+	case float_t:
+	  return sizeof(float);
+	case complex_t:
+	  return sizeof(gr_complex);
+	default:
+	  throw std::runtime_error("bad PDU type");
+	}
+      }
+
+      bool
+      type_matches(vector_type type, pmt::pmt_t v)
+      {
+	switch(type) {
+	case byte_t:
+	  return pmt::is_u8vector(v);
+	case float_t:
+	  return pmt::is_f32vector(v);
+	case complex_t:
+	  return pmt::is_c32vector(v);
+	default:
+	  throw std::runtime_error("bad PDU type");
+	}
+      }
+
+      pmt::pmt_t
+      make_vector(vector_type type, const uint8_t *buf, size_t items)
+      {
+	switch(type) {
+	case byte_t:
+	  return pmt::init_u8vector(items, buf);
+	case float_t:
+	  return pmt::init_f32vector(items, (const float *)buf);
+	case complex_t:
+	  return pmt::init_c32vector(items, (const gr_complex *)buf);
+	default:
+	  throw std::runtime_error("bad PDU type");
+	}
+      }
+
+      vector_type
+      type_from_pmt(pmt::pmt_t vector)
+      {
+	if(pmt::is_u8vector(vector))
+	  return byte_t;
+	if(pmt::is_f32vector(vector))
+	  return float_t;
+	if(pmt::is_c32vector(vector))
+	  return complex_t;
+	throw std::runtime_error("bad PDU type");
+      }
+
+    } /* namespace pdu */
+  } /* namespace blocks */
+} /* namespace gr */
diff --git a/gr-blocks/lib/pdu_to_tagged_stream_impl.cc b/gr-blocks/lib/pdu_to_tagged_stream_impl.cc
new file mode 100644
index 0000000000..1fc4857f2e
--- /dev/null
+++ b/gr-blocks/lib/pdu_to_tagged_stream_impl.cc
@@ -0,0 +1,120 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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 "pdu_to_tagged_stream_impl.h"
+#include <blocks/pdu.h>
+#include <gr_io_signature.h>
+
+namespace gr {
+  namespace blocks {
+
+    pdu_to_tagged_stream::sptr
+    pdu_to_tagged_stream::make(pdu::vector_type type)
+    {
+      return gnuradio::get_initial_sptr(new pdu_to_tagged_stream_impl(type));
+    }
+
+    pdu_to_tagged_stream_impl::pdu_to_tagged_stream_impl(pdu::vector_type type)
+      : gr_sync_block("pdu_to_tagged_stream",
+		      gr_make_io_signature(0, 0, 0),
+		      gr_make_io_signature(1, 1, pdu::itemsize(type))),
+	d_itemsize(pdu::itemsize(type)),
+	d_type(type)
+    {
+      message_port_register_in(PDU_PORT_ID);
+    }
+
+    int
+    pdu_to_tagged_stream_impl::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_blocking(PDU_PORT_ID));
+	if (msg.get() == NULL)
+	  return nout;
+
+	// make sure type is valid
+	if (!pmt::is_pair(msg)) // TODO: implement pdu::is_valid()
+	  throw std::runtime_error("received a malformed pdu message");
+	
+	// grab the components of the pdu message
+	pmt::pmt_t meta(pmt::car(msg));
+	pmt::pmt_t vect(pmt::cdr(msg));
+
+	// 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::from_long(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::dict_keys(meta));
+
+	  while (!pmt::eq(pair, pmt::PMT_NIL) ) {
+            pmt::pmt_t k(pmt::cdr(pair));
+            pmt::pmt_t v(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::length(vect));
+	size_t nsave = pmt::length(vect) - ncopy;
+ 
+	// copy output
+	size_t io(0);
+	nout += ncopy;
+	memcpy(out, 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], uniform_vector_elements(vect,ncopy), nsave*d_itemsize);
+        }
+      }
+      
+      return nout;
+    }
+
+  } /* namespace blocks */
+} /* namespace gr */
diff --git a/gr-blocks/lib/pdu_to_tagged_stream_impl.h b/gr-blocks/lib/pdu_to_tagged_stream_impl.h
new file mode 100644
index 0000000000..ca1c6437bd
--- /dev/null
+++ b/gr-blocks/lib/pdu_to_tagged_stream_impl.h
@@ -0,0 +1,48 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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_PDU_TO_TAGGED_STREAM_IMPL_H
+#define INCLUDED_PDU_TO_TAGGED_STREAM_IMPL_H
+
+#include <blocks/pdu_to_tagged_stream.h>
+
+namespace gr {
+  namespace blocks {
+
+    class BLOCKS_API pdu_to_tagged_stream_impl : public pdu_to_tagged_stream
+    {
+      size_t               d_itemsize;
+      pdu::vector_type     d_type;
+      std::vector<uint8_t> d_remain;
+
+    public:
+      pdu_to_tagged_stream_impl(pdu::vector_type type);
+
+      int work(int noutput_items,
+	       gr_vector_const_void_star &input_items,
+	       gr_vector_void_star &output_items);
+    };
+
+  } /* namespace blocks */
+} /* namespace gr */
+
+#endif 
diff --git a/gr-blocks/lib/socket_pdu_impl.cc b/gr-blocks/lib/socket_pdu_impl.cc
new file mode 100644
index 0000000000..20338f9968
--- /dev/null
+++ b/gr-blocks/lib/socket_pdu_impl.cc
@@ -0,0 +1,198 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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 "socket_pdu_impl.h"
+#include "tcp_connection.h"
+#include <gr_io_signature.h>
+#include <gr_pdu.h>
+
+namespace gr {
+  namespace blocks {
+
+    socket_pdu::sptr
+    socket_pdu::make(std::string type, std::string addr, std::string port, int MTU)
+    {
+      return gnuradio::get_initial_sptr(new socket_pdu_impl(type, addr, port, MTU));
+    }
+
+    socket_pdu_impl::socket_pdu_impl(std::string type, std::string addr, std::string port, int MTU)
+      :	gr_block("socket_pdu",
+		 gr_make_io_signature (0, 0, 0),
+		 gr_make_io_signature (0, 0, 0))
+    {
+      message_port_register_in(PDU_PORT_ID);
+      message_port_register_out(PDU_PORT_ID);
+
+      if ((type == "TCP_SERVER") || (type == "TCP_CLIENT")) {
+        boost::asio::ip::tcp::resolver resolver(d_io_service);
+        boost::asio::ip::tcp::resolver::query query(boost::asio::ip::tcp::v4(), addr, port);
+	d_tcp_endpoint = *resolver.resolve(query);
+      }
+
+      if ((type == "UDP_SERVER") || (type == "UDP_CLIENT")) {
+	boost::asio::ip::udp::resolver resolver(d_io_service);
+        boost::asio::ip::udp::resolver::query query(boost::asio::ip::udp::v4(), addr, port);
+
+        if (type == "UDP_SERVER")
+	  d_udp_endpoint = *resolver.resolve(query);
+        else
+	  d_udp_endpoint_other = *resolver.resolve(query);
+      }
+
+      if (type == "TCP_SERVER") {
+        d_acceptor_tcp.reset(new boost::asio::ip::tcp::acceptor(d_io_service, d_tcp_endpoint));
+        d_acceptor_tcp->set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
+        start_tcp_accept();
+        set_msg_handler(PDU_PORT_ID, boost::bind(&socket_pdu_impl::tcp_server_send, this, _1));
+
+      }
+      else if (type =="TCP_CLIENT") {
+        boost::system::error_code error = boost::asio::error::host_not_found;
+        d_tcp_socket.reset(new boost::asio::ip::tcp::socket(d_io_service));
+        d_tcp_socket->connect(d_tcp_endpoint, error);
+        if (error)
+            throw boost::system::system_error(error);
+
+        set_msg_handler(PDU_PORT_ID, boost::bind(&socket_pdu_impl::tcp_client_send, this, _1));
+
+        d_tcp_socket->async_read_some(
+	  boost::asio::buffer(d_rxbuf),
+	  boost::bind(&socket_pdu_impl::handle_tcp_read, this, 
+                      boost::asio::placeholders::error, 
+                      boost::asio::placeholders::bytes_transferred)
+	);
+      }
+      else if (type =="UDP_SERVER") {
+        d_udp_socket.reset(new boost::asio::ip::udp::socket(d_io_service, d_udp_endpoint));
+        d_udp_socket->async_receive_from(boost::asio::buffer(d_rxbuf), d_udp_endpoint_other, 
+					 boost::bind(&socket_pdu_impl::handle_udp_read, this,
+						     boost::asio::placeholders::error,
+						     boost::asio::placeholders::bytes_transferred));
+        set_msg_handler(PDU_PORT_ID, boost::bind(&socket_pdu_impl::udp_send, this, _1));
+      }
+      else if (type =="UDP_CLIENT") {
+        d_udp_socket.reset(new boost::asio::ip::udp::socket(d_io_service, d_udp_endpoint));
+        d_udp_socket->async_receive_from(boost::asio::buffer(d_rxbuf), d_udp_endpoint_other, 
+					 boost::bind(&socket_pdu_impl::handle_udp_read, this,
+						     boost::asio::placeholders::error,
+						     boost::asio::placeholders::bytes_transferred)); 
+        set_msg_handler(PDU_PORT_ID, boost::bind(&socket_pdu_impl::udp_send, this, _1));
+      }
+      else
+	throw std::runtime_error("gr::blocks:socket_pdu: unknown socket type");
+
+      d_thread = gruel::thread(boost::bind(&socket_pdu_impl::run_io_service, this));
+      d_started = true;
+    }
+
+    void
+    socket_pdu_impl::handle_tcp_read(const boost::system::error_code& error, size_t bytes_transferred)
+    {
+      if (!error) {
+	pmt::pmt_t vector = pmt::init_u8vector(bytes_transferred, (const uint8_t *)&d_rxbuf[0]);
+	pmt::pmt_t pdu = pmt::cons(pmt::PMT_NIL, vector);
+	message_port_pub(PDU_PORT_ID, pdu);
+
+	d_tcp_socket->async_read_some(boost::asio::buffer(d_rxbuf),
+				      boost::bind(&socket_pdu_impl::handle_tcp_read, this,
+						  boost::asio::placeholders::error,
+						  boost::asio::placeholders::bytes_transferred));
+      }
+      else
+        throw boost::system::system_error(error);
+    }
+
+    void
+    socket_pdu_impl::start_tcp_accept()
+    {
+      tcp_connection::sptr new_connection = tcp_connection::make(d_acceptor_tcp->get_io_service());
+
+      d_acceptor_tcp->async_accept(new_connection->socket(),
+				   boost::bind(&socket_pdu_impl::handle_tcp_accept, this, 
+					       new_connection, boost::asio::placeholders::error));
+    }
+
+    void
+    socket_pdu_impl::tcp_server_send(pmt::pmt_t msg)
+    {
+      pmt::pmt_t vector = pmt::cdr(msg);
+      for(size_t i = 0; i < d_tcp_connections.size(); i++)
+        d_tcp_connections[i]->send(vector);
+    }
+
+    void
+    socket_pdu_impl::handle_tcp_accept(tcp_connection::sptr 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
+    socket_pdu_impl::tcp_client_send(pmt::pmt_t msg)
+    {
+      pmt::pmt_t vector = pmt::cdr(msg);
+      size_t len = pmt::length(vector);
+      size_t offset(0);
+      boost::array<char, 10000> txbuf;
+      memcpy(&txbuf[0], pmt::uniform_vector_elements(vector, offset), len);
+      d_tcp_socket->send(boost::asio::buffer(txbuf,len));
+    }
+
+    void
+    socket_pdu_impl::udp_send(pmt::pmt_t msg)
+    {
+      pmt::pmt_t vector = pmt::cdr(msg);
+      size_t len = pmt::length(vector);
+      size_t offset(0);
+      boost::array<char, 10000> txbuf;
+      memcpy(&txbuf[0], pmt::uniform_vector_elements(vector, offset), len);
+      if (d_udp_endpoint_other.address().to_string() != "0.0.0.0")
+        d_udp_socket->send_to(boost::asio::buffer(txbuf,len), d_udp_endpoint_other);
+    }
+
+    void
+    socket_pdu_impl::handle_udp_read(const boost::system::error_code& error, size_t bytes_transferred)
+    {
+      if (!error) {
+        pmt::pmt_t vector = pmt::init_u8vector(bytes_transferred, (const uint8_t*)&d_rxbuf[0]);
+        pmt::pmt_t pdu = pmt::cons( pmt::PMT_NIL, vector);
+        
+        message_port_pub(PDU_PORT_ID, pdu);
+    
+        d_udp_socket->async_receive_from(boost::asio::buffer(d_rxbuf), d_udp_endpoint_other,
+					 boost::bind(&socket_pdu_impl::handle_udp_read, this,
+						     boost::asio::placeholders::error,
+						     boost::asio::placeholders::bytes_transferred));
+      } 
+    }
+
+  } /* namespace blocks */
+}/* namespace gr */
diff --git a/gr-blocks/lib/socket_pdu_impl.h b/gr-blocks/lib/socket_pdu_impl.h
new file mode 100644
index 0000000000..78602754c7
--- /dev/null
+++ b/gr-blocks/lib/socket_pdu_impl.h
@@ -0,0 +1,69 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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_BLOCKS_SOCKET_PDU_IMPL_H
+#define INCLUDED_BLOCKS_SOCKET_PDU_IMPL_H
+
+#include <blocks/socket_pdu.h>
+#include "stream_pdu_base.h"
+#include "tcp_connection.h"
+
+namespace gr {
+  namespace blocks {
+
+    class socket_pdu_impl : public socket_pdu, public stream_pdu_base
+    {
+    private:
+      boost::asio::io_service d_io_service;
+      boost::array<char, 10000> d_rxbuf;
+      void run_io_service() { d_io_service.run(); }
+
+      // TCP specific
+      boost::asio::ip::tcp::endpoint d_tcp_endpoint;
+      std::vector<tcp_connection::sptr> d_tcp_connections;
+      void handle_tcp_read(const boost::system::error_code& error, size_t bytes_transferred);
+
+      // TCP server specific
+      boost::shared_ptr<boost::asio::ip::tcp::acceptor> d_acceptor_tcp;
+      void start_tcp_accept();
+      void tcp_server_send(pmt::pmt_t msg);
+      void handle_tcp_accept(tcp_connection::sptr new_connection, const boost::system::error_code& error);
+
+      // TCP client specific
+      boost::shared_ptr<boost::asio::ip::tcp::socket> d_tcp_socket;
+      void tcp_client_send(pmt::pmt_t msg);
+
+      // UDP specific
+      boost::asio::ip::udp::endpoint d_udp_endpoint;
+      boost::asio::ip::udp::endpoint d_udp_endpoint_other;
+      boost::shared_ptr<boost::asio::ip::udp::socket> d_udp_socket;
+      void handle_udp_read(const boost::system::error_code& error, size_t bytes_transferred);
+      void udp_send(pmt::pmt_t msg);
+    
+    public:
+      socket_pdu_impl(std::string type, std::string addr, std::string port, int MTU);
+    };
+
+  } /* namespace blocks */
+} /* namespace gr */
+
+#endif /* INCLUDED_BLOCKS_SOCKET_PDU_IMPL_H */
diff --git a/gr-blocks/lib/stream_pdu_base.cc b/gr-blocks/lib/stream_pdu_base.cc
index 14e76149a0..ed3dfcde10 100644
--- a/gr-blocks/lib/stream_pdu_base.cc
+++ b/gr-blocks/lib/stream_pdu_base.cc
@@ -28,7 +28,7 @@
 #include <io.h>
 #endif
 
-#include <gr_pdu.h>
+#include <blocks/pdu.h>
 #include <gr_basic_block.h>
 #include "stream_pdu_base.h"
 #include <boost/format.hpp>
@@ -112,12 +112,12 @@ namespace gr {
     {
       pmt::pmt_t vector = pmt::cdr(msg);
       size_t offset(0);
-      size_t itemsize(::gr_pdu_itemsize(type_from_pmt(vector)));
+      size_t itemsize(pdu::itemsize(pdu::type_from_pmt(vector)));
       int len(pmt::length(vector)*itemsize);
     
       const int rv = write(d_fd, 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)")
+        std::cerr << boost::format("WARNING: stream_pdu_base::send(pdu) write failed! (d_fd=%d, len=%d, rv=%d)")
 	  % d_fd % len % rv << std::endl;
       }
     }
diff --git a/gr-blocks/lib/tagged_stream_to_pdu_impl.cc b/gr-blocks/lib/tagged_stream_to_pdu_impl.cc
new file mode 100644
index 0000000000..fc5c8f6424
--- /dev/null
+++ b/gr-blocks/lib/tagged_stream_to_pdu_impl.cc
@@ -0,0 +1,136 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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 "tagged_stream_to_pdu_impl.h"
+#include <blocks/pdu.h>
+#include <gr_io_signature.h>
+
+namespace gr {
+  namespace blocks {
+
+    tagged_stream_to_pdu::sptr
+    tagged_stream_to_pdu::make(pdu::vector_type type)
+    {
+      return gnuradio::get_initial_sptr(new tagged_stream_to_pdu_impl(type));
+    }
+
+    tagged_stream_to_pdu_impl::tagged_stream_to_pdu_impl(pdu::vector_type type)
+      : gr_sync_block("tagged_stream_to_pdu",
+		      gr_make_io_signature(1, 1, pdu::itemsize(type)),
+		      gr_make_io_signature(0, 0, 0)),
+	d_itemsize(pdu::itemsize(type)),
+	d_inpdu(false),
+	d_type(type),
+	d_pdu_meta(pmt::PMT_NIL),
+	d_pdu_vector(pmt::PMT_NIL)
+    {
+      message_port_register_out(PDU_PORT_ID);
+    }
+
+    int
+    tagged_stream_to_pdu_impl::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) {
+	bool found_length_tag(false);
+
+	get_tags_in_range(d_tags, 0, abs_N, abs_N+1);
+
+	for (d_tags_itr = d_tags.begin(); (d_tags_itr != d_tags.end()) && (!found_length_tag); d_tags_itr++) {
+	  if (pmt::eq((*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::to_long((*d_tags_itr).value);
+	  d_pdu_remain = d_pdu_length;
+	  d_pdu_meta = 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::eq((*d_tags_itr).key, PDU_LENGTH_TAG ))
+	  d_pdu_meta = 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 = pdu::make_vector(d_type, 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 = pdu::make_vector(d_type, &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
+    tagged_stream_to_pdu_impl::send_message()
+    {
+      if (pmt::length(d_pdu_vector) != d_pdu_length)
+        throw std::runtime_error("msg length not correct");
+      
+      pmt::pmt_t msg = 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;
+    }
+    
+  } /* namespace blocks */
+} /* namespace gr */
diff --git a/gr-blocks/lib/tagged_stream_to_pdu_impl.h b/gr-blocks/lib/tagged_stream_to_pdu_impl.h
new file mode 100644
index 0000000000..84d7f6c3cb
--- /dev/null
+++ b/gr-blocks/lib/tagged_stream_to_pdu_impl.h
@@ -0,0 +1,58 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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_TAGGED_STREAM_TO_PDU_IMPL_H
+#define INCLUDED_TAGGED_STREAM_TO_PDU_IMPL_H
+
+#include <blocks/tagged_stream_to_pdu.h>
+
+namespace gr {
+  namespace blocks {
+
+    class BLOCKS_API tagged_stream_to_pdu_impl : public tagged_stream_to_pdu
+    {
+      size_t               d_itemsize;
+      size_t               d_pdu_length;
+      size_t               d_pdu_remain;
+      bool                 d_inpdu;
+      pdu::vector_type     d_type;
+      std::vector<uint8_t> d_save;
+      pmt::pmt_t           d_pdu_meta;
+      pmt::pmt_t           d_pdu_vector;
+
+      std::vector<gr_tag_t>::iterator d_tags_itr;
+      std::vector<gr_tag_t> d_tags;
+      
+    public:
+      tagged_stream_to_pdu_impl(pdu::vector_type type);
+
+      int work(int noutput_items,
+	       gr_vector_const_void_star &input_items,
+	       gr_vector_void_star &output_items);
+
+      void send_message();
+    };
+
+  } /* namespace blocks */
+} /* namespace gr */
+
+#endif 
diff --git a/gr-blocks/lib/tcp_connection.cc b/gr-blocks/lib/tcp_connection.cc
new file mode 100644
index 0000000000..f7af371250
--- /dev/null
+++ b/gr-blocks/lib/tcp_connection.cc
@@ -0,0 +1,87 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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 "tcp_connection.h"
+#include <gr_basic_block.h>
+#include <gr_pdu.h>
+//#include <boost/asio.hpp>
+//#include <boost/bind.hpp>
+
+namespace gr {
+  namespace blocks {
+
+    tcp_connection::sptr tcp_connection::make(boost::asio::io_service& io_service)
+    {
+      return sptr(new tcp_connection(io_service));
+    }
+
+    tcp_connection::tcp_connection(boost::asio::io_service& io_service)
+      : d_socket(io_service)
+    {
+    }
+
+    void
+    tcp_connection::send(pmt::pmt_t vector)
+    {
+      size_t len = pmt::length(vector);
+      size_t offset(0);
+      boost::array<char, 10000> txbuf;
+      memcpy(&txbuf[0], pmt::uniform_vector_elements(vector, offset), len);
+      boost::asio::async_write(d_socket, boost::asio::buffer(txbuf, len),
+			       boost::bind(&tcp_connection::handle_write, this,
+					   boost::asio::placeholders::error,
+					   boost::asio::placeholders::bytes_transferred));
+    }
+
+    void
+    tcp_connection::start(gr_basic_block *block)
+    {
+      d_block = block;
+      d_socket.async_read_some(boost::asio::buffer(d_buf),
+			       boost::bind(&tcp_connection::handle_read, this, 
+					   boost::asio::placeholders::error,
+					   boost::asio::placeholders::bytes_transferred));
+    }
+
+    void
+    tcp_connection::handle_read(const boost::system::error_code& error, size_t bytes_transferred)
+    {
+      if (!error) {
+        pmt::pmt_t vector = pmt::init_u8vector(bytes_transferred, (const uint8_t*)&d_buf[0]);
+        pmt::pmt_t pdu = pmt::cons( pmt::PMT_NIL, vector);
+
+        d_block->message_port_pub(PDU_PORT_ID, pdu);
+
+        d_socket.async_read_some(boost::asio::buffer(d_buf),
+				 boost::bind(&tcp_connection::handle_read, this,
+					     boost::asio::placeholders::error,
+					     boost::asio::placeholders::bytes_transferred));
+
+      }
+    }
+
+  } /* namespace blocks */
+}/* namespace gr */
diff --git a/gr-blocks/lib/tcp_connection.h b/gr-blocks/lib/tcp_connection.h
new file mode 100644
index 0000000000..94755e847a
--- /dev/null
+++ b/gr-blocks/lib/tcp_connection.h
@@ -0,0 +1,60 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2013 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_TCP_CONNECTION_H
+#define INCLUDED_TCP_CONNECTION_H
+
+#include <boost/asio.hpp>
+#include <gruel/pmt.h>
+
+class gr_basic_block;
+
+namespace gr {
+  namespace blocks {
+
+    class tcp_connection
+    {
+    private:
+      boost::asio::ip::tcp::socket d_socket;
+      boost::array<char, 10000> d_buf;
+      std::string d_message;
+      gr_basic_block *d_block;
+
+      tcp_connection(boost::asio::io_service& io_service);
+      
+    public:
+      typedef boost::shared_ptr<tcp_connection> sptr;
+      
+      static sptr make(boost::asio::io_service& io_service);
+
+      boost::asio::ip::tcp::socket& socket() { return d_socket; };
+
+      void start(gr_basic_block *block);
+      void send(pmt::pmt_t vector);
+      void handle_read(const boost::system::error_code& error, size_t bytes_transferred);
+      void handle_write(const boost::system::error_code& error, size_t bytes_transferred) { }
+    };
+
+  } /* namespace blocks */
+} /* namespace gr */
+
+#endif /* INCLUDED_TCP_CONNECTION_H */
diff --git a/gr-blocks/lib/tuntap_pdu_impl.cc b/gr-blocks/lib/tuntap_pdu_impl.cc
index 1970a92b69..8de817738f 100644
--- a/gr-blocks/lib/tuntap_pdu_impl.cc
+++ b/gr-blocks/lib/tuntap_pdu_impl.cc
@@ -26,7 +26,7 @@
 
 #include "tuntap_pdu_impl.h"
 #include <gr_io_signature.h>
-#include <gr_pdu.h>
+#include <blocks/pdu.h>
 #include <boost/format.hpp>
 
 #include <sys/types.h>
author	Johnathan Corgan <johnathan@corganlabs.com>	2013-02-28 13:18:28 -0800
committer	Johnathan Corgan <johnathan@corganlabs.com>	2013-02-28 13:38:37 -0800
commit	4a99e708ad0dd54bae4a4275ae2c06e44decbebf (patch)
tree	5ec7ac397cdfce5532e101edc84f9c11d26a5569 /gr-blocks/lib
parent	2f55d7dfc33e8d990e44c5bbb7c6d2fbdaddd563 (diff)
parent	8a49689fbd5d55d74033ad398cab6862fcb17acc (diff)