7 #include <sys/socket.h>
13 #include "serverpool.h"
21 extern ServerPool *servers;
23 UDPInput::UDPInput(const string &url)
27 // Should be verified by the caller.
29 bool ok = parse_url(url, &protocol, &host, &port, &path);
35 UDPInput::UDPInput(const InputProto &serialized)
36 : url(serialized.url()),
37 sock(serialized.sock())
39 // Should be verified by the caller.
41 bool ok = parse_url(url, &protocol, &host, &port, &path);
47 InputProto UDPInput::serialize() const
49 InputProto serialized;
50 serialized.set_url(url);
51 serialized.set_sock(sock);
55 void UDPInput::close_socket()
61 void UDPInput::construct_header()
65 "Content-type: application/octet-stream\r\n"
66 "Cache-control: no-cache\r\n"
67 "Server: " SERVER_IDENTIFICATION "\r\n"
68 "Connection: close\r\n";
71 void UDPInput::add_destination(int stream_index)
73 stream_indices.push_back(stream_index);
74 servers->set_header(stream_index, http_header, "");
77 void UDPInput::do_work()
79 while (!should_stop()) {
81 int port_num = atoi(port.c_str());
82 sock = create_server_socket(port_num, UDP_SOCKET);
84 log(WARNING, "[%s] UDP socket creation failed. Waiting 0.2 seconds and trying again...",
91 // Wait for a packet, or a wakeup.
92 bool activity = wait_for_activity(sock, POLLIN, NULL);
94 // Most likely, should_stop was set.
101 ret = recv(sock, buf, sizeof(buf), 0);
102 } while (ret == -1 && errno == EINTR);
110 for (size_t i = 0; i < stream_indices.size(); ++i) {
111 servers->add_data(stream_indices[i], buf, ret);