3 * Copyright 2007,2008 Free Software Foundation, Inc.
5 * This file is part of GNU Radio
7 * GNU Radio is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License as published by
9 * the Free Software Foundation; either version 3, or (at your option)
12 * GNU Radio is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
17 * You should have received a copy of the GNU General Public License
18 * along with GNU Radio; see the file COPYING. If not, write to
19 * the Free Software Foundation, Inc., 51 Franklin Street,
20 * Boston, MA 02110-1301, USA.
26 #include <gr_udp_source.h>
27 #include <gr_io_signature.h>
32 #if defined(HAVE_SOCKET)
34 typedef void* optval_t;
37 #define inet_aton(N,A) ( (A)->s_addr = inet_addr(N), ( (A)->s_addr != INADDR_NONE ) )
38 typedef char* optval_t;
43 gr_udp_source::gr_udp_source(size_t itemsize, const char *src,
44 unsigned short port_src, int payload_size)
45 : gr_sync_block ("udp_source",
46 gr_make_io_signature(0, 0, 0),
47 gr_make_io_signature(1, 1, itemsize)),
48 d_itemsize(itemsize), d_updated(false), d_payload_size(payload_size), d_residual(0), d_temp_offset(0)
52 // Set up the address stucture for the source address and port numbers
53 // Get the source IP address from the host name
54 struct hostent *hsrc = gethostbyname(src);
55 if(hsrc) { // if the source was provided as a host namex
56 d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];
58 else { // assume it was specified as an IP address
59 if((ret=inet_aton(src, &d_ip_src)) == 0) { // format IP address
60 perror("Not a valid source IP address or host name");
61 throw std::runtime_error("can't initialize source socket");
65 d_port_src = htons(port_src); // format port number
67 d_sockaddr_src.sin_family = AF_INET;
68 d_sockaddr_src.sin_addr = d_ip_src;
69 d_sockaddr_src.sin_port = d_port_src;
71 d_temp_buff = new char[d_payload_size]; // allow it to hold up to payload_size bytes
77 gr_make_udp_source (size_t itemsize, const char *ipaddr,
78 unsigned short port, int payload_size)
80 return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr,
84 gr_udp_source::~gr_udp_source ()
86 delete [] d_temp_buff;
93 omni_mutex_lock l(d_mutex); // hold mutex for duration of this function
95 d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
97 perror("socket open");
98 throw std::runtime_error("can't open socket");
101 // Turn on reuse address
103 if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
104 perror("SO_REUSEADDR");
105 throw std::runtime_error("can't set socket option SO_REUSEADDR");
108 // Don't wait when shutting down
112 if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
114 throw std::runtime_error("can't set socket option SO_LINGER");
117 // Set a timeout on the receive function to not block indefinitely
118 // This value can (and probably should) be changed
122 if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (optval_t)&timeout, sizeof(timeout)) == -1) {
123 perror("SO_RCVTIMEO");
124 throw std::runtime_error("can't set socket option SO_RCVTIMEO");
127 // bind socket to an address and port number to listen on
128 if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
129 perror("socket bind");
130 throw std::runtime_error("can't bind socket");
134 return d_socket != 0;
138 gr_udp_source::close()
140 omni_mutex_lock l(d_mutex); // hold mutex for duration of this function
143 shutdown(d_socket, SHUT_RDWR);
150 gr_udp_source::work (int noutput_items,
151 gr_vector_const_void_star &input_items,
152 gr_vector_void_star &output_items)
154 char *out = (char *) output_items[0];
155 ssize_t r=0, nbytes=0, bytes_received=0;
156 ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
159 printf("\nEntered udp_source\n");
162 // Remove items from temp buffer if they are in there
164 nbytes = std::min(d_residual, total_bytes);
165 memcpy(out, d_temp_buff+d_temp_offset, nbytes);
166 bytes_received = nbytes;
169 printf("\tTemp buff size: %d offset: %d (bytes_received: %d) (noutput_items: %d)\n",
170 d_residual, d_temp_offset, bytes_received, noutput_items);
174 out += bytes_received;
176 // Update indexing of amount of bytes left in the buffer
177 d_residual -= nbytes;
178 d_temp_offset = d_temp_offset+d_residual;
182 // get the data into our output buffer and record the number of bytes
183 // This is a non-blocking call with a timeout set in the constructor
184 r = recv(d_socket, d_temp_buff, d_payload_size, 0); // get the entire payload or the what's available
186 // Check if there was a problem; forget it if the operation just timed out
188 if(errno == EAGAIN) { // handle non-blocking call timeout
190 printf("UDP receive timed out\n");
193 // Break here to allow the rest of the flow graph time to run and so ctrl-C breaks
197 perror("udp_source");
202 // Calculate the number of bytes we can take from the buffer in this call
203 nbytes = std::min(r, total_bytes-bytes_received);
205 // adjust the total number of bytes we have to round down to nearest integer of an itemsize
206 nbytes -= ((bytes_received+nbytes) % d_itemsize);
208 // copy the number of bytes we want to look at here
209 memcpy(out, d_temp_buff, nbytes);
211 d_residual = r - nbytes; // save the number of bytes stored
212 d_temp_offset=nbytes; // reset buffer index
214 // keep track of the total number of bytes received
215 bytes_received += nbytes;
217 // increment the pointer
220 // Immediately return when data comes in
225 printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
230 printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n",
231 bytes_received, bytes_received, noutput_items);
234 // bytes_received is already set to some integer multiple of itemsize
235 return bytes_received/d_itemsize;