src/third_party/libjingle/source/talk/base/asynctcpsocket.cc

   1 /*
   2  * libjingle
   3  * Copyright 2004--2010, Google Inc.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   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.
  15  *
  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.
  26  */
  27
  28 #include "talk/base/asynctcpsocket.h"
  29
  30 #include <cstring>
  31
  32 #include "talk/base/byteorder.h"
  33 #include "talk/base/common.h"
  34 #include "talk/base/logging.h"
  35
  36 #ifdef POSIX
  37 #include <errno.h>
  38 #endif  // POSIX
  39
  40 namespace talk_base {
  41
  42 static const size_t kMaxPacketSize = 64 * 1024;
  43
  44 typedef uint16 PacketLength;
  45 static const size_t kPacketLenSize = sizeof(PacketLength);
  46
  47 static const size_t kBufSize = kMaxPacketSize + kPacketLenSize;
  48
  49 static const int kListenBacklog = 5;
  50
  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();
  59     return NULL;
  60   }
  61   if (socket->Connect(remote_address) < 0) {
  62     LOG(LS_ERROR) << "Connect() failed with error " << socket->GetError();
  63     return NULL;
  64   }
  65   return owned_socket.release();
  66 }
  67
  68 AsyncTCPSocketBase::AsyncTCPSocketBase(AsyncSocket* socket, bool listen,
  69                                        size_t max_packet_size)
  70     : socket_(socket),
  71       listen_(listen),
  72       insize_(max_packet_size),
  73       inpos_(0),
  74       outsize_(max_packet_size),
  75       outpos_(0) {
  76   inbuf_ = new char[insize_];
  77   outbuf_ = new char[outsize_];
  78
  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);
  85
  86   if (listen_) {
  87     if (socket_->Listen(kListenBacklog) < 0) {
  88       LOG(LS_ERROR) << "Listen() failed with error " << socket_->GetError();
  89     }
  90   }
  91 }
  92
  93 AsyncTCPSocketBase::~AsyncTCPSocketBase() {
  94   delete [] inbuf_;
  95   delete [] outbuf_;
  96 }
  97
  98 SocketAddress AsyncTCPSocketBase::GetLocalAddress() const {
  99   return socket_->GetLocalAddress();
 100 }
 101
 102 SocketAddress AsyncTCPSocketBase::GetRemoteAddress() const {
 103   return socket_->GetRemoteAddress();
 104 }
 105
 106 int AsyncTCPSocketBase::Close() {
 107   return socket_->Close();
 108 }
 109
 110 AsyncTCPSocket::State AsyncTCPSocketBase::GetState() const {
 111   switch (socket_->GetState()) {
 112     case Socket::CS_CLOSED:
 113       return STATE_CLOSED;
 114     case Socket::CS_CONNECTING:
 115       if (listen_) {
 116         return STATE_BOUND;
 117       } else {
 118         return STATE_CONNECTING;
 119       }
 120     case Socket::CS_CONNECTED:
 121       return STATE_CONNECTED;
 122     default:
 123       ASSERT(false);
 124       return STATE_CLOSED;
 125   }
 126 }
 127
 128 int AsyncTCPSocketBase::GetOption(Socket::Option opt, int* value) {
 129   return socket_->GetOption(opt, value);
 130 }
 131
 132 int AsyncTCPSocketBase::SetOption(Socket::Option opt, int value) {
 133   return socket_->SetOption(opt, value);
 134 }
 135
 136 int AsyncTCPSocketBase::GetError() const {
 137   return socket_->GetError();
 138 }
 139
 140 void AsyncTCPSocketBase::SetError(int error) {
 141   return socket_->SetError(error);
 142 }
 143
 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);
 149
 150   ASSERT(false);
 151   socket_->SetError(ENOTCONN);
 152   return -1;
 153 }
 154
 155 int AsyncTCPSocketBase::SendRaw(const void * pv, size_t cb) {
 156   if (outpos_ + cb > outsize_) {
 157     socket_->SetError(EMSGSIZE);
 158     return -1;
 159   }
 160
 161   memcpy(outbuf_ + outpos_, pv, cb);
 162   outpos_ += cb;
 163
 164   return FlushOutBuffer();
 165 }
 166
 167 int AsyncTCPSocketBase::FlushOutBuffer() {
 168   int res = socket_->Send(outbuf_, outpos_);
 169   if (res <= 0) {
 170     return res;
 171   }
 172   if (static_cast<size_t>(res) <= outpos_) {
 173     outpos_ -= res;
 174   } else {
 175     ASSERT(false);
 176     return -1;
 177   }
 178   if (outpos_ > 0) {
 179     memmove(outbuf_, outbuf_ + res, outpos_);
 180   }
 181   return res;
 182 }
 183
 184 void AsyncTCPSocketBase::AppendToOutBuffer(const void* pv, size_t cb) {
 185   ASSERT(outpos_ + cb < outsize_);
 186   memcpy(outbuf_ + outpos_, pv, cb);
 187   outpos_ += cb;
 188 }
 189
 190 void AsyncTCPSocketBase::OnConnectEvent(AsyncSocket* socket) {
 191   SignalConnect(this);
 192 }
 193
 194 void AsyncTCPSocketBase::OnReadEvent(AsyncSocket* socket) {
 195   ASSERT(socket_.get() == socket);
 196
 197   if (listen_) {
 198     talk_base::SocketAddress address;
 199     talk_base::AsyncSocket* new_socket = socket->Accept(&address);
 200     if (!new_socket) {
 201       // TODO: Do something better like forwarding the error
 202       // to the user.
 203       LOG(LS_ERROR) << "TCP accept failed with error " << socket_->GetError();
 204       return;
 205     }
 206
 207     HandleIncomingConnection(new_socket);
 208
 209     // Prime a read event in case data is waiting.
 210     new_socket->SignalReadEvent(new_socket);
 211   } else {
 212     int len = socket_->Recv(inbuf_ + inpos_, insize_ - inpos_);
 213     if (len < 0) {
 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();
 217       }
 218       return;
 219     }
 220
 221     inpos_ += len;
 222
 223     ProcessInput(inbuf_, &inpos_);
 224
 225     if (inpos_ >= insize_) {
 226       LOG(LS_ERROR) << "input buffer overflow";
 227       ASSERT(false);
 228       inpos_ = 0;
 229     }
 230   }
 231 }
 232
 233 void AsyncTCPSocketBase::OnWriteEvent(AsyncSocket* socket) {
 234   ASSERT(socket_.get() == socket);
 235
 236   if (outpos_ > 0) {
 237     FlushOutBuffer();
 238   }
 239
 240   if (outpos_ == 0) {
 241     SignalReadyToSend(this);
 242   }
 243 }
 244
 245 void AsyncTCPSocketBase::OnCloseEvent(AsyncSocket* socket, int error) {
 246   SignalClose(this, error);
 247 }
 248
 249 // AsyncTCPSocket
 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(
 254     AsyncSocket* socket,
 255     const SocketAddress& bind_address,
 256     const SocketAddress& remote_address) {
 257   return new AsyncTCPSocket(AsyncTCPSocketBase::ConnectSocket(
 258       socket, bind_address, remote_address), false);
 259 }
 260
 261 AsyncTCPSocket::AsyncTCPSocket(AsyncSocket* socket, bool listen)
 262     : AsyncTCPSocketBase(socket, listen, kBufSize) {
 263 }
 264
 265 int AsyncTCPSocket::Send(const void *pv, size_t cb,
 266                          const talk_base::PacketOptions& options) {
 267   if (cb > kBufSize) {
 268     SetError(EMSGSIZE);
 269     return -1;
 270   }
 271
 272   // If we are blocking on send, then silently drop this packet
 273   if (!IsOutBufferEmpty())
 274     return static_cast<int>(cb);
 275
 276   PacketLength pkt_len = HostToNetwork16(static_cast<PacketLength>(cb));
 277   AppendToOutBuffer(&pkt_len, kPacketLenSize);
 278   AppendToOutBuffer(pv, cb);
 279
 280   int res = FlushOutBuffer();
 281   if (res <= 0) {
 282     // drop packet if we made no progress
 283     ClearOutBuffer();
 284     return res;
 285   }
 286
 287   // We claim to have sent the whole thing, even if we only sent partial
 288   return static_cast<int>(cb);
 289 }
 290
 291 void AsyncTCPSocket::ProcessInput(char * data, size_t* len) {
 292   SocketAddress remote_addr(GetRemoteAddress());
 293
 294   while (true) {
 295     if (*len < kPacketLenSize)
 296       return;
 297
 298     PacketLength pkt_len = talk_base::GetBE16(data);
 299     if (*len < kPacketLenSize + pkt_len)
 300       return;
 301
 302     SignalReadPacket(this, data + kPacketLenSize, pkt_len, remote_addr,
 303                      CreatePacketTime(0));
 304
 305     *len -= kPacketLenSize + pkt_len;
 306     if (*len > 0) {
 307       memmove(data, data + kPacketLenSize + pkt_len, *len);
 308     }
 309   }
 310 }
 311
 312 void AsyncTCPSocket::HandleIncomingConnection(AsyncSocket* socket) {
 313   SignalNewConnection(this, new AsyncTCPSocket(socket, false));
 314 }
 315
 316 }  // namespace talk_base