1 /*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include "webrtc/base/asynctcpsocket.h"
12
13 #include <string.h>
14
15 #include "webrtc/base/byteorder.h"
16 #include "webrtc/base/common.h"
17 #include "webrtc/base/logging.h"
18
19 #if defined(WEBRTC_POSIX)
20 #include <errno.h>
21 #endif // WEBRTC_POSIX
22
23 namespace rtc {
24
25 static const size_t kMaxPacketSize = 64 * 1024;
26
27 typedef uint16_t PacketLength;
28 static const size_t kPacketLenSize = sizeof(PacketLength);
29
30 static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;
31
32 static const int kListenBacklog = 5;
33
34 // Binds and connects |socket|
ConnectSocket(rtc::AsyncSocket * socket,const rtc::SocketAddress & bind_address,const rtc::SocketAddress & remote_address)35 AsyncSocket* AsyncTCPSocketBase::ConnectSocket(
36 rtc::AsyncSocket* socket,
37 const rtc::SocketAddress& bind_address,
38 const rtc::SocketAddress& remote_address) {
39 rtc::scoped_ptr<rtc::AsyncSocket> owned_socket(socket);
40 if (socket->Bind(bind_address) < 0) {
41 LOG(LS_ERROR) << "Bind() failed with error " << socket->GetError();
42 return NULL;
43 }
44 if (socket->Connect(remote_address) < 0) {
45 LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
46 return NULL;
47 }
48 return owned_socket.release();
49 }
50
AsyncTCPSocketBase(AsyncSocket * socket,bool listen,size_t max_packet_size)51 AsyncTCPSocketBase::AsyncTCPSocketBase(AsyncSocket* socket, bool listen,
52 size_t max_packet_size)
53 : socket_(socket),
54 listen_(listen),
55 insize_(max_packet_size),
56 inpos_(0),
57 outsize_(max_packet_size),
58 outpos_(0) {
59 inbuf_ = new char[insize_];
60 outbuf_ = new char[outsize_];
61
62 ASSERT(socket_.get() != NULL);
63 socket_->SignalConnectEvent.connect(
64 this, &AsyncTCPSocketBase::OnConnectEvent);
65 socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent);
66 socket_->SignalWriteEvent.connect(this, &AsyncTCPSocketBase::OnWriteEvent);
67 socket_->SignalCloseEvent.connect(this, &AsyncTCPSocketBase::OnCloseEvent);
68
69 if (listen_) {
70 if (socket_->Listen(kListenBacklog) < 0) {
71 LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
72 }
73 }
74 }
75
~AsyncTCPSocketBase()76 AsyncTCPSocketBase::~AsyncTCPSocketBase() {
77 delete [] inbuf_;
78 delete [] outbuf_;
79 }
80
GetLocalAddress() const81 SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
82 return socket_->GetLocalAddress();
83 }
84
GetRemoteAddress() const85 SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
86 return socket_->GetRemoteAddress();
87 }
88
Close()89 int AsyncTCPSocketBase::Close() {
90 return socket_->Close();
91 }
92
GetState() const93 AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
94 switch (socket_->GetState()) {
95 case Socket::CS_CLOSED:
96 return STATE_CLOSED;
97 case Socket::CS_CONNECTING:
98 if (listen_) {
99 return STATE_BOUND;
100 } else {
101 return STATE_CONNECTING;
102 }
103 case Socket::CS_CONNECTED:
104 return STATE_CONNECTED;
105 default:
106 ASSERT(false);
107 return STATE_CLOSED;
108 }
109 }
110
GetOption(Socket::Option opt,int * value)111 int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
112 return socket_->GetOption(opt, value);
113 }
114
SetOption(Socket::Option opt,int value)115 int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
116 return socket_->SetOption(opt, value);
117 }
118
GetError() const119 int AsyncTCPSocketBase::GetError() const {
120 return socket_->GetError();
121 }
122
SetError(int error)123 void AsyncTCPSocketBase::SetError(int error) {
124 return socket_->SetError(error);
125 }
126
SendTo(const void * pv,size_t cb,const SocketAddress & addr,const rtc::PacketOptions & options)127 int AsyncTCPSocketBase::SendTo(const void *pv, size_t cb,
128 const SocketAddress& addr,
129 const rtc::PacketOptions& options) {
130 const SocketAddress& remote_address = GetRemoteAddress();
131 if (addr == remote_address)
132 return Send(pv, cb, options);
133 // Remote address may be empty if there is a sudden network change.
134 ASSERT(remote_address.IsNil());
135 socket_->SetError(ENOTCONN);
136 return -1;
137 }
138
SendRaw(const void * pv,size_t cb)139 int AsyncTCPSocketBase::SendRaw(const void * pv, size_t cb) {
140 if (outpos_ + cb > outsize_) {
141 socket_->SetError(EMSGSIZE);
142 return -1;
143 }
144
145 memcpy(outbuf_ + outpos_, pv, cb);
146 outpos_ += cb;
147
148 return FlushOutBuffer();
149 }
150
FlushOutBuffer()151 int AsyncTCPSocketBase::FlushOutBuffer() {
152 int res = socket_->Send(outbuf_, outpos_);
153 if (res <= 0) {
154 return res;
155 }
156 if (static_cast<size_t>(res) <= outpos_) {
157 outpos_ -= res;
158 } else {
159 ASSERT(false);
160 return -1;
161 }
162 if (outpos_ > 0) {
163 memmove(outbuf_, outbuf_ + res, outpos_);
164 }
165 return res;
166 }
167
AppendToOutBuffer(const void * pv,size_t cb)168 void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
169 ASSERT(outpos_ + cb < outsize_);
170 memcpy(outbuf_ + outpos_, pv, cb);
171 outpos_ += cb;
172 }
173
OnConnectEvent(AsyncSocket * socket)174 void AsyncTCPSocketBase::OnConnectEvent(AsyncSocket* socket) {
175 SignalConnect(this);
176 }
177
OnReadEvent(AsyncSocket * socket)178 void AsyncTCPSocketBase::OnReadEvent(AsyncSocket* socket) {
179 ASSERT(socket_.get() == socket);
180
181 if (listen_) {
182 rtc::SocketAddress address;
183 rtc::AsyncSocket* new_socket = socket->Accept(&address);
184 if (!new_socket) {
185 // TODO: Do something better like forwarding the error
186 // to the user.
187 LOG(LS_ERROR) << "TCP accept failed with error " << socket_->GetError();
188 return;
189 }
190
191 HandleIncomingConnection(new_socket);
192
193 // Prime a read event in case data is waiting.
194 new_socket->SignalReadEvent(new_socket);
195 } else {
196 int len = socket_->Recv(inbuf_ + inpos_, insize_ - inpos_);
197 if (len < 0) {
198 // TODO: Do something better like forwarding the error to the user.
199 if (!socket_->IsBlocking()) {
200 LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
201 }
202 return;
203 }
204
205 inpos_ += len;
206
207 ProcessInput(inbuf_, &inpos_);
208
209 if (inpos_ >= insize_) {
210 LOG(LS_ERROR) << "input buffer overflow";
211 ASSERT(false);
212 inpos_ = 0;
213 }
214 }
215 }
216
OnWriteEvent(AsyncSocket * socket)217 void AsyncTCPSocketBase::OnWriteEvent(AsyncSocket* socket) {
218 ASSERT(socket_.get() == socket);
219
220 if (outpos_ > 0) {
221 FlushOutBuffer();
222 }
223
224 if (outpos_ == 0) {
225 SignalReadyToSend(this);
226 }
227 }
228
OnCloseEvent(AsyncSocket * socket,int error)229 void AsyncTCPSocketBase::OnCloseEvent(AsyncSocket* socket, int error) {
230 SignalClose(this, error);
231 }
232
233 // AsyncTCPSocket
234 // Binds and connects |socket| and creates AsyncTCPSocket for
235 // it. Takes ownership of |socket|. Returns NULL if bind() or
236 // connect() fail (|socket| is destroyed in that case).
Create(AsyncSocket * socket,const SocketAddress & bind_address,const SocketAddress & remote_address)237 AsyncTCPSocket* AsyncTCPSocket::Create(
238 AsyncSocket* socket,
239 const SocketAddress& bind_address,
240 const SocketAddress& remote_address) {
241 return new AsyncTCPSocket(AsyncTCPSocketBase::ConnectSocket(
242 socket, bind_address, remote_address), false);
243 }
244
AsyncTCPSocket(AsyncSocket * socket,bool listen)245 AsyncTCPSocket::AsyncTCPSocket(AsyncSocket* socket, bool listen)
246 : AsyncTCPSocketBase(socket, listen, kBufSize) {
247 }
248
Send(const void * pv,size_t cb,const rtc::PacketOptions & options)249 int AsyncTCPSocket::Send(const void *pv, size_t cb,
250 const rtc::PacketOptions& options) {
251 if (cb > kBufSize) {
252 SetError(EMSGSIZE);
253 return -1;
254 }
255
256 // If we are blocking on send, then silently drop this packet
257 if (!IsOutBufferEmpty())
258 return static_cast<int>(cb);
259
260 PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
261 AppendToOutBuffer(&pkt_len, kPacketLenSize);
262 AppendToOutBuffer(pv, cb);
263
264 int res = FlushOutBuffer();
265 if (res <= 0) {
266 // drop packet if we made no progress
267 ClearOutBuffer();
268 return res;
269 }
270
271 rtc::SentPacket sent_packet(options.packet_id, rtc::Time());
272 SignalSentPacket(this, sent_packet);
273
274 // We claim to have sent the whole thing, even if we only sent partial
275 return static_cast<int>(cb);
276 }
277
ProcessInput(char * data,size_t * len)278 void AsyncTCPSocket::ProcessInput(char * data, size_t* len) {
279 SocketAddress remote_addr(GetRemoteAddress());
280
281 while (true) {
282 if (*len < kPacketLenSize)
283 return;
284
285 PacketLength pkt_len = rtc::GetBE16(data);
286 if (*len < kPacketLenSize + pkt_len)
287 return;
288
289 SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
290 CreatePacketTime(0));
291
292 *len -= kPacketLenSize + pkt_len;
293 if (*len > 0) {
294 memmove(data, data + kPacketLenSize + pkt_len, *len);
295 }
296 }
297 }
298
HandleIncomingConnection(AsyncSocket * socket)299 void AsyncTCPSocket::HandleIncomingConnection(AsyncSocket* socket) {
300 SignalNewConnection(this, new AsyncTCPSocket(socket, false));
301 }
302
303 } // namespace rtc
304