socket.cpp

贡献一份commoncpp2,有兴趣的可以研究一下

#else

void TCPStream::connect(const char *target, unsigned mss)
{
        char namebuf[128];
        char *cp;
        bool connected = false;
	struct servent *svc;
	tpport_t port;

        snprintf(namebuf, sizeof(namebuf), "%s", target);
        cp = strrchr(namebuf, '/');
        if(!cp)
                cp = strrchr(namebuf, ':');

        if(!cp)
        {
                endStream();
                connectError();
                return;
        }

	*(cp++) = 0;

	if(isdigit(*cp))
		port = atoi(cp);
	else
	{
		mutex.enter();
		svc = getservbyname(cp, "tcp");
		if(svc)
			port = ntohs(svc->s_port);
		mutex.leave();
		if(!svc)
		{
			endStream();
			connectError();
			return;
		}
	}
	
	switch(family)
	{
	case IPV4:
		connect(IPV4Host(namebuf), port, mss);
		break;
#ifdef	CCXX_IPV6
	case IPV6:
		connect(IPV6Host(namebuf), port, mss);
		break;
#endif
	default:
		endStream();
		connectError();
	}
}

#endif

void TCPStream::connect(const IPV4Host &host, tpport_t port, unsigned mss)
{
	size_t i;
	fd_set fds;
	struct timeval to;
	bool connected = false;
	int rtn;
	long sockopt;
	socklen_t len = sizeof(sockopt);
	socklen_t alen = sizeof(mss);

#ifdef	TCP_MAXSEG
	if(mss)
		setsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&mss, sizeof(mss));
#endif

	for(i = 0 ; i < host.getAddressCount(); i++)
	{
		struct sockaddr_in addr;
		memset(&addr, 0, sizeof(addr));
		addr.sin_family = AF_INET;
		addr.sin_addr = host.getAddress(i);
		addr.sin_port = htons(port);
	
		if(timeout)
			setCompletion(false);

		// Win32 will crash if you try to connect to INADDR_ANY.
		if ( INADDR_ANY == addr.sin_addr.s_addr )
			addr.sin_addr.s_addr = INADDR_LOOPBACK;
		rtn = ::connect(so, (struct sockaddr *)&addr, (socklen_t)sizeof(addr));
		if(!rtn)
		{
			connected = true;
			break;
		}

#ifndef WIN32
		if(errno == EINPROGRESS)
#else
		if(WSAGetLastError() == WSAEINPROGRESS)
#endif
		{
			FD_ZERO(&fds);
			FD_SET(so, &fds);
			to.tv_sec = timeout / 1000;
			to.tv_usec = timeout % 1000 * 1000;
			
			// timeout check for connect completion

			if(::select((int)so + 1, NULL, &fds, NULL, &to) < 1)
				continue;

			getsockopt(so, SOL_SOCKET, SO_ERROR, (char *)&sockopt, &len);
			if(!sockopt)
			{
				connected = true;
				break;
			}
			endSocket();
			so = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
			if(so == INVALID_SOCKET)
				break;
		}
	}

	setCompletion(true);
	if(!connected)
	{
		rtn = errno;
		endStream();
		errno = rtn;
		connectError();
		return;
	}

	segmentBuffering(mss);
	Socket::state = CONNECTED;
}

#ifdef  CCXX_IPV6
void TCPStream::connect(const IPV6Host &host, tpport_t port, unsigned mss)
{
	size_t i;
	fd_set fds;
	struct timeval to;
	bool connected = false;
	int rtn;
	long sockopt;
	socklen_t len = sizeof(sockopt);
	socklen_t alen = sizeof(mss);

#ifdef	TCP_MAXSEG
	if(mss)
		setsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&mss, sizeof(mss));
#endif

	for(i = 0 ; i < host.getAddressCount(); i++)
	{
		struct sockaddr_in6 addr;
		memset(&addr, 0, sizeof(addr));
		addr.sin6_family = AF_INET6;
		addr.sin6_addr = host.getAddress(i);
		addr.sin6_port = htons(port);
	
		if(timeout)
			setCompletion(false);

		// Win32 will crash if you try to connect to INADDR_ANY.
		if ( !memcmp(&addr.sin6_addr, &in6addr_any, sizeof(in6addr_any)))
			memcpy(&addr.sin6_addr, &in6addr_loopback, sizeof(in6addr_loopback));
		rtn = ::connect(so, (struct sockaddr *)&addr, (socklen_t)sizeof(addr));
		if(!rtn)
		{
			connected = true;
			break;
		}

#ifndef WIN32
		if(errno == EINPROGRESS)
#else
		if(WSAGetLastError() == WSAEINPROGRESS)
#endif
		{
			FD_ZERO(&fds);
			FD_SET(so, &fds);
			to.tv_sec = timeout / 1000;
			to.tv_usec = timeout % 1000 * 1000;
			
			// timeout check for connect completion

			if(::select((int)so + 1, NULL, &fds, NULL, &to) < 1)
				continue;

			getsockopt(so, SOL_SOCKET, SO_ERROR, (char *)&sockopt, &len);
			if(!sockopt)
			{
				connected = true;
				break;
			}
			endSocket();
			so = socket(AF_INET6, SOCK_STREAM, IPPROTO_TCP);
			if(so == INVALID_SOCKET)
				break;
		}
	}

	setCompletion(true);
	if(!connected)
	{
		rtn = errno;
		endStream();
		errno = rtn;
		connectError();
		return;
	}

	segmentBuffering(mss);
	Socket::state = CONNECTED;
}
#endif

TCPStream::TCPStream(const char *target, Family fam, unsigned mss, bool throwflag, timeout_t to) :
        streambuf(), Socket(PF_INET, SOCK_STREAM, IPPROTO_TCP),
#ifdef  HAVE_OLD_IOSTREAM
        iostream(),
#else
        iostream((streambuf *)this),
#endif
	timeout(to),
	bufsize(0),gbuf(NULL),pbuf(NULL)
{
	family = fam;
#ifdef	HAVE_OLD_IOSTREAM
	init((streambuf *)this);
#endif
	setError(throwflag);
	connect(target, mss);
}

TCPStream::TCPStream(Family fam, bool throwflag, timeout_t to) :
	streambuf(), Socket(PF_INET, SOCK_STREAM, IPPROTO_TCP),
#ifdef	HAVE_OLD_IOSTREAM
	iostream(),
#else
        iostream((streambuf *)this),
#endif
	timeout(to),
        bufsize(0),gbuf(NULL),pbuf(NULL)
{
	family = fam;
#ifdef	HAVE_OLD_IOSTREAM
	init((streambuf *)this);
#endif
	setError(throwflag);
}

TCPStream::TCPStream(const TCPStream &source) :
streambuf(), Socket(DUP_SOCK(source.so,source.state)),
#ifdef	HAVE_OLD_IOSTREAM
iostream()
#else
iostream((streambuf *)this) 
#endif
{
	family = source.family;
#ifdef	HAVE_OLD_IOSTREAM
	init((streambuf *)this);
#endif
	bufsize = source.bufsize;
	allocate(bufsize);
}

void TCPStream::connect(TCPSocket &tcpip)
{
	tpport_t port;
	
	endStream();
	family = IPV4;
	so = accept(tcpip.getSocket(), NULL, NULL);
	if(so == INVALID_SOCKET)
		return;

	IPV4Host host = getPeer(&port);
	if(!tcpip.onAccept(host, port))
	{
		endSocket();
		clear(ios::failbit | rdstate());
		return;
	}

	segmentBuffering(tcpip.getSegmentSize());
	Socket::state = CONNECTED;
}	

#ifdef	CCXX_IPV6

void TCPStream::connect(TCPV6Socket &tcpip)
{
        tpport_t port;

        endStream();
        family = IPV6;
        so = accept(tcpip.getSocket(), NULL, NULL);
        if(so == INVALID_SOCKET)
                return;

        IPV6Host host = getIPV6Peer(&port);
        if(!tcpip.onAccept(host, port))
        {
                endSocket();
                clear(ios::failbit | rdstate());
                return;
        }

	segmentBuffering(tcpip.getSegmentSize());
        Socket::state = CONNECTED;
}
#endif

void TCPStream::segmentBuffering(unsigned mss)
{
	unsigned max = 0;
	socklen_t alen = sizeof(max);

	if(mss == 1)	// special interactive
	{
		allocate(1);
		return;
	}

#ifdef	TCP_MAXSEG
	if(mss)
		setsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&max, sizeof(max));
	getsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&max, &alen);
#endif

	if(max && max < mss)
		mss = max;

	if(!mss)
	{
		if(max)
			mss = max;
		else
			mss = 536;
		allocate(mss);
		return;
	}

#ifdef	TCP_MAXSEG
	setsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&mss, sizeof(mss));
#endif

	if(mss < 80)
		mss = 80;

	if(mss * 7 < 64000)
		bufferSize(mss * 7);
	else if(mss * 6 < 64000)
		bufferSize(mss * 6);
	else
		bufferSize(mss * 5);

	if(mss < 512)
		sendLimit(mss * 4);

	allocate(mss);
}	

int TCPStream::getSegmentSize(void)
{
	unsigned mss = 0;
#ifdef	TCP_MAXSEG
	socklen_t alen = sizeof(mss);
	
	getsockopt(so, IPPROTO_TCP, TCP_MAXSEG, (char *)&mss, &alen);
#endif
	if(!mss)
		return (int)bufsize;

	return mss;	
}

void TCPStream::disconnect(void)
{
	if(Socket::state == AVAILABLE)
		return;

	endStream();
	so = socket(family, SOCK_STREAM, IPPROTO_TCP);
	if(so != INVALID_SOCKET)
		Socket::state = AVAILABLE;
}

void TCPStream::endStream(void)
{
	if(bufsize) sync();
	if(gbuf)
		delete[] gbuf;
	if(pbuf)
		delete[] pbuf;
	gbuf = pbuf = NULL;
	bufsize = 0;
	clear();
	endSocket();
}

void TCPStream::allocate(size_t size)
{
	if(size < 2)
	{
		bufsize = 1;
		gbuf = pbuf = 0;
		return;
	}

	gbuf = new char[size];
	pbuf = new char[size];
	if(!pbuf || !gbuf)
	{
		error(errResourceFailure, "Could not allocate socket stream buffers");
		return;
	}
	bufsize = size;
	clear();

#if (defined(__GNUC__) && (__GNUC__ < 3)) && !defined(WIN32) && !defined(STLPORT)
	setb(gbuf, gbuf + size, 0);
#endif
	setg(gbuf, gbuf + size, gbuf + size);
	setp(pbuf, pbuf + size);
}

int TCPStream::doallocate()
{
	if(bufsize)
		return 0;

	allocate(1);
	return 1;
}

int TCPStream::uflow()
{
	int ret = underflow();

	if (ret == EOF)
		return EOF; 

	if (bufsize != 1)
		gbump(1); 

	return ret; 
}

int TCPStream::underflow()
{
	ssize_t rlen = 1;
	unsigned char ch;

	if(bufsize == 1)
	{
		if(Socket::state == STREAM)
			rlen = ::read((int)so, (char *)&ch, 1);
		else if(timeout && !Socket::isPending(pendingInput, timeout))
        {
            clear(ios::failbit | rdstate());
            error(errTimeout,"Socket read timed out",socket_errno);
            return EOF;
		}
		else
			rlen = ::recv(so, (char *)&ch, 1, 0);
		if(rlen < 1)
		{
			if(rlen < 0)
			{
				clear(ios::failbit | rdstate());
				error(errInput,"Could not read from socket",socket_errno);
			}
			return EOF;
		}
		return ch;
	}

	if(!gptr())
		return EOF;

	if(gptr() < egptr())
		return (unsigned char)*gptr();

	rlen = (ssize_t)((gbuf + bufsize) - eback());
	if(Socket::state == STREAM)
		rlen = ::read((int)so, (char *)eback(), _IOLEN64 rlen);
	else if(timeout && !Socket::isPending(pendingInput, timeout))
	{
		clear(ios::failbit | rdstate());
		error(errTimeout,"Socket read timed out",socket_errno);
		return EOF;
	}
	else
		rlen = ::recv(so, (char *)eback(), rlen, 0);
	if(rlen < 1)
	{
//		clear(ios::failbit | rdstate());
		if(rlen < 0)
                        error(errNotConnected,"Connection error",socket_errno);
		else
		{
			error(errInput,"Could not read from socket",socket_errno);
			clear(ios::failbit | rdstate());
		}
		return EOF;
	}
	error(errSuccess);

	setg(eback(), eback(), eback() + rlen);
	return (unsigned char) *gptr();
}

bool TCPStream::isPending(Pending pending, timeout_t timer)
{
	if(pending == pendingInput && in_avail())
		return true;
	else if(pending == pendingOutput)
		flush();

	return Socket::isPending(pending, timer);
}

int TCPStream::sync(void)
{
	overflow(EOF);
	setg(gbuf, gbuf + bufsize, gbuf + bufsize);
	return 0;
}

#ifdef	HAVE_SNPRINTF
size_t TCPStream::printf(const char *format, ...)
{
	va_list args;
	size_t len;
	char *buf;

	va_start(args, format);
	overflow(EOF);
	len = pptr() - pbase();
	buf = pptr();
	vsnprintf(buf, len, format, args);
	va_end(args);
	len = strlen(buf);
	if(Socket::state == STREAM)
		return ::write((int)so, buf, _IOLEN64 len);
	else
		return ::send(so, buf, _IOLEN64 len, 0);
}
#endif

int TCPStream::overflow(int c)
{
	unsigned char ch;
	ssize_t rlen, req;

	if(bufsize == 1)
	{
		if(c == EOF)
			return 0;

		ch = (unsigned char)(c);
		if(Socket::state == STREAM)
			rlen = ::write((int)so, (const char *)&ch, 1);
		else
			rlen = ::send(so, (const char *)&ch, 1, 0);
		if(rlen < 1)
		{
			if(rlen < 0)
			{
				clear(ios::failbit | rdstate());
				error(errOutput,"Could not write to socket",socket_errno);
			}
			return EOF;
		}
		else
			return c;
	}

	if(!pbase())
		return EOF;

	req = (ssize_t)(pptr() - pbase());
	if(req)
	{
		if(Socket::state == STREAM)
			rlen = ::write((int)so, (const char *)pbase(), req);
		else
			rlen = ::send(so, (const char *)pbase(), req, 0);
		if(rlen < 1)
		{
			if(rlen < 0)
			{
				clear(ios::failbit | rdstate());
				error(errOutput,"Could not write to socket",socket_errno);
			}
			return EOF;
		}
		req -= rlen;
	}

	// if write "partial", rebuffer remainder

	if(req)
//		memmove(pbuf, pptr() + rlen, req);
		memmove(pbuf, pbuf + rlen, req);
	setp(pbuf, pbuf + bufsize);
	pbump(req);

	if(c != EOF)
	{
		*pptr() = (unsigned char)c;
		pbump(1);
	}
	return c;
}

TCPSession::TCPSession(const IPV4Host &ia, 
		       tpport_t port, size_t size, int pri, size_t stack) :
Thread(pri, stack), TCPStream(IPV4)
{
	setCompletion(false);
	setError(false);
	allocate(size);
	
	size_t i;
	for(i = 0 ; i < ia.getAddressCount(); i++)
	{
		struct sockaddr_i