Note: We no longer publish the latest version of our code here. We primarily use a kumc-bmi github organization. The heron ETL repository, in particular, is not public. Peers in the informatics community should see MultiSiteDev for details on requesting access.

source: webrtc/talk/p2p/base/tcpport.cc @ 0:4bda6873e34c

pub_scrub_3792 tip
Last change on this file since 0:4bda6873e34c was 0:4bda6873e34c, checked in by Michael Prittie <mprittie@…>, 6 years ago

Scrubbed password for publication.

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