3 * Copyright 2004--2010, Google Inc.
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
8 * 1. Redistributions of source code must retain the above copyright notice,
9 * this list of conditions and the following disclaimer.
10 * 2. Redistributions in binary form must reproduce the above copyright notice,
11 * this list of conditions and the following disclaimer in the documentation
12 * and/or other materials provided with the distribution.
13 * 3. The name of the author may not be used to endorse or promote products
14 * derived from this software without specific prior written permission.
16 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
17 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
18 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
19 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
20 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
22 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
23 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
24 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
25 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
28 #include "talk/base/asynctcpsocket.h"
32 #include "talk/base/byteorder.h"
33 #include "talk/base/common.h"
34 #include "talk/base/logging.h"
42 static const size_t kMaxPacketSize = 64 * 1024;
44 typedef uint16 PacketLength;
45 static const size_t kPacketLenSize = sizeof(PacketLength);
47 static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;
49 static const int kListenBacklog = 5;
51 // Binds and connects |socket|
52 AsyncSocket* AsyncTCPSocketBase::ConnectSocket(
53 talk_base::AsyncSocket* socket,
54 const talk_base::SocketAddress& bind_address,
55 const talk_base::SocketAddress& remote_address) {
56 talk_base::scoped_ptr<talk_base::AsyncSocket> owned_socket(socket);
57 if (socket->Bind(bind_address) < 0) {
58 LOG(LS_ERROR) << "Bind() failed with error " << socket->GetError();
61 if (socket->Connect(remote_address) < 0) {
62 LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
65 return owned_socket.release();
68 AsyncTCPSocketBase::AsyncTCPSocketBase(AsyncSocket* socket, bool listen,
69 size_t max_packet_size)
72 insize_(max_packet_size),
74 outsize_(max_packet_size),
76 inbuf_ = new char[insize_];
77 outbuf_ = new char[outsize_];
79 ASSERT(socket_.get() != NULL);
80 socket_->SignalConnectEvent.connect(
81 this, &AsyncTCPSocketBase::OnConnectEvent);
82 socket_->SignalReadEvent.connect(this, &AsyncTCPSocketBase::OnReadEvent);
83 socket_->SignalWriteEvent.connect(this, &AsyncTCPSocketBase::OnWriteEvent);
84 socket_->SignalCloseEvent.connect(this, &AsyncTCPSocketBase::OnCloseEvent);
87 if (socket_->Listen(kListenBacklog) < 0) {
88 LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
93 AsyncTCPSocketBase::~AsyncTCPSocketBase() {
98 SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
99 return socket_->GetLocalAddress();
102 SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
103 return socket_->GetRemoteAddress();
106 int AsyncTCPSocketBase::Close() {
107 return socket_->Close();
110 AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
111 switch (socket_->GetState()) {
112 case Socket::CS_CLOSED:
114 case Socket::CS_CONNECTING:
118 return STATE_CONNECTING;
120 case Socket::CS_CONNECTED:
121 return STATE_CONNECTED;
128 int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
129 return socket_->GetOption(opt, value);
132 int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
133 return socket_->SetOption(opt, value);
136 int AsyncTCPSocketBase::GetError() const {
137 return socket_->GetError();
140 void AsyncTCPSocketBase::SetError(int error) {
141 return socket_->SetError(error);
144 int AsyncTCPSocketBase::SendTo(const void *pv, size_t cb,
145 const SocketAddress& addr,
146 const talk_base::PacketOptions& options) {
147 if (addr == GetRemoteAddress())
148 return Send(pv, cb, options);
151 socket_->SetError(ENOTCONN);
155 int AsyncTCPSocketBase::SendRaw(const void * pv, size_t cb) {
156 if (outpos_ + cb > outsize_) {
157 socket_->SetError(EMSGSIZE);
161 memcpy(outbuf_ + outpos_, pv, cb);
164 return FlushOutBuffer();
167 int AsyncTCPSocketBase::FlushOutBuffer() {
168 int res = socket_->Send(outbuf_, outpos_);
172 if (static_cast<size_t>(res) <= outpos_) {
179 memmove(outbuf_, outbuf_ + res, outpos_);
184 void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
185 ASSERT(outpos_ + cb < outsize_);
186 memcpy(outbuf_ + outpos_, pv, cb);
190 void AsyncTCPSocketBase::OnConnectEvent(AsyncSocket* socket) {
194 void AsyncTCPSocketBase::OnReadEvent(AsyncSocket* socket) {
195 ASSERT(socket_.get() == socket);
198 talk_base::SocketAddress address;
199 talk_base::AsyncSocket* new_socket = socket->Accept(&address);
201 // TODO: Do something better like forwarding the error
203 LOG(LS_ERROR) << "TCP accept failed with error " << socket_->GetError();
207 HandleIncomingConnection(new_socket);
209 // Prime a read event in case data is waiting.
210 new_socket->SignalReadEvent(new_socket);
212 int len = socket_->Recv(inbuf_ + inpos_, insize_ - inpos_);
214 // TODO: Do something better like forwarding the error to the user.
215 if (!socket_->IsBlocking()) {
216 LOG(LS_ERROR) << "Recv() returned error: " << socket_->GetError();
223 ProcessInput(inbuf_, &inpos_);
225 if (inpos_ >= insize_) {
226 LOG(LS_ERROR) << "input buffer overflow";
233 void AsyncTCPSocketBase::OnWriteEvent(AsyncSocket* socket) {
234 ASSERT(socket_.get() == socket);
241 SignalReadyToSend(this);
245 void AsyncTCPSocketBase::OnCloseEvent(AsyncSocket* socket, int error) {
246 SignalClose(this, error);
250 // Binds and connects |socket| and creates AsyncTCPSocket for
251 // it. Takes ownership of |socket|. Returns NULL if bind() or
252 // connect() fail (|socket| is destroyed in that case).
253 AsyncTCPSocket* AsyncTCPSocket::Create(
255 const SocketAddress& bind_address,
256 const SocketAddress& remote_address) {
257 return new AsyncTCPSocket(AsyncTCPSocketBase::ConnectSocket(
258 socket, bind_address, remote_address), false);
261 AsyncTCPSocket::AsyncTCPSocket(AsyncSocket* socket, bool listen)
262 : AsyncTCPSocketBase(socket, listen, kBufSize) {
265 int AsyncTCPSocket::Send(const void *pv, size_t cb,
266 const talk_base::PacketOptions& options) {
272 // If we are blocking on send, then silently drop this packet
273 if (!IsOutBufferEmpty())
274 return static_cast<int>(cb);
276 PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
277 AppendToOutBuffer(&pkt_len, kPacketLenSize);
278 AppendToOutBuffer(pv, cb);
280 int res = FlushOutBuffer();
282 // drop packet if we made no progress
287 // We claim to have sent the whole thing, even if we only sent partial
288 return static_cast<int>(cb);
291 void AsyncTCPSocket::ProcessInput(char * data, size_t* len) {
292 SocketAddress remote_addr(GetRemoteAddress());
295 if (*len < kPacketLenSize)
298 PacketLength pkt_len = talk_base::GetBE16(data);
299 if (*len < kPacketLenSize + pkt_len)
302 SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
303 CreatePacketTime(0));
305 *len -= kPacketLenSize + pkt_len;
307 memmove(data, data + kPacketLenSize + pkt_len, *len);
312 void AsyncTCPSocket::HandleIncomingConnection(AsyncSocket* socket) {
313 SignalNewConnection(this, new AsyncTCPSocket(socket, false));
316 } // namespace talk_base