Shaka Packager SDK
udp_file.cc
1 // Copyright 2014 Google Inc. All rights reserved.
2 //
3 // Use of this source code is governed by a BSD-style
4 // license that can be found in the LICENSE file or at
5 // https://developers.google.com/open-source/licenses/bsd
6 
7 #include "packager/file/udp_file.h"
8 
9 #if defined(OS_WIN)
10 
11 #include <windows.h>
12 #include <ws2tcpip.h>
13 #define close closesocket
14 
15 #else
16 
17 #include <arpa/inet.h>
18 #include <errno.h>
19 #include <strings.h>
20 #include <sys/socket.h>
21 #include <unistd.h>
22 #define INVALID_SOCKET -1
23 
24 // IP_MULTICAST_ALL has been supported since kernel version 2.6.31 but we may be
25 // building on a machine that is older than that.
26 #ifndef IP_MULTICAST_ALL
27 #define IP_MULTICAST_ALL 49
28 #endif
29 
30 #endif // defined(OS_WIN)
31 
32 #include <limits>
33 
34 #include "packager/base/logging.h"
35 #include "packager/file/udp_options.h"
36 
37 namespace shaka {
38 
39 namespace {
40 
41 bool IsIpv4MulticastAddress(const struct in_addr& addr) {
42  return (ntohl(addr.s_addr) & 0xf0000000) == 0xe0000000;
43 }
44 
45 } // anonymous namespace
46 
47 UdpFile::UdpFile(const char* file_name)
48  : File(file_name), socket_(INVALID_SOCKET) {}
49 
50 UdpFile::~UdpFile() {}
51 
53  if (socket_ != INVALID_SOCKET) {
54  close(socket_);
55  socket_ = INVALID_SOCKET;
56  }
57  delete this;
58  return true;
59 }
60 
61 int64_t UdpFile::Read(void* buffer, uint64_t length) {
62  DCHECK(buffer);
63  DCHECK_GE(length, 65535u)
64  << "Buffer may be too small to read entire datagram.";
65 
66  if (socket_ == INVALID_SOCKET)
67  return -1;
68 
69  int64_t result;
70  do {
71  result =
72  recvfrom(socket_, reinterpret_cast<char*>(buffer), length, 0, NULL, 0);
73  } while ((result == -1) && (errno == EINTR));
74 
75  return result;
76 }
77 
78 int64_t UdpFile::Write(const void* buffer, uint64_t length) {
79  NOTIMPLEMENTED();
80  return -1;
81 }
82 
83 int64_t UdpFile::Size() {
84  if (socket_ == INVALID_SOCKET)
85  return -1;
86 
87  return std::numeric_limits<int64_t>::max();
88 }
89 
91  NOTIMPLEMENTED();
92  return false;
93 }
94 
95 bool UdpFile::Seek(uint64_t position) {
96  NOTIMPLEMENTED();
97  return false;
98 }
99 
100 bool UdpFile::Tell(uint64_t* position) {
101  NOTIMPLEMENTED();
102  return false;
103 }
104 
105 #if defined(OS_WIN)
106 class LibWinsockInitializer {
107  public:
108  LibWinsockInitializer() {
109  WSADATA wsa_data;
110  error_ = WSAStartup(MAKEWORD(2, 2), &wsa_data);
111  }
112 
113  ~LibWinsockInitializer() {
114  if (error_ == 0)
115  WSACleanup();
116  }
117 
118  int error() const { return error_; }
119 
120  private:
121  int error_;
122 };
123 #endif // defined(OS_WIN)
124 
125 class ScopedSocket {
126  public:
127  explicit ScopedSocket(SOCKET sock_fd) : sock_fd_(sock_fd) {}
128 
129  ~ScopedSocket() {
130  if (sock_fd_ != INVALID_SOCKET)
131  close(sock_fd_);
132  }
133 
134  SOCKET get() { return sock_fd_; }
135 
136  SOCKET release() {
137  SOCKET socket = sock_fd_;
138  sock_fd_ = INVALID_SOCKET;
139  return socket;
140  }
141 
142  private:
143  SOCKET sock_fd_;
144 
145  DISALLOW_COPY_AND_ASSIGN(ScopedSocket);
146 };
147 
149 #if defined(OS_WIN)
150  static LibWinsockInitializer lib_winsock_initializer;
151  if (lib_winsock_initializer.error() != 0) {
152  LOG(ERROR) << "Winsock start up failed with error "
153  << lib_winsock_initializer.error();
154  return false;
155  }
156 #endif // defined(OS_WIN)
157 
158  DCHECK_EQ(INVALID_SOCKET, socket_);
159 
160  std::unique_ptr<UdpOptions> options =
162  if (!options)
163  return false;
164 
165  ScopedSocket new_socket(socket(AF_INET, SOCK_DGRAM, 0));
166  if (new_socket.get() == INVALID_SOCKET) {
167  LOG(ERROR) << "Could not allocate socket.";
168  return false;
169  }
170 
171  struct in_addr local_in_addr = {0};
172  if (inet_pton(AF_INET, options->address().c_str(), &local_in_addr) != 1) {
173  LOG(ERROR) << "Malformed IPv4 address " << options->address();
174  return false;
175  }
176 
177  struct sockaddr_in local_sock_addr = {0};
178  // TODO(kqyang): Support IPv6.
179  local_sock_addr.sin_family = AF_INET;
180  local_sock_addr.sin_port = htons(options->port());
181  const bool is_multicast = IsIpv4MulticastAddress(local_in_addr);
182  if (is_multicast) {
183  local_sock_addr.sin_addr.s_addr = htonl(INADDR_ANY);
184  } else {
185  local_sock_addr.sin_addr = local_in_addr;
186  }
187 
188  if (options->reuse()) {
189  const int optval = 1;
190  if (setsockopt(new_socket.get(), SOL_SOCKET, SO_REUSEADDR,
191  reinterpret_cast<const char*>(&optval),
192  sizeof(optval)) < 0) {
193  LOG(ERROR)
194  << "Could not apply the SO_REUSEADDR property to the UDP socket";
195  return false;
196  }
197  }
198 
199  if (bind(new_socket.get(),
200  reinterpret_cast<struct sockaddr*>(&local_sock_addr),
201  sizeof(local_sock_addr))) {
202  LOG(ERROR) << "Could not bind UDP socket";
203  return false;
204  }
205 
206  if (is_multicast) {
207  if (options->is_source_specific_multicast()) {
208  struct ip_mreq_source source_multicast_group;
209 
210  source_multicast_group.imr_multiaddr = local_in_addr;
211  if (inet_pton(AF_INET,
212  options->interface_address().c_str(),
213  &source_multicast_group.imr_interface) != 1) {
214  LOG(ERROR) << "Malformed IPv4 interface address "
215  << options->interface_address();
216  return false;
217  }
218  if (inet_pton(AF_INET,
219  options->source_address().c_str(),
220  &source_multicast_group.imr_sourceaddr) != 1) {
221  LOG(ERROR) << "Malformed IPv4 source specific multicast address "
222  << options->source_address();
223  return false;
224  }
225 
226  if (setsockopt(new_socket.get(),
227  IPPROTO_IP,
228  IP_ADD_SOURCE_MEMBERSHIP,
229  reinterpret_cast<const char*>(&source_multicast_group),
230  sizeof(source_multicast_group)) < 0) {
231  LOG(ERROR) << "Failed to join multicast group.";
232  return false;
233  }
234  } else {
235  // this is a v2 join without a specific source.
236  struct ip_mreq multicast_group;
237 
238  multicast_group.imr_multiaddr = local_in_addr;
239 
240  if (inet_pton(AF_INET, options->interface_address().c_str(),
241  &multicast_group.imr_interface) != 1) {
242  LOG(ERROR) << "Malformed IPv4 interface address "
243  << options->interface_address();
244  return false;
245  }
246 
247  if (setsockopt(new_socket.get(), IPPROTO_IP, IP_ADD_MEMBERSHIP,
248  reinterpret_cast<const char*>(&multicast_group),
249  sizeof(multicast_group)) < 0) {
250  LOG(ERROR) << "Failed to join multicast group.";
251  return false;
252  }
253 
254  }
255 
256 #if defined(__linux__)
257  // Disable IP_MULTICAST_ALL to avoid interference caused when two sockets
258  // are bound to the same port but joined to different multicast groups.
259  const int optval_zero = 0;
260  if (setsockopt(new_socket.get(), IPPROTO_IP, IP_MULTICAST_ALL,
261  reinterpret_cast<const char*>(&optval_zero),
262  sizeof(optval_zero)) < 0 &&
263  errno != ENOPROTOOPT) {
264  LOG(ERROR) << "Failed to disable IP_MULTICAST_ALL option.";
265  return false;
266  }
267 #endif // #if defined(__linux__)
268  }
269 
270  // Set timeout if needed.
271  if (options->timeout_us() != 0) {
272  struct timeval tv;
273  tv.tv_sec = options->timeout_us() / 1000000;
274  tv.tv_usec = options->timeout_us() % 1000000;
275  if (setsockopt(new_socket.get(), SOL_SOCKET, SO_RCVTIMEO,
276  reinterpret_cast<const char*>(&tv), sizeof(tv)) < 0) {
277  LOG(ERROR) << "Failed to set socket timeout.";
278  return false;
279  }
280  }
281 
282  if (options->buffer_size() > 0) {
283  const int receive_buffer_size = options->buffer_size();
284  if (setsockopt(new_socket.get(), SOL_SOCKET, SO_RCVBUF,
285  reinterpret_cast<const char*>(&receive_buffer_size),
286  sizeof(receive_buffer_size)) < 0) {
287  LOG(ERROR) << "Failed to set the maximum receive buffer size: "
288  << strerror(errno);
289  return false;
290  }
291  }
292 
293  socket_ = new_socket.release();
294  return true;
295 }
296 
297 } // namespace shaka
int64_t Read(void *buffer, uint64_t length) override
Definition: udp_file.cc:61
bool Tell(uint64_t *position) override
Definition: udp_file.cc:100
bool Seek(uint64_t position) override
Definition: udp_file.cc:95
UdpFile(const char *address_and_port)
Definition: udp_file.cc:47
static std::unique_ptr< UdpOptions > ParseFromString(base::StringPiece udp_url)
Definition: udp_options.cc:75
Define an abstract file interface.
Definition: file.h:26
int64_t Write(const void *buffer, uint64_t length) override
Definition: udp_file.cc:78
const std::string & file_name() const
Definition: file.h:94
All the methods that are virtual are virtual for mocking.
bool Close() override
Definition: udp_file.cc:52
bool Open() override
Internal open. Should not be used directly.
Definition: udp_file.cc:148
int64_t Size() override
Definition: udp_file.cc:83
bool Flush() override
Definition: udp_file.cc:90