6 #include <sys/socket.h>
13 #include "mutexlock.h"
14 #include "serverpool.h"
23 extern ServerPool *servers;
25 UDPInput::UDPInput(const string &url)
29 // Should be verified by the caller.
31 bool ok = parse_url(url, &protocol, &host, &port, &path);
36 pthread_mutex_init(&stats_mutex, NULL);
38 stats.bytes_received = 0;
39 stats.data_bytes_received = 0;
40 stats.connect_time = time(NULL);
43 UDPInput::UDPInput(const InputProto &serialized)
44 : url(serialized.url()),
45 sock(serialized.sock())
47 // Should be verified by the caller.
49 bool ok = parse_url(url, &protocol, &host, &port, &path);
54 pthread_mutex_init(&stats_mutex, NULL);
56 stats.bytes_received = serialized.bytes_received();
57 stats.data_bytes_received = serialized.data_bytes_received();
58 if (serialized.has_connect_time()) {
59 stats.connect_time = serialized.connect_time();
61 stats.connect_time = time(NULL);
65 InputProto UDPInput::serialize() const
67 InputProto serialized;
68 serialized.set_url(url);
69 serialized.set_sock(sock);
70 serialized.set_bytes_received(stats.bytes_received);
71 serialized.set_data_bytes_received(stats.data_bytes_received);
72 serialized.set_connect_time(stats.connect_time);
76 void UDPInput::close_socket()
82 void UDPInput::construct_header()
86 "Content-type: application/octet-stream\r\n"
87 "Cache-control: no-cache\r\n"
88 "Server: " SERVER_IDENTIFICATION "\r\n"
89 "Connection: close\r\n";
92 void UDPInput::add_destination(int stream_index)
94 stream_indices.push_back(stream_index);
95 servers->set_header(stream_index, http_header, "");
98 void UDPInput::do_work()
100 while (!should_stop()) {
102 int port_num = atoi(port.c_str());
103 sock = create_server_socket(port_num, UDP_SOCKET);
105 log(WARNING, "[%s] UDP socket creation failed. Waiting 0.2 seconds and trying again...",
112 // Wait for a packet, or a wakeup.
113 bool activity = wait_for_activity(sock, POLLIN, NULL);
115 // Most likely, should_stop was set.
121 ret = recv(sock, packet_buf, sizeof(packet_buf), 0);
122 } while (ret == -1 && errno == EINTR);
131 MutexLock lock(&stats_mutex);
132 stats.bytes_received += ret;
133 stats.data_bytes_received += ret;
136 for (size_t i = 0; i < stream_indices.size(); ++i) {
137 servers->add_data(stream_indices[i], packet_buf, ret, SUITABLE_FOR_STREAM_START);
142 InputStats UDPInput::get_stats() const
144 MutexLock lock(&stats_mutex);