merging r4318:4437 to fix ticket:131 from branche trondeau/udp udp source/sink pairs working

git-svn-id: http://gnuradio.org/svn/gnuradio/trunk@4438 221aa14e-8319-0410-a670-987f0aec2ac5

6 files changed, 274 insertions, 143 deletions

diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.cc b/gnuradio-core/src/lib/io/gr_udp_sink.cc
index 43d0dd664..d799262b4 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.cc
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -20,42 +20,59 @@
  * Boston, MA 02110-1301, USA.
  */
 
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
 #include <gr_udp_sink.h>
 #include <gr_io_signature.h>
-#include <cstdio>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
 #include <stdexcept>
+#include <netdb.h>
 
 #define SNK_VERBOSE 0
 
 gr_udp_sink::gr_udp_sink (size_t itemsize, 
-			  const char *ipaddrl, unsigned short portl,
-			  const char *ipaddrr, unsigned short portr,
-			  unsigned int mtu)
+			  const char *src, unsigned short port_src,
+			  const char *dst, unsigned short port_dst,
+			  int payload_size)
   : gr_sync_block ("udp_sink",
 		   gr_make_io_signature (1, 1, itemsize),
 		   gr_make_io_signature (0, 0, 0)),
-    d_itemsize (itemsize), d_updated(false), d_mtu(mtu)
+    d_itemsize (itemsize), d_updated(false), d_payload_size(payload_size)
 {
-  // Set up the address stucture for the local address and port numbers
-  inet_aton(ipaddrl, &d_ipaddr_local);     // format IP address
-  inet_aton(ipaddrr, &d_ipaddr_remote);    // format IP address
-  d_port_local  = htons(portl);            // format port number
-  d_port_remote = htons(portr);            // format port number
-
-  d_sockaddr_local.sin_family = AF_INET;
-  d_sockaddr_local.sin_addr   = d_ipaddr_local;
-  d_sockaddr_local.sin_port   = d_port_local;
-
-  d_sockaddr_remote.sin_family = AF_INET;
-  d_sockaddr_remote.sin_addr   = d_ipaddr_remote;
-  d_sockaddr_remote.sin_port   = d_port_remote;
+  int ret = 0;
+  
+  // Set up the address stucture for the source address and port numbers
+  // Get the source IP address from the host name
+  struct hostent *hsrc = gethostbyname(src);
+  if(hsrc) {   // if the source was provided as a host namex
+    d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];    
+  }
+  else { // assume it was specified as an IP address
+    if((ret=inet_aton(src, &d_ip_src)) == 0) {            // format IP address
+      perror("Not a valid source IP address or host name");
+      throw std::runtime_error("can't initialize source socket");
+    }
+  }
+
+  // Get the destination IP address from the host name
+  struct hostent *hdst = gethostbyname(dst);
+  if(hdst) {   // if the source was provided as a host namex
+    d_ip_dst = *(struct in_addr*)hdst->h_addr_list[0];    
+  }
+  else { // assume it was specified as an IP address
+    if((ret=inet_aton(dst, &d_ip_dst)) == 0) {            // format IP address
+      perror("Not a valid destination IP address or host name");
+      throw std::runtime_error("can't initialize destination socket");
+    }
+  }
+
+  d_port_src = htons(port_src);           // format port number
+  d_port_dst = htons(port_dst);           // format port number
+
+  d_sockaddr_src.sin_family = AF_INET;
+  d_sockaddr_src.sin_addr   = d_ip_src;
+  d_sockaddr_src.sin_port   = d_port_src;
+
+  d_sockaddr_dst.sin_family = AF_INET;
+  d_sockaddr_dst.sin_addr   = d_ip_dst;
+  d_sockaddr_dst.sin_port   = d_port_dst;
   
   open();
 }
@@ -64,14 +81,14 @@ gr_udp_sink::gr_udp_sink (size_t itemsize,
 
 gr_udp_sink_sptr
 gr_make_udp_sink (size_t itemsize, 
-		  const char *ipaddrl, unsigned short portl,
-		  const char *ipaddrr, unsigned short portr,
-		  unsigned int mtu)
+		  const char *src, unsigned short port_src,
+		  const char *dst, unsigned short port_dst,
+		  int payload_size)
 {
   return gr_udp_sink_sptr (new gr_udp_sink (itemsize, 
-						      ipaddrl, portl,
-						      ipaddrr, portr,
-						      mtu));
+					    src, port_src,
+					    dst, port_dst,
+					    payload_size));
 }
 
 gr_udp_sink::~gr_udp_sink ()
@@ -91,7 +108,7 @@ gr_udp_sink::open()
   }
 
   // Turn on reuse address
-  bool opt_val = true;
+  int opt_val = true;
   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (void*)&opt_val, sizeof(int)) == -1) {
     perror("SO_REUSEADDR");
     throw std::runtime_error("can't set socket option SO_REUSEADDR");
@@ -107,13 +124,13 @@ gr_udp_sink::open()
   }
 
   // bind socket to an address and port number to listen on
-  if(bind (d_socket, (sockaddr*)&d_sockaddr_local, sizeof(struct sockaddr)) == -1) {
+  if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
     perror("socket bind");
     throw std::runtime_error("can't bind socket");
   }
 
   // Not sure if we should throw here or allow retries
-  if(connect(d_socket, (sockaddr*)&d_sockaddr_remote, sizeof(struct sockaddr)) == -1) {
+  if(connect(d_socket, (sockaddr*)&d_sockaddr_dst, sizeof(struct sockaddr)) == -1) {
     perror("socket connect");
     throw std::runtime_error("can't connect to socket");
   }
@@ -139,14 +156,27 @@ gr_udp_sink::work (int noutput_items,
 		   gr_vector_const_void_star &input_items,
 		   gr_vector_void_star &output_items)
 {
-  char *in = (char *) input_items[0];
-  socklen_t bytes=0, bytes_sent=0, bytes_to_send=0;
-  unsigned int total_size = noutput_items*d_itemsize;
-
-  while(bytes_sent < total_size) {
-    bytes_to_send = (bytes_sent+d_mtu < total_size ? d_mtu : total_size-bytes_sent);
-    bytes = send(d_socket, (in+bytes_sent), bytes_to_send, MSG_DONTWAIT);
-    bytes_sent += bytes;
+  const char *in = (const char *) input_items[0];
+  ssize_t r=0, bytes_sent=0, bytes_to_send=0;
+  ssize_t total_size = noutput_items*d_itemsize;
+
+  #if SNK_VERBOSE
+  printf("Entered upd_sink\n");
+  #endif
+
+  while(bytes_sent <  total_size) {
+    bytes_to_send = std::min(d_payload_size, (total_size-bytes_sent));
+  
+    r = send(d_socket, (in+bytes_sent), bytes_to_send, 0);
+    if(r == -1) {         // error on send command
+      perror("udp_sink"); // there should be no error case where this function 
+      return -1;          // should not exit immediately
+    }
+    bytes_sent += r;
+    
+    #if SNK_VERBOSE
+    printf("\tbyte sent: %d bytes\n", bytes);
+    #endif
   }
 
   #if SNK_VERBOSE
diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.h b/gnuradio-core/src/lib/io/gr_udp_sink.h
index ee346f40c..9263ade79 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.h
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.h
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -31,43 +31,66 @@
 class gr_udp_sink;
 typedef boost::shared_ptr<gr_udp_sink> gr_udp_sink_sptr;
 
+gr_udp_sink_sptr
+gr_make_udp_sink (size_t itemsize, 
+		  const char *src, unsigned short port_src,
+		  const char *dst, unsigned short port_dst,
+		  int payload_size=1472);
+
 /*!
- * \brief Write stream to an Udp port (over UDP).
+ * \brief Write stream to an UDP socket.
  * \ingroup sink
+ * 
+ * \param itemsize     The size (in bytes) of the item datatype
+ * \param src          The source address as either the host name or the 'numbers-and-dots'
+ *                     IP address
+ * \param port_src     Destination port to bind to (0 allows socket to choose an appropriate port)
+ * \param dst          The destination address as either the host name or the 'numbers-and-dots'
+ *                     IP address
+ * \param port_dst     Destination port to connect to
+ * \param payload_size UDP payload size by default set to 
+ *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
  */
 
-gr_udp_sink_sptr
-gr_make_udp_sink (size_t itemsize, 
-		  const char *ipaddrl, unsigned short portl,
-		  const char *ipaddrr, unsigned short portr,
-		  unsigned int mtu=540);
-
 class gr_udp_sink : public gr_sync_block
 {
   friend gr_udp_sink_sptr gr_make_udp_sink (size_t itemsize, 
-					    const char *ipaddrl, unsigned short portl,
-					    const char *ipaddrr, unsigned short portr,
-					    unsigned int mtu);
+					    const char *src, unsigned short port_src,
+					    const char *dst, unsigned short port_dst,
+					    int payload_size);
  private:
   size_t	d_itemsize;
   bool		d_updated;
   omni_mutex	d_mutex;
 
-  unsigned int   d_mtu;             // maximum transmission unit (packet length)
+  int            d_payload_size;    // maximum transmission unit (packet length)
   int            d_socket;          // handle to socket
   int            d_socket_rcv;      // handle to socket retuned in the accept call
-  struct in_addr d_ipaddr_local;    // store the local IP address to use
-  struct in_addr d_ipaddr_remote;   // store the remote IP address that connected to us
-  unsigned short d_port_local;      // the port number to open for connections to this service
-  unsigned short d_port_remote;     // port number of the remove system
-  sockaddr_in    d_sockaddr_local;  // store the local sockaddr data (formatted IP address and port number)
-  sockaddr_in    d_sockaddr_remote; // store the remote sockaddr data (formatted IP address and port number)
+  struct in_addr d_ip_src;          // store the source ip info
+  struct in_addr d_ip_dst;          // store the destination ip info
+  unsigned short d_port_src;        // the port number to open for connections to this service
+  unsigned short d_port_dst;        // port number of the remove system
+  sockaddr_in    d_sockaddr_src;    // store the source sockaddr data (formatted IP address and port number)
+  sockaddr_in    d_sockaddr_dst;    // store the destination sockaddr data (formatted IP address and port number)
 
  protected:
+  /*!
+   * \brief UDP Sink Constructor
+   * 
+   * \param itemsize     The size (in bytes) of the item datatype
+   * \param src          The source address as either the host name or the 'numbers-and-dots'
+   *                     IP address
+   * \param port_src     Destination port to bind to (0 allows socket to choose an appropriate port)
+   * \param dst          The destination address as either the host name or the 'numbers-and-dots'
+   *                     IP address
+   * \param port_dst     Destination port to connect to
+   * \param payload_size UDP payload size by default set to 
+   *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+   */
   gr_udp_sink (size_t itemsize, 
-		    const char *ipaddrl, unsigned short portl,
-		    const char *ipaddrr, unsigned short portr,
-		    unsigned int mtu);
+	       const char *src, unsigned short port_src,
+	       const char *dst, unsigned short port_dst,
+	       int payload_size);
 
  public:
   ~gr_udp_sink ();
@@ -87,11 +110,8 @@ class gr_udp_sink : public gr_sync_block
    */
   void close();
 
-  /*! \brief set the MTU of the socket */
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-
-  /*! \brief return the MTU of the socket */
-  unsigned int mtu() { return d_mtu; }
+  /*! \brief return the PAYLOAD_SIZE of the socket */
+  int payload_size() { return d_payload_size; }
 
   // should we export anything else?
 
diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.i b/gnuradio-core/src/lib/io/gr_udp_sink.i
index bd4accb3e..ad3a452d0 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.i
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.i
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -25,22 +25,21 @@ GR_SWIG_BLOCK_MAGIC(gr,udp_sink)
 
 gr_udp_sink_sptr 
 gr_make_udp_sink (size_t itemsize, 
-		  const char *ipaddrl, unsigned short portl,
-		  const char *ipaddrr, unsigned short portr,
-		  unsigned int mtu=540);
+		  const char *src, unsigned short port_src,
+		  const char *dst, unsigned short port_dst,
+		  int payload_size=1472);
 
 class gr_udp_sink : public gr_sync_block
 {
  protected:
   gr_udp_sink (size_t itemsize, 
-	       const char *ipaddrl, unsigned short portl,
-	       const char *ipaddrr, unsigned short portr,
-	       unsigned int mtu);
+	       const char *src, unsigned short port_src,
+	       const char *dst, unsigned short port_dst,
+	       int payload_size);
 
   bool open();
   void close();
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-  unsigned int mtu() { return d_mtu; }
+  int payload_size() { return d_payload_size; }
 
  public:
   ~gr_udp_sink ();
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.cc b/gnuradio-core/src/lib/io/gr_udp_source.cc
index e2def4c37..e12d33b80 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.cc
+++ b/gnuradio-core/src/lib/io/gr_udp_source.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -20,48 +20,58 @@
  * Boston, MA 02110-1301, USA.
  */
 
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
 #include <gr_udp_source.h>
 #include <gr_io_signature.h>
-#include <cstdio>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
 #include <stdexcept>
+#include <errno.h>
+#include <netdb.h>
 
 #define SRC_VERBOSE 0
 
-gr_udp_source::gr_udp_source(size_t itemsize, const char *ipaddr, 
-			     unsigned short port, unsigned int mtu)
+gr_udp_source::gr_udp_source(size_t itemsize, const char *src, 
+			     unsigned short port_src, int payload_size)
   : gr_sync_block ("udp_source",
 		   gr_make_io_signature(0, 0, 0),
 		   gr_make_io_signature(1, 1, itemsize)),
-    d_itemsize(itemsize), d_updated(false), d_mtu(mtu)
+    d_itemsize(itemsize), d_updated(false), d_payload_size(payload_size), d_residual(0), d_temp_offset(0)
 {
-  // Set up the address stucture for the local address and port numbers
-  inet_aton(ipaddr, &d_ipaddr_local);     // format IP address
-  d_port_local = htons(port);             // format port number
+  int ret = 0;
+  
+  // Set up the address stucture for the source address and port numbers
+  // Get the source IP address from the host name
+  struct hostent *hsrc = gethostbyname(src);
+  if(hsrc) {   // if the source was provided as a host namex
+    d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];    
+  }
+  else { // assume it was specified as an IP address
+    if((ret=inet_aton(src, &d_ip_src)) == 0) {            // format IP address
+      perror("Not a valid source IP address or host name");
+      throw std::runtime_error("can't initialize source socket");
+    }
+  }
+
+  d_port_src = htons(port_src);     // format port number
   
-  d_sockaddr_local.sin_family = AF_INET;
-  d_sockaddr_local.sin_addr   = d_ipaddr_local;
-  d_sockaddr_local.sin_port   = d_port_local;
+  d_sockaddr_src.sin_family = AF_INET;
+  d_sockaddr_src.sin_addr   = d_ip_src;
+  d_sockaddr_src.sin_port   = d_port_src;
+
+  d_temp_buff = new char[d_payload_size];   // allow it to hold up to payload_size bytes
   
   open();
 }
 
 gr_udp_source_sptr
 gr_make_udp_source (size_t itemsize, const char *ipaddr, 
-		    unsigned short port, unsigned int mtu)
+		    unsigned short port, int payload_size)
 {
   return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr, 
-						port, mtu));
+						port, payload_size));
 }
 
 gr_udp_source::~gr_udp_source ()
 {
+  delete [] d_temp_buff;
   close();
 }
 
@@ -69,16 +79,15 @@ bool
 gr_udp_source::open()
 {
   omni_mutex_lock l(d_mutex);	// hold mutex for duration of this function
-   
   // create socket
-  d_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+  d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
   if(d_socket == -1) {
     perror("socket open");
     throw std::runtime_error("can't open socket");
   }
 
   // Turn on reuse address
-  bool opt_val = true;
+  int opt_val = 1;
   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (void*)&opt_val, sizeof(int)) == -1) {
     perror("SO_REUSEADDR");
     throw std::runtime_error("can't set socket option SO_REUSEADDR");
@@ -104,7 +113,7 @@ gr_udp_source::open()
   }
 
   // bind socket to an address and port number to listen on
-  if(bind (d_socket, (sockaddr*)&d_sockaddr_local, sizeof(struct sockaddr)) == -1) {
+  if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
     perror("socket bind");
     throw std::runtime_error("can't bind socket");
   }
@@ -131,33 +140,86 @@ gr_udp_source::work (int noutput_items,
 		     gr_vector_void_star &output_items)
 {
   char *out = (char *) output_items[0];
-  socklen_t bytes_to_receive=0, bytes_received=0;
-  int bytes=0;
+  ssize_t r=0, nbytes=0, bytes_received=0;
+  ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
+
+  #if SRC_VERBOSE
+  printf("\nEntered udp_source\n");
+  #endif
 
-  while((bytes_received < (unsigned)noutput_items) && (bytes>-1)) {
-    // caclulate the number of byte left if we can fit in all d_mtu bytes
-    bytes_to_receive = (bytes_received+d_mtu < noutput_items ? 
-			d_mtu : noutput_items-bytes_received);
+  // Remove items from temp buffer if they are in there
+  if(d_residual) {
+    nbytes = std::min(d_residual, total_bytes);
+    memcpy(out, d_temp_buff+d_temp_offset, nbytes);
+    bytes_received = nbytes;
+
+    #if SRC_VERBOSE
+    printf("\tTemp buff size: %d  offset: %d (bytes_received: %d) (noutput_items: %d)\n", 
+	   d_residual, d_temp_offset, bytes_received, noutput_items);
+    #endif
+
+    // Increment pointer
+    out += bytes_received;
     
+    // Update indexing of amount of bytes left in the buffer
+    d_residual -= nbytes;
+    d_temp_offset = d_temp_offset+d_residual;
+  }
+
+  while(1) {
     // get the data into our output buffer and record the number of bytes
-    // This is a blocking call, but it's timeout has been set in the constructor
-    bytes = recv(d_socket, out, bytes_to_receive, 0);
+    // This is a non-blocking call with a timeout set in the constructor
+    r = recv(d_socket, d_temp_buff, d_payload_size, 0);  // get the entire payload or the what's available
+
+    // Check if there was a problem; forget it if the operation just timed out
+    if(r == -1) {
+      if(errno == EAGAIN) {  // handle non-blocking call timeout
+        #if SRC_VERBOSE
+	printf("UDP receive timed out\n"); 
+        #endif
+
+	// Break here to allow the rest of the flow graph time to run and so ctrl-C breaks
+	break;
+      }
+      else {
+	perror("udp_source");
+	return -1;
+      }
+    }
+    else {
+      // Calculate the number of bytes we can take from the buffer in this call
+      nbytes = std::min(r, total_bytes-bytes_received);
+      
+      // adjust the total number of bytes we have to round down to nearest integer of an itemsize
+      nbytes -= ((bytes_received+nbytes) % d_itemsize);   
 
-    // FIXME if bytes < 0 bail
+      // copy the number of bytes we want to look at here
+      memcpy(out, d_temp_buff, nbytes);    
+
+      d_residual = r - nbytes;                      // save the number of bytes stored
+      d_temp_offset=nbytes;                         // reset buffer index
 
-    if(bytes > 0) {
       // keep track of the total number of bytes received
-      bytes_received += bytes;
+      bytes_received += nbytes;
 
       // increment the pointer
-      out += bytes;
+      out += nbytes;
+
+      // Immediately return when data comes in
+      break;
     }
+
+    #if SNK_VERBOSE
+    printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
+    #endif
   }
 
   #if SRC_VERBOSE
-  printf("\nTotal Bytes Received: %d (noutput_items=%d)\n", bytes_received, noutput_items); 
+  printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n", 
+	 bytes_received, bytes_received, noutput_items);
   #endif
 
-  // FIXME what if (bytes_received % d_itemsize) != 0 ???
-  return int(bytes_received / d_itemsize);
+  // bytes_received is already set to some integer multiple of itemsize
+  return bytes_received/d_itemsize;
 }
+
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.h b/gnuradio-core/src/lib/io/gr_udp_source.h
index c57f4fc8f..5e4a81cb7 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.h
+++ b/gnuradio-core/src/lib/io/gr_udp_source.h
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -31,31 +31,55 @@
 class gr_udp_source;
 typedef boost::shared_ptr<gr_udp_source> gr_udp_source_sptr;
 
-gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *ipaddr, 
-				      unsigned short port, unsigned int mtu=540);
+gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *src, 
+				      unsigned short port_src, int payload_size=1472);
+
+/*! 
+ * \brief Read stream from an UDP socket.
+ * \ingroup sink
+ *
+ * \param itemsize     The size (in bytes) of the item datatype
+ * \param src          The source address as either the host name or the 'numbers-and-dots'
+ *                     IP address
+ * \param port_src     The port number on which the socket listens for data
+ * \param payload_size UDP payload size by default set to 
+ *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+ *
+*/
 
 class gr_udp_source : public gr_sync_block
 {
-  friend gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *ipaddr, 
-					       unsigned short port, unsigned int mtu);
+  friend gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *src, 
+					       unsigned short port_src, int payload_size);
 
  private:
   size_t	d_itemsize;
   bool		d_updated;
   omni_mutex	d_mutex;
 
-  unsigned int   d_mtu;           // maximum transmission unit (packet length)
+  int            d_payload_size;  // maximum transmission unit (packet length)
   int            d_socket;        // handle to socket
   int            d_socket_rcv;    // handle to socket retuned in the accept call
-  struct in_addr d_ipaddr_local;  // store the local IP address to use
-  struct in_addr d_ipaddr_remote; // store the remote IP address that connected to us
-  unsigned short d_port_local;    // the port number to open for connections to this service
-  unsigned short d_port_remote;   // port number of the remove system
-  sockaddr_in    d_sockaddr_local;  // store the local sockaddr data (formatted IP address and port number)
-  sockaddr_in    d_sockaddr_remote; // store the remote sockaddr data (formatted IP address and port number)
-  
+  struct in_addr d_ip_src;        // store the source IP address to use
+  unsigned short d_port_src;      // the port number to open for connections to this service
+  sockaddr_in    d_sockaddr_src;  // store the source sockaddr data (formatted IP address and port number)
+
+  char *d_temp_buff;    // hold buffer between calls
+  ssize_t d_residual;   // hold information about number of bytes stored in the temp buffer
+  size_t d_temp_offset; // point to temp buffer location offset
+
  protected:
-  gr_udp_source(size_t itemsize, const char *ipaddr, unsigned short port, unsigned int mtu);
+  /*!
+   * \brief UDP Source Constructor
+   * 
+   * \param itemsize     The size (in bytes) of the item datatype
+   * \param src          The source address as either the host name or the 'numbers-and-dots'
+   *                     IP address
+   * \param port_src     The port number on which the socket listens for data
+   * \param payload_size UDP payload size by default set to 
+   *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+   */
+  gr_udp_source(size_t itemsize, const char *src, unsigned short port_src, int payload_size);
 
  public:
   ~gr_udp_source();
@@ -75,11 +99,8 @@ class gr_udp_source : public gr_sync_block
    */
   void close();
 
-  /*! \brief set the MTU of the socket */
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-
-  /*! \brief return the MTU of the socket */
-  unsigned int mtu() { return d_mtu; }
+  /*! \brief return the PAYLOAD_SIZE of the socket */
+  int payload_size() { return d_payload_size; }
 
   // should we export anything else?
 
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.i b/gnuradio-core/src/lib/io/gr_udp_source.i
index 1d6f4c904..5661f6e7d 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.i
+++ b/gnuradio-core/src/lib/io/gr_udp_source.i
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -23,21 +23,20 @@
 GR_SWIG_BLOCK_MAGIC(gr,udp_source)
 
 gr_udp_source_sptr 
-gr_make_udp_source (size_t itemsize, const char *ipaddr, 
-		    unsigned short port, unsigned int mtu=540);
+gr_make_udp_source (size_t itemsize, const char *src, 
+		    unsigned short port_src, int payload_size=1472);
 
 class gr_udp_source : public gr_sync_block
 {
  protected:
-  gr_udp_source (size_t itemsize, const char *ipaddr, 
-		 unsigned short port, unsigned int mtu);
+  gr_udp_source (size_t itemsize, const char *src, 
+		 unsigned short port_src, int payload_size);
 
  public:
   ~gr_udp_source ();
 
   bool open();
   void close();
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-  unsigned int mtu() { return d_mtu; }
+  int payload_size() { return d_payload_size; }
 
 };