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>
34 gr_udp_source::gr_udp_source(size_t itemsize, const char *src,
35 unsigned short port_src, int payload_size)
36 : gr_sync_block ("udp_source",
37 gr_make_io_signature(0, 0, 0),
38 gr_make_io_signature(1, 1, itemsize)),
39 d_itemsize(itemsize), d_updated(false), d_payload_size(payload_size), d_residual(0), d_temp_offset(0)
43 // Set up the address stucture for the source address and port numbers
44 // Get the source IP address from the host name
45 struct hostent *hsrc = gethostbyname(src);
46 if(hsrc) { // if the source was provided as a host namex
47 d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];
49 else { // assume it was specified as an IP address
50 if((ret=inet_aton(src, &d_ip_src)) == 0) { // format IP address
51 perror("Not a valid source IP address or host name");
52 throw std::runtime_error("can't initialize source socket");
56 d_port_src = htons(port_src); // format port number
58 d_sockaddr_src.sin_family = AF_INET;
59 d_sockaddr_src.sin_addr = d_ip_src;
60 d_sockaddr_src.sin_port = d_port_src;
62 d_temp_buff = new char[d_payload_size]; // allow it to hold up to payload_size bytes
68 gr_make_udp_source (size_t itemsize, const char *ipaddr,
69 unsigned short port, int payload_size)
71 return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr,
75 gr_udp_source::~gr_udp_source ()
77 delete [] d_temp_buff;
84 omni_mutex_lock l(d_mutex); // hold mutex for duration of this function
86 d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
88 perror("socket open");
89 throw std::runtime_error("can't open socket");
92 // Turn on reuse address
94 if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (void*)&opt_val, sizeof(int)) == -1) {
95 perror("SO_REUSEADDR");
96 throw std::runtime_error("can't set socket option SO_REUSEADDR");
99 // Don't wait when shutting down
103 if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (void*)&lngr, sizeof(linger)) == -1) {
105 throw std::runtime_error("can't set socket option SO_LINGER");
108 // Set a timeout on the receive function to not block indefinitely
109 // This value can (and probably should) be changed
113 if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (void*)&timeout, sizeof(timeout)) == -1) {
114 perror("SO_RCVTIMEO");
115 throw std::runtime_error("can't set socket option SO_RCVTIMEO");
118 // bind socket to an address and port number to listen on
119 if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
120 perror("socket bind");
121 throw std::runtime_error("can't bind socket");
125 return d_socket != 0;
129 gr_udp_source::close()
131 omni_mutex_lock l(d_mutex); // hold mutex for duration of this function
134 shutdown(d_socket, SHUT_RDWR);
141 gr_udp_source::work (int noutput_items,
142 gr_vector_const_void_star &input_items,
143 gr_vector_void_star &output_items)
145 char *out = (char *) output_items[0];
146 ssize_t r=0, nbytes=0, bytes_received=0;
147 ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
150 printf("\nEntered udp_source\n");
153 // Remove items from temp buffer if they are in there
155 nbytes = std::min(d_residual, total_bytes);
156 memcpy(out, d_temp_buff+d_temp_offset, nbytes);
157 bytes_received = nbytes;
160 printf("\tTemp buff size: %d offset: %d (bytes_received: %d) (noutput_items: %d)\n",
161 d_residual, d_temp_offset, bytes_received, noutput_items);
165 out += bytes_received;
167 // Update indexing of amount of bytes left in the buffer
168 d_residual -= nbytes;
169 d_temp_offset = d_temp_offset+d_residual;
173 // get the data into our output buffer and record the number of bytes
174 // This is a non-blocking call with a timeout set in the constructor
175 r = recv(d_socket, d_temp_buff, d_payload_size, 0); // get the entire payload or the what's available
177 // Check if there was a problem; forget it if the operation just timed out
179 if(errno == EAGAIN) { // handle non-blocking call timeout
181 printf("UDP receive timed out\n");
184 // Break here to allow the rest of the flow graph time to run and so ctrl-C breaks
188 perror("udp_source");
193 // Calculate the number of bytes we can take from the buffer in this call
194 nbytes = std::min(r, total_bytes-bytes_received);
196 // adjust the total number of bytes we have to round down to nearest integer of an itemsize
197 nbytes -= ((bytes_received+nbytes) % d_itemsize);
199 // copy the number of bytes we want to look at here
200 memcpy(out, d_temp_buff, nbytes);
202 d_residual = r - nbytes; // save the number of bytes stored
203 d_temp_offset=nbytes; // reset buffer index
205 // keep track of the total number of bytes received
206 bytes_received += nbytes;
208 // increment the pointer
211 // Immediately return when data comes in
216 printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
221 printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n",
222 bytes_received, bytes_received, noutput_items);
225 // bytes_received is already set to some integer multiple of itemsize
226 return bytes_received/d_itemsize;