Search Site ~Sockets~  C++ Sockets   Tutorial  Classes  License   Java Sockets  ~Examples~  SMTPProxy   PortProxy   ChatServer   Small MUD   ClientServer   HttpGet   DChat   WebServer   UDP   ICMP  ~Contact~ message board  email author

TcpSocket.cpp Go to the documentation of this file. /* Copyright (C) 2004-2007 Anders Hedstrom This library is made available under the terms of the GNU GPL. If you would like to use this library in a closed-source application, a separate license agreement is available. For information about the closed-source license agreement for the C++ sockets library, please visit http://www.alhem.net/Sockets/license.html and/or email license@alhem.net. This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. */ #ifdef _WIN32 #ifdef _MSC_VER #pragma warning(disable:4786) #endif #include <stdlib.h> #else #include <errno.h> #endif #include "ISocketHandler.h" #include <fcntl.h> #include <assert.h> #include <stdarg.h> #ifdef HAVE_OPENSSL #include <openssl/rand.h> #include <openssl/err.h> #endif #include <map> #include "TcpSocket.h" #include "Utility.h" #include "Ipv4Address.h" #include "Ipv6Address.h" #include "IFile.h" #include "Lock.h" #ifdef SOCKETS_NAMESPACE namespace SOCKETS_NAMESPACE { #endif #ifdef _DEBUG #define DEB(x) x #else #define DEB(x) #endif // statics #ifdef HAVE_OPENSSL SSLInitializer TcpSocket::m_ssl_init; Mutex TcpSocket::m_server_ssl_mutex; std::map<std::string, SSL_CTX *> TcpSocket::m_client_contexts; std::map<std::string, SSL_CTX *> TcpSocket::m_server_contexts; #endif // thanks, q #ifdef _MSC_VER #pragma warning(disable:4355) #endif TcpSocket::TcpSocket(ISocketHandler& h) : StreamSocket(h) ,ibuf(TCP_BUFSIZE_READ) ,m_b_input_buffer_disabled(false) ,m_bytes_sent(0) ,m_bytes_received(0) ,m_skip_c(false) #ifdef SOCKETS_DYNAMIC_TEMP ,m_buf(new char[TCP_BUFSIZE_READ + 1]) #endif ,m_obuf_top(NULL) ,m_transfer_limit(0) ,m_output_length(0) #ifdef HAVE_OPENSSL ,m_ssl_ctx(NULL) ,m_ssl(NULL) ,m_sbio(NULL) #endif #ifdef ENABLE_SOCKS4 ,m_socks4_state(0) #endif #ifdef ENABLE_RESOLVER ,m_resolver_id(0) #endif #ifdef ENABLE_RECONNECT ,m_b_reconnect(false) ,m_b_is_reconnect(false) #endif { } #ifdef _MSC_VER #pragma warning(default:4355) #endif #ifdef _MSC_VER #pragma warning(disable:4355) #endif TcpSocket::TcpSocket(ISocketHandler& h,size_t isize,size_t osize) : StreamSocket(h) ,ibuf(isize) ,m_b_input_buffer_disabled(false) ,m_bytes_sent(0) ,m_bytes_received(0) ,m_skip_c(false) #ifdef SOCKETS_DYNAMIC_TEMP ,m_buf(new char[TCP_BUFSIZE_READ + 1]) #endif ,m_obuf_top(NULL) ,m_transfer_limit(0) ,m_output_length(0) #ifdef HAVE_OPENSSL ,m_ssl_ctx(NULL) ,m_ssl(NULL) ,m_sbio(NULL) #endif #ifdef ENABLE_SOCKS4 ,m_socks4_state(0) #endif #ifdef ENABLE_RESOLVER ,m_resolver_id(0) #endif #ifdef ENABLE_RECONNECT ,m_b_reconnect(false) ,m_b_is_reconnect(false) #endif { } #ifdef _MSC_VER #pragma warning(default:4355) #endif TcpSocket::~TcpSocket() { #ifdef SOCKETS_DYNAMIC_TEMP delete[] m_buf; #endif // %! empty m_obuf while (m_obuf.size()) { output_l::iterator it = m_obuf.begin(); OUTPUT *p = *it; delete p; m_obuf.erase(it); } #ifdef HAVE_OPENSSL if (m_ssl) { SSL_free(m_ssl); } #endif } bool TcpSocket::Open(ipaddr_t ip,port_t port,bool skip_socks) { Ipv4Address ad(ip, port); Ipv4Address local; return Open(ad, local, skip_socks); } #ifdef ENABLE_IPV6 #ifdef IPPROTO_IPV6 bool TcpSocket::Open(in6_addr ip,port_t port,bool skip_socks) { Ipv6Address ad(ip, port); return Open(ad, skip_socks); } #endif #endif bool TcpSocket::Open(SocketAddress& ad,bool skip_socks) { Ipv4Address bind_ad("0.0.0.0", 0); return Open(ad, bind_ad, skip_socks); } bool TcpSocket::Open(SocketAddress& ad,SocketAddress& bind_ad,bool skip_socks) { if (!ad.IsValid()) { Handler().LogError(this, "Open", 0, "Invalid SocketAddress", LOG_LEVEL_FATAL); SetCloseAndDelete(); return false; } if (Handler().GetCount() >= FD_SETSIZE) { Handler().LogError(this, "Open", 0, "no space left in fd_set", LOG_LEVEL_FATAL); SetCloseAndDelete(); return false; } SetConnecting(false); #ifdef ENABLE_SOCKS4 SetSocks4(false); #endif // check for pooling #ifdef ENABLE_POOL if (Handler().PoolEnabled()) { ISocketHandler::PoolSocket *pools = Handler().FindConnection(SOCK_STREAM, "tcp", ad); if (pools) { CopyConnection( pools ); delete pools; SetIsClient(); SetCallOnConnect(); // ISocketHandler must call OnConnect Handler().LogError(this, "SetCallOnConnect", 0, "Found pooled connection", LOG_LEVEL_INFO); return true; } } #endif // if not, create new connection SOCKET s = CreateSocket(ad.GetFamily(), SOCK_STREAM, "tcp"); if (s == INVALID_SOCKET) { return false; } // socket must be nonblocking for async connect if (!SetNonblocking(true, s)) { SetCloseAndDelete(); closesocket(s); return false; } #ifdef ENABLE_POOL SetIsClient(); // client because we connect #endif SetClientRemoteAddress(ad); int n = 0; if (bind_ad.GetPort() != 0) { bind(s, bind_ad, bind_ad); } #ifdef ENABLE_SOCKS4 if (!skip_socks && GetSocks4Host() && GetSocks4Port()) { Ipv4Address sa(GetSocks4Host(), GetSocks4Port()); { std::string sockshost; Utility::l2ip(GetSocks4Host(), sockshost); Handler().LogError(this, "Open", 0, "Connecting to socks4 server @ " + sockshost + ":" + Utility::l2string(GetSocks4Port()), LOG_LEVEL_INFO); } SetSocks4(); n = connect(s, sa, sa); SetRemoteAddress(sa); } else #endif { n = connect(s, ad, ad); SetRemoteAddress(ad); } if (n == -1) { // check error code that means a connect is in progress #ifdef _WIN32 if (Errno == WSAEWOULDBLOCK) #else if (Errno == EINPROGRESS) #endif { Attach(s); SetConnecting( true ); // this flag will control fd_set's } else #ifdef ENABLE_SOCKS4 if (Socks4() && Handler().Socks4TryDirect() ) // retry { closesocket(s); return Open(ad, true); } else #endif #ifdef ENABLE_RECONNECT if (Reconnect()) { Handler().LogError(this, "connect: failed, reconnect pending", Errno, StrError(Errno), LOG_LEVEL_INFO); Attach(s); SetConnecting( true ); // this flag will control fd_set's } else #endif { Handler().LogError(this, "connect: failed", Errno, StrError(Errno), LOG_LEVEL_FATAL); SetCloseAndDelete(); closesocket(s); return false; } } else { Attach(s); SetCallOnConnect(); // ISocketHandler must call OnConnect } // 'true' means connected or connecting(not yet connected) // 'false' means something failed return true; } bool TcpSocket::Open(const std::string &host,port_t port) { #ifdef ENABLE_IPV6 #ifdef IPPROTO_IPV6 if (IsIpv6()) { #ifdef ENABLE_RESOLVER if (!Handler().ResolverEnabled() || Utility::isipv6(host) ) { #endif in6_addr a; if (!Utility::u2ip(host, a)) { SetCloseAndDelete(); return false; } Ipv6Address ad(a, port); Ipv6Address local; return Open(ad, local); #ifdef ENABLE_RESOLVER } m_resolver_id = Resolve6(host, port); return true; #endif } #endif #endif #ifdef ENABLE_RESOLVER if (!Handler().ResolverEnabled() || Utility::isipv4(host) ) { #endif ipaddr_t l; if (!Utility::u2ip(host,l)) { SetCloseAndDelete(); return false; } Ipv4Address ad(l, port); Ipv4Address local; return Open(ad, local); #ifdef ENABLE_RESOLVER } // resolve using async resolver thread m_resolver_id = Resolve(host, port); return true; #endif } #ifdef ENABLE_RESOLVER void TcpSocket::OnResolved(int id,ipaddr_t a,port_t port) { DEB( fprintf(stderr, "TcpSocket::OnResolved id %d addr %x port %d\n", id, a, port);) if (id == m_resolver_id) { if (a && port) { Ipv4Address ad(a, port); Ipv4Address local; if (Open(ad, local)) { if (!Handler().Valid(this)) { Handler().Add(this); } } } else { Handler().LogError(this, "OnResolved", 0, "Resolver failed", LOG_LEVEL_FATAL); SetCloseAndDelete(); } } else { Handler().LogError(this, "OnResolved", id, "Resolver returned wrong job id", LOG_LEVEL_FATAL); SetCloseAndDelete(); } } #ifdef ENABLE_IPV6 void TcpSocket::OnResolved(int id,in6_addr& a,port_t port) { if (id == m_resolver_id) { Ipv6Address ad(a, port); if (ad.IsValid()) { Ipv6Address local; if (Open(ad, local)) { if (!Handler().Valid(this)) { Handler().Add(this); } } } } else { Handler().LogError(this, "OnResolved", id, "Resolver returned wrong job id", LOG_LEVEL_FATAL); SetCloseAndDelete(); } } #endif #endif void TcpSocket::OnRead() { int n = 0; #ifdef SOCKETS_DYNAMIC_TEMP char *buf = m_buf; #else char buf[TCP_BUFSIZE_READ]; #endif #ifdef HAVE_OPENSSL if (IsSSL()) { if (!Ready()) return; n = SSL_read(m_ssl, buf, TCP_BUFSIZE_READ); if (n == -1) { n = SSL_get_error(m_ssl, n); switch (n) { case SSL_ERROR_NONE: case SSL_ERROR_WANT_READ: case SSL_ERROR_WANT_WRITE: break; case SSL_ERROR_ZERO_RETURN: DEB( fprintf(stderr, "SSL_read() returns zero - closing socket\n");) OnDisconnect(); OnDisconnect(TCP_DISCONNECT_SSL|TCP_DISCONNECT_ERROR, n); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); break; default: DEB( fprintf(stderr, "SSL read problem, errcode = %d\n",n);) OnDisconnect(); OnDisconnect(TCP_DISCONNECT_SSL|TCP_DISCONNECT_ERROR, n); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); } return; } else if (!n) { OnDisconnect(); OnDisconnect(TCP_DISCONNECT_SSL, 0); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); SetShutdown(SHUT_WR); return; } else if (n > 0 && n <= TCP_BUFSIZE_READ) { m_bytes_received += n; if (GetTrafficMonitor()) { GetTrafficMonitor() -> fwrite(buf, 1, n); } if (!m_b_input_buffer_disabled && !ibuf.Write(buf,n)) { Handler().LogError(this, "OnRead(ssl)", 0, "ibuf overflow", LOG_LEVEL_WARNING); } } else { Handler().LogError(this, "OnRead(ssl)", n, "abnormal value from SSL_read", LOG_LEVEL_ERROR); } } else #endif // HAVE_OPENSSL { n = recv(GetSocket(), buf, TCP_BUFSIZE_READ, MSG_NOSIGNAL); if (n == -1) { Handler().LogError(this, "read", Errno, StrError(Errno), LOG_LEVEL_FATAL); OnDisconnect(); OnDisconnect(TCP_DISCONNECT_ERROR, Errno); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); return; } else if (!n) { OnDisconnect(); OnDisconnect(0, 0); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); SetShutdown(SHUT_WR); return; } else if (n > 0 && n <= TCP_BUFSIZE_READ) { m_bytes_received += n; if (GetTrafficMonitor()) { GetTrafficMonitor() -> fwrite(buf, 1, n); } if (!m_b_input_buffer_disabled && !ibuf.Write(buf,n)) { Handler().LogError(this, "OnRead", 0, "ibuf overflow", LOG_LEVEL_WARNING); } } else { Handler().LogError(this, "OnRead", n, "abnormal value from recv", LOG_LEVEL_ERROR); } } // OnRead( buf, n ); } void TcpSocket::OnRead( char *buf, size_t n ) { // unbuffered if (n > 0 && n <= TCP_BUFSIZE_READ) { if (LineProtocol()) { buf[n] = 0; size_t i = 0; if (m_skip_c && (buf[i] == 13 || buf[i] == 10) && buf[i] != m_c) { m_skip_c = false; i++; } size_t x = i; for (; i < n && LineProtocol(); i++) { while ((buf[i] == 13 || buf[i] == 10) && LineProtocol()) { char c = buf[i]; buf[i] = 0; if (buf[x]) { m_line += (buf + x); } OnLine( m_line ); i++; m_skip_c = true; m_c = c; if (i < n && (buf[i] == 13 || buf[i] == 10) && buf[i] != c) { m_skip_c = false; i++; } x = i; m_line = ""; } if (!LineProtocol()) { break; } } if (!LineProtocol()) { if (i < n) { OnRawData(buf + i, n - i); } } else if (buf[x]) { m_line += (buf + x); } } else { OnRawData(buf, n); } } if (m_b_input_buffer_disabled) { return; } // further processing: socks4 #ifdef ENABLE_SOCKS4 if (Socks4()) { bool need_more = false; while (GetInputLength() && !need_more && !CloseAndDelete()) { need_more = OnSocks4Read(); } } #endif } void TcpSocket::OnWriteComplete() { } void TcpSocket::OnWrite() { if (Connecting()) { int err = SoError(); // don't reset connecting flag on error here, we want the OnConnectFailed timeout later on if (!err) // ok { Set(!IsDisableRead(), false); SetConnecting(false); SetCallOnConnect(); return; } Handler().LogError(this, "tcp: connect failed", err, StrError(err), LOG_LEVEL_FATAL); Set(false, false); // no more monitoring because connection failed // failed #ifdef ENABLE_SOCKS4 if (Socks4()) { // %! leave 'Connecting' flag set? OnSocks4ConnectFailed(); return; } #endif if (GetConnectionRetry() == -1 || (GetConnectionRetry() && GetConnectionRetries() < GetConnectionRetry()) ) { // even though the connection failed at once, only retry after // the connection timeout. // should we even try to connect again, when CheckConnect returns // false it's because of a connection error - not a timeout... return; } SetConnecting(false); SetCloseAndDelete( true ); OnConnectFailed(); return; } // try send next block in buffer // if full block is sent, repeat // if all blocks are sent, reset m_wfds bool repeat = false; size_t sz = m_transfer_limit ? GetOutputLength() : 0; do { output_l::iterator it = m_obuf.begin(); OUTPUT *p = *it; repeat = false; int n = TryWrite(p -> Buf(), p -> Len()); if (n > 0) { size_t left = p -> Remove(n); m_output_length -= n; if (!left) { delete p; m_obuf.erase(it); if (!m_obuf.size()) { m_obuf_top = NULL; OnWriteComplete(); } else { repeat = true; } } } } while (repeat); if (m_transfer_limit && sz > m_transfer_limit && GetOutputLength() < m_transfer_limit) { OnTransferLimit(); } // check output buffer set, set/reset m_wfds accordingly { bool br; bool bw; bool bx; Handler().Get(GetSocket(), br, bw, bx); if (m_obuf.size()) Set(br, true); else Set(br, false); } } int TcpSocket::TryWrite(const char *buf, size_t len) { int n = 0; #ifdef HAVE_OPENSSL if (IsSSL()) { n = SSL_write(m_ssl, buf, (int)len); if (n == -1) { int errnr = SSL_get_error(m_ssl, n); if ( errnr != SSL_ERROR_WANT_READ && errnr != SSL_ERROR_WANT_WRITE ) { OnDisconnect(); OnDisconnect(TCP_DISCONNECT_WRITE|TCP_DISCONNECT_ERROR|TCP_DISCONNECT_SSL, errnr); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); { char errbuf[256]; ERR_error_string_n(errnr, errbuf, 256); Handler().LogError(this, "OnWrite/SSL_write", errnr, errbuf, LOG_LEVEL_FATAL); } } return 0; } else if (!n) { OnDisconnect(); OnDisconnect(TCP_DISCONNECT_WRITE|TCP_DISCONNECT_SSL, 0); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); DEB( { int errnr = SSL_get_error(m_ssl, n); char errbuf[256]; ERR_error_string_n(errnr, errbuf, 256); fprintf(stderr, "SSL_write() returns 0: %d : %s\n",errnr, errbuf); }) } } else #endif // HAVE_OPENSSL { n = send(GetSocket(), buf, (int)len, MSG_NOSIGNAL); if (n == -1) { // normal error codes: // WSAEWOULDBLOCK // EAGAIN or EWOULDBLOCK #ifdef _WIN32 if (Errno != WSAEWOULDBLOCK) #else if (Errno != EWOULDBLOCK) #endif { Handler().LogError(this, "send", Errno, StrError(Errno), LOG_LEVEL_FATAL); OnDisconnect(); OnDisconnect(TCP_DISCONNECT_WRITE|TCP_DISCONNECT_ERROR, Errno); SetCloseAndDelete(true); SetFlushBeforeClose(false); SetLost(); } return 0; } } if (n > 0) { m_bytes_sent += n; if (GetTrafficMonitor()) { GetTrafficMonitor() -> fwrite(buf, 1, n); } } return n; } void TcpSocket::Buffer(const char *buf, size_t len) { size_t ptr = 0; m_output_length += len; while (ptr < len) { // buf/len => pbuf/sz size_t space = 0; if (m_obuf_top && (space = m_obuf_top -> Space()) > 0) { const char *pbuf = buf + ptr; size_t sz = len - ptr; if (space >= sz) { m_obuf_top -> Add(pbuf, sz); ptr += sz; } else { m_obuf_top -> Add(pbuf, space); ptr += space; } } else { m_obuf_top = new OUTPUT; m_obuf.push_back( m_obuf_top ); } } } void TcpSocket::Send(const std::string &str,int i) { SendBuf(str.c_str(),str.size(),i); } void TcpSocket::SendBuf(const char *buf,size_t len,int) { if (!Ready() && !Connecting()) { Handler().LogError(this, "SendBuf", -1, "Attempt to write to a non-ready socket" ); // warning if (GetSocket() == INVALID_SOCKET) Handler().LogError(this, "SendBuf", 0, " * GetSocket() == INVALID_SOCKET", LOG_LEVEL_INFO); if (Connecting()) Handler().LogError(this, "SendBuf", 0, " * Connecting()", LOG_LEVEL_INFO); if (CloseAndDelete()) Handler().LogError(this, "SendBuf", 0, " * CloseAndDelete()", LOG_LEVEL_INFO); return; } if (!IsConnected()) { Handler().LogError(this, "SendBuf", -1, "Attempt to write to a non-connected socket, will be sent on connect" ); // warning Buffer(buf, len); return; } if (m_obuf_top) { Buffer(buf, len); return; } int n = TryWrite(buf, len); if (n >= 0 && n < (int)len) { Buffer(buf + n, len - n); } // if ( data in buffer || !IsConnected ) // { // add to buffer // } // else // try_send // if any data is unsent, buffer it and set m_wfds // check output buffer set, set/reset m_wfds accordingly { bool br; bool bw; bool bx; Handler().Get(GetSocket(), br, bw, bx); if (