7 #include <sys/socket.h>
15 #include "serverpool.h"
24 extern ServerPool *servers;
28 // Similar to parse_hostport(), but only parses the IP address,
29 // and does not use mapped-v4 addresses, since multicast seems
30 // to not like that too much.
31 bool parse_ip_address(const string &ip, sockaddr_storage *addr)
33 memset(addr, 0, sizeof(*addr));
37 sockaddr_in6 *addr6 = (sockaddr_in6 *)addr;
38 addr6->sin6_family = AF_INET6;
39 if (ip[ip.size() - 1] != ']') {
40 log(ERROR, "address '%s' is malformed; must be either [ipv6addr] or ipv4addr",
44 string raw_ip(ip.begin() + 1, ip.end() - 1);
45 if (inet_pton(AF_INET6, raw_ip.c_str(), &addr6->sin6_addr) != 1) {
46 log(ERROR, "'%s' is not a valid IPv6 address", raw_ip.c_str());
50 sockaddr_in *addr4 = (sockaddr_in *)addr;
51 addr4->sin_family = AF_INET;
52 if (inet_pton(AF_INET, ip.c_str(), &addr4->sin_addr) != 1) {
53 log(ERROR, "'%s' is not a valid IPv4 address");
61 bool maybe_join_multicast_group(int sock, const string &group, const string &source)
68 // Join the given multicast group (ASM or SSM).
69 // TODO: Also support sources apart from multicast groups,
70 // e.g. udp://[::1]:1234 for only receiving from localhost.
71 if (!source.empty()) {
72 // Single-Source Multicast (SSM).
74 memset(&gsr, 0, sizeof(gsr));
75 if (!parse_ip_address(group, &gsr.gsr_group)) {
78 if (!parse_ip_address(source, &gsr.gsr_source)) {
81 int level = (gsr.gsr_group.ss_family == AF_INET) ? SOL_IP : SOL_IPV6;
82 if (setsockopt(sock, level, MCAST_JOIN_SOURCE_GROUP, &gsr, sizeof(gsr)) == -1) {
83 log_perror("setsockopt(MCAST_JOIN_SOURCE_GROUP)");
87 // Any-Source Multicast (ASM).
89 memset(&gr, 0, sizeof(gr));
90 if (!parse_ip_address(group, &gr.gr_group)) {
93 int level = (gr.gr_group.ss_family == AF_INET) ? SOL_IP : SOL_IPV6;
94 if (setsockopt(sock, level, MCAST_JOIN_GROUP, &gr, sizeof(gr)) == -1) {
95 log_perror("setsockopt(MCAST_JOIN_GROUP)");
105 UDPInput::UDPInput(const string &url)
109 // Should be verified by the caller.
111 bool ok = parse_url(url, &protocol, &user, &host, &port, &path);
117 stats.connect_time = time(nullptr);
120 UDPInput::UDPInput(const InputProto &serialized)
121 : url(serialized.url()),
122 sock(serialized.sock())
124 // Set back the close-on-exec flag for the socket.
125 // (This can't leak into a child, since we haven't been started yet.)
126 fcntl(sock, F_SETFD, FD_CLOEXEC);
128 // Should be verified by the caller.
130 bool ok = parse_url(url, &protocol, &user, &host, &port, &path);
136 stats.bytes_received = serialized.bytes_received();
137 stats.data_bytes_received = serialized.data_bytes_received();
138 if (serialized.has_connect_time()) {
139 stats.connect_time = serialized.connect_time();
141 stats.connect_time = time(nullptr);
145 InputProto UDPInput::serialize() const
147 // Unset the close-on-exec flag for the socket.
148 // (This can't leak into a child, since there's only one thread left.)
149 fcntl(sock, F_SETFD, 0);
151 InputProto serialized;
152 serialized.set_url(url);
153 serialized.set_sock(sock);
154 serialized.set_bytes_received(stats.bytes_received);
155 serialized.set_data_bytes_received(stats.data_bytes_received);
156 serialized.set_connect_time(stats.connect_time);
157 serialized.set_is_metacube_encoded(false);
161 void UDPInput::close_socket()
167 void UDPInput::construct_header()
170 "HTTP/1.0 200 OK\r\n"
171 "Content-type: application/octet-stream\r\n"
172 "Cache-control: no-cache\r\n"
173 "Server: " SERVER_IDENTIFICATION "\r\n"
174 "Connection: close\r\n";
177 void UDPInput::add_destination(int stream_index)
179 stream_indices.push_back(stream_index);
180 servers->set_header(stream_index, http_header, "");
183 void UDPInput::do_work()
185 while (!should_stop()) {
187 int port_num = atoi(port.c_str());
188 sockaddr_in6 addr = create_any_address(port_num);
189 sock = create_server_socket(addr, UDP_SOCKET);
191 log(WARNING, "[%s] UDP socket creation failed. Waiting 0.2 seconds and trying again...",
197 // The syntax udp://source@group (abusing the username field
198 // to store the sender in SSM) seems to be a VLC invention.
200 if (!maybe_join_multicast_group(sock, host, user)) {
201 log(WARNING, "[%s] Multicast join failed. Waiting 0.2 seconds and trying again...",
210 // Wait for a packet, or a wakeup.
211 bool activity = wait_for_activity(sock, POLLIN, nullptr);
213 // Most likely, should_stop was set.
219 ret = recv(sock, packet_buf, sizeof(packet_buf), 0);
220 } while (ret == -1 && errno == EINTR);
222 if (ret < 0) { // Note that zero-byte packets are legal.
229 lock_guard<mutex> lock(stats_mutex);
230 stats.bytes_received += ret;
231 stats.data_bytes_received += ret;
234 for (size_t stream_index : stream_indices) {
235 servers->add_data(stream_index, packet_buf, ret, /*metacube_flags=*/0, /*pts=*/RationalPTS());
240 InputStats UDPInput::get_stats() const
242 lock_guard<mutex> lock(stats_mutex);