Home | History | Annotate | Download | only in base 
      1 /*
      2  * libjingle
      3  * Copyright 2004--2005, 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/p2p/base/tcpport.h"
     29 
     30 #include "talk/base/common.h"
     31 #include "talk/base/logging.h"
     32 #include "talk/p2p/base/common.h"
     33 
     34 namespace cricket {
     35 
     36 TCPPort::TCPPort(talk_base::Thread* thread,
     37                  talk_base::PacketSocketFactory* factory,
     38                  talk_base::Network* network, const talk_base::IPAddress& ip,
     39                  int min_port, int max_port, const std::string& username,
     40                  const std::string& password, bool allow_listen)
     41     : Port(thread, LOCAL_PORT_TYPE, factory, network, ip, min_port, max_port,
     42            username, password),
     43       incoming_only_(false),
     44       allow_listen_(allow_listen),
     45       socket_(NULL),
     46       error_(0) {
     47   // TODO(mallinath) - Set preference value as per RFC 6544.
     48   // http://b/issue?id=7141794
     49 }
     50 
     51 bool TCPPort::Init() {
     52   if (allow_listen_) {
     53     // Treat failure to create or bind a TCP socket as fatal.  This
     54     // should never happen.
     55     socket_ = socket_factory()->CreateServerTcpSocket(
     56         talk_base::SocketAddress(ip(), 0), min_port(), max_port(),
     57         false /* ssl */);
     58     if (!socket_) {
     59       LOG_J(LS_ERROR, this) << "TCP socket creation failed.";
     60       return false;
     61     }
     62     socket_->SignalNewConnection.connect(this, &TCPPort::OnNewConnection);
     63     socket_->SignalAddressReady.connect(this, &TCPPort::OnAddressReady);
     64   }
     65   return true;
     66 }
     67 
     68 TCPPort::~TCPPort() {
     69   delete socket_;
     70 }
     71 
     72 Connection* TCPPort::CreateConnection(const Candidate& address,
     73                                       CandidateOrigin origin) {
     74   // We only support TCP protocols
     75   if ((address.protocol() != TCP_PROTOCOL_NAME) &&
     76       (address.protocol() != SSLTCP_PROTOCOL_NAME)) {
     77     return NULL;
     78   }
     79 
     80   // We can't accept TCP connections incoming on other ports
     81   if (origin == ORIGIN_OTHER_PORT)
     82     return NULL;
     83 
     84   // Check if we are allowed to make outgoing TCP connections
     85   if (incoming_only_ && (origin == ORIGIN_MESSAGE))
     86     return NULL;
     87 
     88   // We don't know how to act as an ssl server yet
     89   if ((address.protocol() == SSLTCP_PROTOCOL_NAME) &&
     90       (origin == ORIGIN_THIS_PORT)) {
     91     return NULL;
     92   }
     93 
     94   if (!IsCompatibleAddress(address.address())) {
     95     return NULL;
     96   }
     97 
     98   TCPConnection* conn = NULL;
     99   if (talk_base::AsyncPacketSocket* socket =
    100       GetIncoming(address.address(), true)) {
    101     socket->SignalReadPacket.disconnect(this);
    102     conn = new TCPConnection(this, address, socket);
    103   } else {
    104     conn = new TCPConnection(this, address);
    105   }
    106   AddConnection(conn);
    107   return conn;
    108 }
    109 
    110 void TCPPort::PrepareAddress() {
    111   if (socket_) {
    112     // If socket isn't bound yet the address will be added in
    113     // OnAddressReady(). Socket may be in the CLOSED state if Listen()
    114     // failed, we still want ot add the socket address.
    115     LOG(LS_VERBOSE) << "Preparing TCP address, current state: "
    116                     << socket_->GetState();
    117     if (socket_->GetState() == talk_base::AsyncPacketSocket::STATE_BOUND ||
    118         socket_->GetState() == talk_base::AsyncPacketSocket::STATE_CLOSED)
    119       AddAddress(socket_->GetLocalAddress(), socket_->GetLocalAddress(),
    120                  TCP_PROTOCOL_NAME, LOCAL_PORT_TYPE,
    121                  ICE_TYPE_PREFERENCE_HOST_TCP, true);
    122   } else {
    123     LOG_J(LS_INFO, this) << "Not listening due to firewall restrictions.";
    124     // Note: We still add the address, since otherwise the remote side won't
    125     // recognize our incoming TCP connections.
    126     AddAddress(talk_base::SocketAddress(ip(), 0),
    127                talk_base::SocketAddress(ip(), 0), TCP_PROTOCOL_NAME,
    128                LOCAL_PORT_TYPE, ICE_TYPE_PREFERENCE_HOST_TCP, true);
    129   }
    130 }
    131 
    132 int TCPPort::SendTo(const void* data, size_t size,
    133                     const talk_base::SocketAddress& addr, bool payload) {
    134   talk_base::AsyncPacketSocket * socket = NULL;
    135   if (TCPConnection * conn = static_cast<TCPConnection*>(GetConnection(addr))) {
    136     socket = conn->socket();
    137   } else {
    138     socket = GetIncoming(addr);
    139   }
    140   if (!socket) {
    141     LOG_J(LS_ERROR, this) << "Attempted to send to an unknown destination, "
    142                           << addr.ToSensitiveString();
    143     return -1;  // TODO: Set error_
    144   }
    145 
    146   int sent = socket->Send(data, size);
    147   if (sent < 0) {
    148     error_ = socket->GetError();
    149     LOG_J(LS_ERROR, this) << "TCP send of " << size
    150                           << " bytes failed with error " << error_;
    151   }
    152   return sent;
    153 }
    154 
    155 int TCPPort::GetOption(talk_base::Socket::Option opt, int* value) {
    156   if (socket_) {
    157     return socket_->GetOption(opt, value);
    158   } else {
    159     return SOCKET_ERROR;
    160   }
    161 }
    162 
    163 int TCPPort::SetOption(talk_base::Socket::Option opt, int value) {
    164   if (socket_) {
    165     return socket_->SetOption(opt, value);
    166   } else {
    167     return SOCKET_ERROR;
    168   }
    169 }
    170 
    171 int TCPPort::GetError() {
    172   return error_;
    173 }
    174 
    175 void TCPPort::OnNewConnection(talk_base::AsyncPacketSocket* socket,
    176                               talk_base::AsyncPacketSocket* new_socket) {
    177   ASSERT(socket == socket_);
    178 
    179   Incoming incoming;
    180   incoming.addr = new_socket->GetRemoteAddress();
    181   incoming.socket = new_socket;
    182   incoming.socket->SignalReadPacket.connect(this, &TCPPort::OnReadPacket);
    183   incoming.socket->SignalReadyToSend.connect(this, &TCPPort::OnReadyToSend);
    184 
    185   LOG_J(LS_VERBOSE, this) << "Accepted connection from "
    186                           << incoming.addr.ToSensitiveString();
    187   incoming_.push_back(incoming);
    188 }
    189 
    190 talk_base::AsyncPacketSocket* TCPPort::GetIncoming(
    191     const talk_base::SocketAddress& addr, bool remove) {
    192   talk_base::AsyncPacketSocket* socket = NULL;
    193   for (std::list<Incoming>::iterator it = incoming_.begin();
    194        it != incoming_.end(); ++it) {
    195     if (it->addr == addr) {
    196       socket = it->socket;
    197       if (remove)
    198         incoming_.erase(it);
    199       break;
    200     }
    201   }
    202   return socket;
    203 }
    204 
    205 void TCPPort::OnReadPacket(talk_base::AsyncPacketSocket* socket,
    206                            const char* data, size_t size,
    207                            const talk_base::SocketAddress& remote_addr) {
    208   Port::OnReadPacket(data, size, remote_addr, PROTO_TCP);
    209 }
    210 
    211 void TCPPort::OnReadyToSend(talk_base::AsyncPacketSocket* socket) {
    212   Port::OnReadyToSend();
    213 }
    214 
    215 void TCPPort::OnAddressReady(talk_base::AsyncPacketSocket* socket,
    216                              const talk_base::SocketAddress& address) {
    217   AddAddress(address, address, "tcp",
    218              LOCAL_PORT_TYPE, ICE_TYPE_PREFERENCE_HOST_TCP,
    219              true);
    220 }
    221 
    222 TCPConnection::TCPConnection(TCPPort* port, const Candidate& candidate,
    223                              talk_base::AsyncPacketSocket* socket)
    224     : Connection(port, 0, candidate), socket_(socket), error_(0) {
    225   bool outgoing = (socket_ == NULL);
    226   if (outgoing) {
    227     // TODO: Handle failures here (unlikely since TCP).
    228     int opts = (candidate.protocol() == SSLTCP_PROTOCOL_NAME) ?
    229         talk_base::PacketSocketFactory::OPT_SSLTCP : 0;
    230     socket_ = port->socket_factory()->CreateClientTcpSocket(
    231         talk_base::SocketAddress(port_->Network()->ip(), 0),
    232         candidate.address(), port->proxy(), port->user_agent(), opts);
    233     if (socket_) {
    234       LOG_J(LS_VERBOSE, this) << "Connecting from "
    235                               << socket_->GetLocalAddress().ToSensitiveString()
    236                               << " to "
    237                               << candidate.address().ToSensitiveString();
    238       set_connected(false);
    239       socket_->SignalConnect.connect(this, &TCPConnection::OnConnect);
    240     } else {
    241       LOG_J(LS_WARNING, this) << "Failed to create connection to "
    242                               << candidate.address().ToSensitiveString();
    243     }
    244   } else {
    245     // Incoming connections should match the network address.
    246     ASSERT(socket_->GetLocalAddress().ipaddr() == port->ip());
    247   }
    248 
    249   if (socket_) {
    250     socket_->SignalReadPacket.connect(this, &TCPConnection::OnReadPacket);
    251     socket_->SignalReadyToSend.connect(this, &TCPConnection::OnReadyToSend);
    252     socket_->SignalClose.connect(this, &TCPConnection::OnClose);
    253   }
    254 }
    255 
    256 TCPConnection::~TCPConnection() {
    257   delete socket_;
    258 }
    259 
    260 int TCPConnection::Send(const void* data, size_t size) {
    261   if (!socket_) {
    262     error_ = ENOTCONN;
    263     return SOCKET_ERROR;
    264   }
    265 
    266   if (write_state() != STATE_WRITABLE) {
    267     // TODO: Should STATE_WRITE_TIMEOUT return a non-blocking error?
    268     error_ = EWOULDBLOCK;
    269     return SOCKET_ERROR;
    270   }
    271   int sent = socket_->Send(data, size);
    272   if (sent < 0) {
    273     error_ = socket_->GetError();
    274   } else {
    275     send_rate_tracker_.Update(sent);
    276   }
    277   return sent;
    278 }
    279 
    280 int TCPConnection::GetError() {
    281   return error_;
    282 }
    283 
    284 void TCPConnection::OnConnect(talk_base::AsyncPacketSocket* socket) {
    285   ASSERT(socket == socket_);
    286   LOG_J(LS_VERBOSE, this) << "Connection established to "
    287                           << socket->GetRemoteAddress().ToSensitiveString();
    288   set_connected(true);
    289 }
    290 
    291 void TCPConnection::OnClose(talk_base::AsyncPacketSocket* socket, int error) {
    292   ASSERT(socket == socket_);
    293   LOG_J(LS_VERBOSE, this) << "Connection closed with error " << error;
    294   set_connected(false);
    295   set_write_state(STATE_WRITE_TIMEOUT);
    296 }
    297 
    298 void TCPConnection::OnReadPacket(talk_base::AsyncPacketSocket* socket,
    299                                  const char* data, size_t size,
    300                                  const talk_base::SocketAddress& remote_addr) {
    301   ASSERT(socket == socket_);
    302   Connection::OnReadPacket(data, size);
    303 }
    304 
    305 void TCPConnection::OnReadyToSend(talk_base::AsyncPacketSocket* socket) {
    306   ASSERT(socket == socket_);
    307   Connection::OnReadyToSend();
    308 }
    309 
    310 }  // namespace cricket
    311