Run include-what-you-use.
[cubemap] / udpinput.cpp
1 #include <assert.h>
2 #include <errno.h>
3 #include <poll.h>
4 #include <stddef.h>
5 #include <stdlib.h>
6 #include <sys/socket.h>
7 #include <unistd.h>
8 #include <string>
9
10 #include "acceptor.h"
11 #include "log.h"
12 #include "serverpool.h"
13 #include "state.pb.h"
14 #include "udpinput.h"
15 #include "util.h"
16 #include "version.h"
17
18 using namespace std;
19
20 extern ServerPool *servers;
21
22 UDPInput::UDPInput(const string &url)
23         : url(url),
24           sock(-1)
25 {
26         // Should be verified by the caller.
27         string protocol;
28         bool ok = parse_url(url, &protocol, &host, &port, &path);
29         assert(ok);
30
31         construct_header();
32 }
33
34 UDPInput::UDPInput(const InputProto &serialized)
35         : url(serialized.url()),
36           sock(serialized.sock())
37 {
38         // Should be verified by the caller.
39         string protocol;
40         bool ok = parse_url(url, &protocol, &host, &port, &path);
41         assert(ok);
42
43         construct_header();
44 }
45
46 InputProto UDPInput::serialize() const
47 {
48         InputProto serialized;
49         serialized.set_url(url);
50         serialized.set_sock(sock);
51         return serialized;
52 }
53
54 void UDPInput::close_socket()
55 {
56         safe_close(sock);
57         sock = -1;
58 }
59         
60 void UDPInput::construct_header()
61 {
62         http_header =
63                 "HTTP/1.0 200 OK\r\n"
64                 "Content-type: application/octet-stream\r\n"
65                 "Cache-control: no-cache\r\n"
66                 "Server: " SERVER_IDENTIFICATION "\r\n"
67                 "Connection: close\r\n";
68 }
69         
70 void UDPInput::add_destination(int stream_index)
71 {
72         stream_indices.push_back(stream_index);
73         servers->set_header(stream_index, http_header, "");
74 }
75
76 void UDPInput::do_work()
77 {
78         while (!should_stop()) {
79                 if (sock == -1) {
80                         int port_num = atoi(port.c_str());
81                         sock = create_server_socket(port_num, UDP_SOCKET);
82                         if (sock == -1) {
83                                 log(WARNING, "[%s] UDP socket creation failed. Waiting 0.2 seconds and trying again...",
84                                              url.c_str());
85                                 usleep(200000);
86                                 continue;
87                         }
88                 }
89
90                 // Wait for a packet, or a wakeup.
91                 bool activity = wait_for_activity(sock, POLLIN, NULL);
92                 if (!activity) {
93                         // Most likely, should_stop was set.
94                         continue;
95                 }
96
97                 char buf[4096];
98                 int ret;
99                 do {
100                         ret = recv(sock, buf, sizeof(buf), 0);
101                 } while (ret == -1 && errno == EINTR);
102
103                 if (ret <= 0) {
104                         log_perror("recv");
105                         close_socket();
106                         continue;
107                 }
108                 
109                 for (size_t i = 0; i < stream_indices.size(); ++i) {
110                         servers->add_data(stream_indices[i], buf, ret);
111                 }
112         }
113 }