1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "remoting/jingle_glue/chromium_socket_factory.h"
6
7 #include "base/bind.h"
8 #include "base/logging.h"
9 #include "base/memory/scoped_ptr.h"
10 #include "jingle/glue/utils.h"
11 #include "net/base/io_buffer.h"
12 #include "net/base/ip_endpoint.h"
13 #include "net/base/net_errors.h"
14 #include "net/udp/udp_server_socket.h"
15 #include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
16 #include "third_party/libjingle/source/talk/base/nethelpers.h"
17
18 namespace remoting {
19
20 namespace {
21
22 // Size of the buffer to allocate for RecvFrom().
23 const int kReceiveBufferSize = 65536;
24
25 // Maximum amount of data in the send buffers. This is necessary to
26 // prevent out-of-memory crashes if the caller sends data faster than
27 // Pepper's UDP API can handle it. This maximum should never be
28 // reached under normal conditions.
29 const int kMaxSendBufferSize = 256 * 1024;
30
31 // Defines set of transient errors. These errors are ignored when we get them
32 // from sendto() calls.
IsTransientError(int error)33 bool IsTransientError(int error) {
34 return error == net::ERR_ADDRESS_UNREACHABLE ||
35 error == net::ERR_ADDRESS_INVALID;
36 }
37
38 class UdpPacketSocket : public talk_base::AsyncPacketSocket {
39 public:
40 UdpPacketSocket();
41 virtual ~UdpPacketSocket();
42
43 bool Init(const talk_base::SocketAddress& local_address,
44 int min_port, int max_port);
45
46 // talk_base::AsyncPacketSocket interface.
47 virtual talk_base::SocketAddress GetLocalAddress() const OVERRIDE;
48 virtual talk_base::SocketAddress GetRemoteAddress() const OVERRIDE;
49 virtual int Send(const void* data, size_t data_size,
50 talk_base::DiffServCodePoint dscp) OVERRIDE;
51 virtual int SendTo(const void* data, size_t data_size,
52 const talk_base::SocketAddress& address,
53 talk_base::DiffServCodePoint dscp) OVERRIDE;
54 virtual int Close() OVERRIDE;
55 virtual State GetState() const OVERRIDE;
56 virtual int GetOption(talk_base::Socket::Option option, int* value) OVERRIDE;
57 virtual int SetOption(talk_base::Socket::Option option, int value) OVERRIDE;
58 virtual int GetError() const OVERRIDE;
59 virtual void SetError(int error) OVERRIDE;
60
61 private:
62 struct PendingPacket {
63 PendingPacket(const void* buffer,
64 int buffer_size,
65 const net::IPEndPoint& address);
66
67 scoped_refptr<net::IOBufferWithSize> data;
68 net::IPEndPoint address;
69 };
70
71 void OnBindCompleted(int error);
72
73 void DoSend();
74 void OnSendCompleted(int result);
75
76 void DoRead();
77 void OnReadCompleted(int result);
78 void HandleReadResult(int result);
79
80 scoped_ptr<net::UDPServerSocket> socket_;
81
82 State state_;
83 int error_;
84
85 talk_base::SocketAddress local_address_;
86
87 // Receive buffer and address are populated by asynchronous reads.
88 scoped_refptr<net::IOBuffer> receive_buffer_;
89 net::IPEndPoint receive_address_;
90
91 bool send_pending_;
92 std::list<PendingPacket> send_queue_;
93 int send_queue_size_;
94
95 DISALLOW_COPY_AND_ASSIGN(UdpPacketSocket);
96 };
97
PendingPacket(const void * buffer,int buffer_size,const net::IPEndPoint & address)98 UdpPacketSocket::PendingPacket::PendingPacket(
99 const void* buffer,
100 int buffer_size,
101 const net::IPEndPoint& address)
102 : data(new net::IOBufferWithSize(buffer_size)),
103 address(address) {
104 memcpy(data->data(), buffer, buffer_size);
105 }
106
UdpPacketSocket()107 UdpPacketSocket::UdpPacketSocket()
108 : state_(STATE_CLOSED),
109 error_(0),
110 send_pending_(false),
111 send_queue_size_(0) {
112 }
113
~UdpPacketSocket()114 UdpPacketSocket::~UdpPacketSocket() {
115 Close();
116 }
117
Init(const talk_base::SocketAddress & local_address,int min_port,int max_port)118 bool UdpPacketSocket::Init(const talk_base::SocketAddress& local_address,
119 int min_port, int max_port) {
120 net::IPEndPoint local_endpoint;
121 if (!jingle_glue::SocketAddressToIPEndPoint(
122 local_address, &local_endpoint)) {
123 return false;
124 }
125
126 for (int port = min_port; port <= max_port; ++port) {
127 socket_.reset(new net::UDPServerSocket(NULL, net::NetLog::Source()));
128 int result = socket_->Listen(
129 net::IPEndPoint(local_endpoint.address(), port));
130 if (result == net::OK) {
131 break;
132 } else {
133 socket_.reset();
134 }
135 }
136
137 if (!socket_.get()) {
138 // Failed to bind the socket.
139 return false;
140 }
141
142 if (socket_->GetLocalAddress(&local_endpoint) != net::OK ||
143 !jingle_glue::IPEndPointToSocketAddress(local_endpoint,
144 &local_address_)) {
145 return false;
146 }
147
148 state_ = STATE_BOUND;
149 DoRead();
150
151 return true;
152 }
153
GetLocalAddress() const154 talk_base::SocketAddress UdpPacketSocket::GetLocalAddress() const {
155 DCHECK_EQ(state_, STATE_BOUND);
156 return local_address_;
157 }
158
GetRemoteAddress() const159 talk_base::SocketAddress UdpPacketSocket::GetRemoteAddress() const {
160 // UDP sockets are not connected - this method should never be called.
161 NOTREACHED();
162 return talk_base::SocketAddress();
163 }
164
Send(const void * data,size_t data_size,talk_base::DiffServCodePoint dscp)165 int UdpPacketSocket::Send(const void* data, size_t data_size,
166 talk_base::DiffServCodePoint dscp) {
167 // UDP sockets are not connected - this method should never be called.
168 NOTREACHED();
169 return EWOULDBLOCK;
170 }
171
SendTo(const void * data,size_t data_size,const talk_base::SocketAddress & address,talk_base::DiffServCodePoint dscp)172 int UdpPacketSocket::SendTo(const void* data, size_t data_size,
173 const talk_base::SocketAddress& address,
174 talk_base::DiffServCodePoint dscp) {
175 if (state_ != STATE_BOUND) {
176 NOTREACHED();
177 return EINVAL;
178 }
179
180 if (error_ != 0) {
181 return error_;
182 }
183
184 net::IPEndPoint endpoint;
185 if (!jingle_glue::SocketAddressToIPEndPoint(address, &endpoint)) {
186 return EINVAL;
187 }
188
189 if (send_queue_size_ >= kMaxSendBufferSize) {
190 return EWOULDBLOCK;
191 }
192
193 send_queue_.push_back(PendingPacket(data, data_size, endpoint));
194 send_queue_size_ += data_size;
195
196 DoSend();
197 return data_size;
198 }
199
Close()200 int UdpPacketSocket::Close() {
201 state_ = STATE_CLOSED;
202 socket_.reset();
203 return 0;
204 }
205
GetState() const206 talk_base::AsyncPacketSocket::State UdpPacketSocket::GetState() const {
207 return state_;
208 }
209
GetOption(talk_base::Socket::Option option,int * value)210 int UdpPacketSocket::GetOption(talk_base::Socket::Option option, int* value) {
211 // This method is never called by libjingle.
212 NOTIMPLEMENTED();
213 return -1;
214 }
215
SetOption(talk_base::Socket::Option option,int value)216 int UdpPacketSocket::SetOption(talk_base::Socket::Option option, int value) {
217 if (state_ != STATE_BOUND) {
218 NOTREACHED();
219 return EINVAL;
220 }
221
222 switch (option) {
223 case talk_base::Socket::OPT_DONTFRAGMENT:
224 NOTIMPLEMENTED();
225 return -1;
226
227 case talk_base::Socket::OPT_RCVBUF: {
228 bool success = socket_->SetReceiveBufferSize(value);
229 return success ? 0 : -1;
230 }
231
232 case talk_base::Socket::OPT_SNDBUF: {
233 bool success = socket_->SetSendBufferSize(value);
234 return success ? 0 : -1;
235 }
236
237 case talk_base::Socket::OPT_NODELAY:
238 // OPT_NODELAY is only for TCP sockets.
239 NOTREACHED();
240 return -1;
241
242 case talk_base::Socket::OPT_IPV6_V6ONLY:
243 NOTIMPLEMENTED();
244 return -1;
245
246 case talk_base::Socket::OPT_DSCP:
247 NOTIMPLEMENTED();
248 return -1;
249 }
250
251 NOTREACHED();
252 return -1;
253 }
254
GetError() const255 int UdpPacketSocket::GetError() const {
256 return error_;
257 }
258
SetError(int error)259 void UdpPacketSocket::SetError(int error) {
260 error_ = error;
261 }
262
DoSend()263 void UdpPacketSocket::DoSend() {
264 if (send_pending_ || send_queue_.empty())
265 return;
266
267 PendingPacket& packet = send_queue_.front();
268 int result = socket_->SendTo(
269 packet.data.get(),
270 packet.data->size(),
271 packet.address,
272 base::Bind(&UdpPacketSocket::OnSendCompleted, base::Unretained(this)));
273 if (result == net::ERR_IO_PENDING) {
274 send_pending_ = true;
275 } else {
276 OnSendCompleted(result);
277 }
278 }
279
OnSendCompleted(int result)280 void UdpPacketSocket::OnSendCompleted(int result) {
281 send_pending_ = false;
282
283 if (result < 0) {
284 if (!IsTransientError(result)) {
285 LOG(ERROR) << "Send failed on a UDP socket: " << result;
286 error_ = EINVAL;
287 return;
288 }
289 }
290
291 // Don't need to worry about partial sends because this is a datagram
292 // socket.
293 send_queue_size_ -= send_queue_.front().data->size();
294 send_queue_.pop_front();
295 DoSend();
296 }
297
DoRead()298 void UdpPacketSocket::DoRead() {
299 int result = 0;
300 while (result >= 0) {
301 receive_buffer_ = new net::IOBuffer(kReceiveBufferSize);
302 result = socket_->RecvFrom(
303 receive_buffer_.get(),
304 kReceiveBufferSize,
305 &receive_address_,
306 base::Bind(&UdpPacketSocket::OnReadCompleted, base::Unretained(this)));
307 HandleReadResult(result);
308 }
309 }
310
OnReadCompleted(int result)311 void UdpPacketSocket::OnReadCompleted(int result) {
312 HandleReadResult(result);
313 if (result >= 0) {
314 DoRead();
315 }
316 }
317
HandleReadResult(int result)318 void UdpPacketSocket::HandleReadResult(int result) {
319 if (result == net::ERR_IO_PENDING) {
320 return;
321 }
322
323 if (result > 0) {
324 talk_base::SocketAddress address;
325 if (!jingle_glue::IPEndPointToSocketAddress(receive_address_, &address)) {
326 NOTREACHED();
327 LOG(ERROR) << "Failed to convert address received from RecvFrom().";
328 return;
329 }
330 SignalReadPacket(this, receive_buffer_->data(), result, address,
331 talk_base::CreatePacketTime(0));
332 } else {
333 LOG(ERROR) << "Received error when reading from UDP socket: " << result;
334 }
335 }
336
337 } // namespace
338
ChromiumPacketSocketFactory()339 ChromiumPacketSocketFactory::ChromiumPacketSocketFactory() {
340 }
341
~ChromiumPacketSocketFactory()342 ChromiumPacketSocketFactory::~ChromiumPacketSocketFactory() {
343 }
344
CreateUdpSocket(const talk_base::SocketAddress & local_address,int min_port,int max_port)345 talk_base::AsyncPacketSocket* ChromiumPacketSocketFactory::CreateUdpSocket(
346 const talk_base::SocketAddress& local_address,
347 int min_port, int max_port) {
348 scoped_ptr<UdpPacketSocket> result(new UdpPacketSocket());
349 if (!result->Init(local_address, min_port, max_port))
350 return NULL;
351 return result.release();
352 }
353
354 talk_base::AsyncPacketSocket*
CreateServerTcpSocket(const talk_base::SocketAddress & local_address,int min_port,int max_port,int opts)355 ChromiumPacketSocketFactory::CreateServerTcpSocket(
356 const talk_base::SocketAddress& local_address,
357 int min_port, int max_port,
358 int opts) {
359 // We don't use TCP sockets for remoting connections.
360 NOTREACHED();
361 return NULL;
362 }
363
364 talk_base::AsyncPacketSocket*
CreateClientTcpSocket(const talk_base::SocketAddress & local_address,const talk_base::SocketAddress & remote_address,const talk_base::ProxyInfo & proxy_info,const std::string & user_agent,int opts)365 ChromiumPacketSocketFactory::CreateClientTcpSocket(
366 const talk_base::SocketAddress& local_address,
367 const talk_base::SocketAddress& remote_address,
368 const talk_base::ProxyInfo& proxy_info,
369 const std::string& user_agent,
370 int opts) {
371 // We don't use TCP sockets for remoting connections.
372 NOTREACHED();
373 return NULL;
374 }
375
376 talk_base::AsyncResolverInterface*
CreateAsyncResolver()377 ChromiumPacketSocketFactory::CreateAsyncResolver() {
378 return new talk_base::AsyncResolver();
379 }
380
381 } // namespace remoting
382