~drizzle-trunk/drizzle/development

/* Copyright (C) 2008 PrimeBase Technologies GmbH, Germany

 *

 * PrimeBase Media Stream for MySQL

 *

 * 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

 *

 * Original author: Paul McCullagh (H&G2JCtL)

 * Continued development: Barry Leslie

 *

 * 2007-05-24

 *

 * CORE SYSTEM:

 * Basic socket I/O.

 *

 */

#include "CSConfig.h"

#include <stdio.h>

#include <sys/types.h>

#ifdef OS_WINDOWS

#include <winsock.h>

typedef int socklen_t;

#define SHUT_RDWR 2

#define CLOSE_SOCKET(s)	closesocket(s)

#define IOCTL_SOCKET	ioctlsocket

#define SOCKET_ERRORNO	WSAGetLastError()

#define	EWOULDBLOCK		WSAEWOULDBLOCK

#else

#include <sys/ioctl.h>

#include <sys/socket.h>

#include <netdb.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <netinet/in.h>

#include <netinet/tcp.h>

#include <sys/select.h>

#include <fcntl.h>

extern void unix_close(int h);

#define CLOSE_SOCKET(s)	unix_close(s)

#define IOCTL_SOCKET	ioctl

#define SOCKET_ERRORNO	errno

#endif

#include <ctype.h>

#include <string.h>

#include <stdlib.h>

#include "CSSocket.h"

#include "CSStream.h"

#include "CSGlobal.h"

#include "CSStrUtil.h"

#include "CSFile.h"

void CSSocket::initSockets()

{

#ifdef OS_WINDOWS

	int		err;

	WSADATA	data;

	WORD	version = MAKEWORD (1,1);

	static int inited = 0;

	if (!inited) {

		err = WSAStartup(version, &data);

		if (err != 0) {

			CSException::throwException(CS_CONTEXT, err, "WSAStartup error");

		}

		inited = 1;

	}

#endif

}

/*

 * ---------------------------------------------------------------

 * CORE SYSTEM SOCKET FACTORY

 */

CSSocket *CSSocket::newSocket()

{

	CSSocket *s;

	new_(s, CSSocket());

	return s;

}

/*

 * ---------------------------------------------------------------

 * INTERNAL UTILITIES

 */

void CSSocket::formatAddress(size_t size, char *buffer)

{

	if (iHost) {

		cs_strcpy(size, buffer, iHost);

		if (iService)

			cs_strcat(size, buffer, ":");

	}

	else

		*buffer = 0;

	if (iService)

		cs_strcat(size, buffer, iService);

}

void CSSocket::throwError(const char *func, const char *file, int line, char *address, int err)

{

	if (err)

		CSException::throwFileError(func, file, line, address, err);

	else

		CSException::throwEOFError(func, file, line, address);

}

void CSSocket::throwError(const char *func, const char *file, int line, int err)

{

	char address[CS_SOCKET_ADDRESS_SIZE];

	formatAddress(CS_SOCKET_ADDRESS_SIZE, address);

	throwError(func, file, line, address, err);

}

void CSSocket::setNoDelay()

{

	int flag = 1;

	if (setsockopt(iHandle, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)) == -1)

		CSException::throwOSError(CS_CONTEXT, SOCKET_ERRORNO);

}

void CSSocket::setNonBlocking()

{

	if (iTimeout) {

		unsigned long block = 1;

		if (IOCTL_SOCKET(iHandle, FIONBIO, &block) != 0)

			throwError(CS_CONTEXT, SOCKET_ERRORNO);

	}

}

void CSSocket::setBlocking()

{

	/* No timeout, set blocking: */

	if (!iTimeout) {

		unsigned long block = 0;

		if (IOCTL_SOCKET(iHandle, FIONBIO, &block) != 0)

			throwError(CS_CONTEXT, SOCKET_ERRORNO);

	}

}

void CSSocket::openInternal()

{

	iHandle = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);

	if (iHandle == -1)

		CSException::throwOSError(CS_CONTEXT, SOCKET_ERRORNO);

	setNoDelay();

	setNonBlocking();

}

void CSSocket::writeBlock(const void *data, size_t len)

{

	ssize_t	out;

	enter_();

	while (len > 0) {

		out = send(iHandle, (const char *) data, len, 0);

		self->interrupted();

		if (out == -1) {

			int err = SOCKET_ERRORNO;

			if (err == EWOULDBLOCK || err == EINTR)

				continue;

			throwError(CS_CONTEXT, err);

		}

		if ((size_t) out > len)

			break;

		len -= (size_t) out;

		data = ((char *) data) + (size_t) out;

	}

	exit_();

}

int CSSocket::timeoutRead(CSThread *self, void *buffer, size_t length)

{      

	int			in;

	uint64_t	start_time;

	uint64_t	timeout = iTimeout * 1000;

	start_time = CSTime::getTimeCurrentTicks();

	retry:

	in = recv(iHandle, (char *) buffer, length, 0);

	if (in == -1) {

		if (SOCKET_ERRORNO == EWOULDBLOCK) {

			fd_set			readfds;

			uint64_t		time_diff;

			struct timeval	tv_timeout;

			FD_ZERO(&readfds);

			self->interrupted();

			time_diff = CSTime::getTimeCurrentTicks() - start_time;

			if (time_diff >= timeout) {

				char address[CS_SOCKET_ADDRESS_SIZE];

				formatAddress(CS_SOCKET_ADDRESS_SIZE, address);

				CSException::throwExceptionf(CS_CONTEXT, CS_ERR_RECEIVE_TIMEOUT, "Receive timeout: %lu ms, on: %s", iTimeout, address);

			}

			/* Calculate how much time we can wait: */

			time_diff = timeout - time_diff;

			tv_timeout.tv_sec = (long)time_diff / 1000000;

			tv_timeout.tv_usec = (long)time_diff % 1000000;

			FD_SET(iHandle, &readfds);

			in = select(iHandle+1, &readfds, NULL, NULL, &tv_timeout);

			if (in != -1)

				goto retry;

		}

	}

 	return in;

}

/*

 * ---------------------------------------------------------------

 * SOCKET BASED ON THE STANDARD C SOCKET

 */

void CSSocket::setTimeout(uint32_t milli_sec)

{

	if (iTimeout != milli_sec) {

		if ((iTimeout = milli_sec))

			setNonBlocking();

		else

			setBlocking();

	}

}

CSOutputStream *CSSocket::getOutputStream()

{

	return CSSocketOutputStream::newStream(RETAIN(this));

}

CSInputStream *CSSocket::getInputStream()

{

	return CSSocketInputStream::newStream(RETAIN(this));

}

void CSSocket::publish(char *service, int default_port)

{

	enter_();

	close();

	try_(a) {

		struct servent		*servp;

		struct sockaddr_in	server;

		struct servent		s;

		int					flag = 1;

		openInternal();

		if (service) {

			if (isdigit(service[0])) {

				int i =  atoi(service);

				if (!i)

					CSException::throwCoreError(CS_CONTEXT, CS_ERR_BAD_ADDRESS, service);

				servp = &s;

				s.s_port = htons((uint16_t) i);

				iService = cs_strdup(service);

			}

			else if ((servp = getservbyname(service, "tcp")) == NULL) {

				if (!default_port)

					CSException::throwCoreError(CS_CONTEXT, CS_ERR_UNKNOWN_SERVICE, service);

				servp = &s;

				s.s_port = htons((uint16_t) default_port);

				iService = cs_strdup(default_port);

			}

			else

				iService = cs_strdup(service);

		}

		else {

			if (!default_port)

				CSException::throwCoreError(CS_CONTEXT, CS_ERR_UNKNOWN_SERVICE, "");

			servp = &s;

			s.s_port = htons((uint16_t) default_port);

			iService = cs_strdup(default_port);

		}

		iPort = ntohs(servp->s_port);

		memset(&server, 0, sizeof(server));

		server.sin_family = AF_INET;

		server.sin_addr.s_addr = INADDR_ANY;

		server.sin_port = (uint16_t) servp->s_port;

		if (setsockopt(iHandle, SOL_SOCKET, SO_REUSEADDR, (char *) &flag, sizeof(int)) == -1)

			CSException::throwOSError(CS_CONTEXT, SOCKET_ERRORNO);

		if (bind(iHandle, (struct sockaddr *) &server, sizeof(server)) == -1)

			CSException::throwOSError(CS_CONTEXT, SOCKET_ERRORNO);

		if (listen(iHandle, SOMAXCONN) == -1)

			CSException::throwOSError(CS_CONTEXT, SOCKET_ERRORNO);

	}

	catch_(a) {

		close();

		throw_();

	}

	cont_(a);

	exit_();

}

void CSSocket::open(CSSocket *listener)

{

	enter_();

	close();

	try_(a) {

		int listener_handle;

		char address[CS_SOCKET_ADDRESS_SIZE];

		struct sockaddr_in	remote;

		socklen_t			addrlen = sizeof(remote);

		/* First get all the information we need from the listener: */

		listener_handle = ((CSSocket *) listener)->iHandle;

		listener->formatAddress(CS_SOCKET_ADDRESS_SIZE, address);

		/* I want to make sure no error occurs after the connect!

		 * So I allocate a buffer for the host name up front.

		 * This means it may be to small, but this is not a problem

		 * because the host name stored here is is only used for display

		 * of error message etc.

		 */

		iHost = (char *) cs_malloc(100);

		iHandle = accept(listener_handle, (struct sockaddr *) &remote, &addrlen);

		if (iHandle == -1)

			throwError(CS_CONTEXT, address, SOCKET_ERRORNO);

		cs_strcpy(100, iHost, inet_ntoa(remote.sin_addr));

		iPort = ntohs(remote.sin_port);

		setNoDelay();

		setNonBlocking();

	}

	catch_(a) {

		close();

		throw_();

	}

	cont_(a);

	exit_();

}

void CSSocket::open(char *address, int default_port)

{

	enter_();

	close();

	try_(a) {

		char				*portp = strchr(address, ':');

		struct servent		s;

		struct servent		*servp;

		struct hostent		*hostp;

		struct sockaddr_in	server;

		openInternal();

		if (!portp) {

			iHost = cs_strdup(address);

			if (!default_port)

				CSException::throwCoreError(CS_CONTEXT, CS_ERR_BAD_ADDRESS, address);

			iService = cs_strdup(default_port);

		}

		else {

			iHost = cs_strdup(address, (size_t) (portp - address));

			iService = cs_strdup(portp+1);

		}

		if (isdigit(iService[0])) {

			int i =  atoi(iService);

			if (!i)

				CSException::throwCoreError(CS_CONTEXT, CS_ERR_BAD_ADDRESS, address);

			servp = &s;

			s.s_port = htons((uint16_t) i);

		}

		else if ((servp = getservbyname(iService, "tcp")) == NULL)

			CSException::throwCoreError(CS_CONTEXT, CS_ERR_UNKNOWN_SERVICE, iService);

		iPort = (int) ntohs(servp->s_port);

		if ((hostp = gethostbyname(iHost)) == 0)

			CSException::throwCoreError(CS_CONTEXT, CS_ERR_UNKNOWN_HOST, iHost);

		memset(&server, 0, sizeof(server));

		server.sin_family = AF_INET;

		memcpy(&server.sin_addr, hostp->h_addr, (size_t) hostp->h_length);

		server.sin_port = (uint16_t) servp->s_port;

		if (connect(iHandle, (struct sockaddr *) &server, sizeof(server)) == -1)

			throwError(CS_CONTEXT, SOCKET_ERRORNO);

	}

	catch_(a) {

		close();

		throw_();

	}

	cont_(a);

	exit_();

}

void CSSocket::close()

{

	flush();

	if (iHandle != -1) {

		shutdown(iHandle, SHUT_RDWR);

		/* shutdown does not close the socket!!? */

		CLOSE_SOCKET(iHandle);

		iHandle = -1;

	}

	if (iHost) {

		cs_free(iHost);

		iHost = NULL;

	}

	if (iService) {

		cs_free(iService);

		iService = NULL;

	}

	if (iIdentity) {

		cs_free(iIdentity);

		iIdentity = NULL;

	}

	iPort = 0;

}

size_t CSSocket::read(void *data, size_t len)

{

	ssize_t in;

	enter_();

	/* recv, by default will block until at lease one byte is

	 * returned.

	 * So a return of zero means EOF!

	 */

	retry:

	if (iTimeout)

		in = timeoutRead(self, data, len);

	else

		in = recv(iHandle, (char *) data, len, 0);

	self->interrupted();

	if (in == -1) {

		/* Note, we actually ignore all errors on the socket.

		 * If no data was returned by the read so far, then

		 * the error will be considered EOF.

		 */

		int err = SOCKET_ERRORNO;

		if (err == EWOULDBLOCK || err == EINTR)

			goto retry;

		throwError(CS_CONTEXT, err);

		in = 0;

	}

	return_((size_t) in);

}

int CSSocket::read()

{

	int		ch;

	u_char	buffer[1];

	enter_();

	if (read(buffer, 1) == 1)

		ch = buffer[0];

	else

		ch = -1;

	return_(ch);

}

int CSSocket::peek()

{

	return -1;

}

void CSSocket::write(const void *data, size_t len)

{

#ifdef CS_USE_OUTPUT_BUFFER

	if (len <= CS_MIN_WRITE_SIZE) {

		if (iDataLen + len > CS_OUTPUT_BUFFER_SIZE) {

			/* This is the amount of data that will still fit

			 * intp the buffer:

			 */

			size_t tfer = CS_OUTPUT_BUFFER_SIZE - iDataLen;

			memcpy(iOutputBuffer + iDataLen, data, tfer);

			flush();

			len -= tfer;

			memcpy(iOutputBuffer, ((char *) data) + tfer, len);

			iDataLen = len;

		}

		else {

			memcpy(iOutputBuffer + iDataLen, data, len);

			iDataLen += len;

		}

	}

	else {

		/* If the block give is large enough, the

		 * writing directly from the block saves copying the

		 * data to the local output buffer buffer:

		 */

		flush();

		writeBlock(data, len);

	}

#else

	writeBlock(data, len);

#endif

}

void CSSocket::write(char ch)

{

	enter_();

	writeBlock(&ch, 1);

	exit_();

}

void CSSocket::flush()

{

#ifdef CS_USE_OUTPUT_BUFFER

	uint32_t len;

	if ((len = iDataLen)) {

		iDataLen = 0;

		/* Note: we discard the data to be written if an

		 * exception occurs.

		 */

		writeBlock(iOutputBuffer, len);

	}

#endif

}

const char *CSSocket::identify()

{

	enter_();

	if (!iIdentity) {

		char buffer[200];

		formatAddress(200, buffer);

		iIdentity = cs_strdup(buffer);

	}

	return_(iIdentity);

}