libApp/src/transport/tcp.c

/* $Id$ */
/* Copyright (c) 2012-2020 Pierre Pronchery <khorben@defora.org> */
/* This file is part of DeforaOS System libApp */
/* 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, version 3 of the License.
 *
 * 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, see <http://www.gnu.org/licenses/>. */
/* FIXME:
 * - added status reports as message callbacks (connecting, connected...) */



#include <sys/socket.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#ifdef DEBUG
# include <stdio.h>
#endif
#include <string.h>
#include <limits.h>
#include <errno.h>
#ifdef __WIN32__
# include <Winsock2.h>
#else
# include <netinet/in.h>
# include <arpa/inet.h>
# include <netdb.h>
#endif
#include <System.h>
#include "App/appmessage.h"
#include "App/apptransport.h"

/* portability */
#ifdef __WIN32__
# define close(fd) closesocket(fd)
#endif

/* for tcp4 and tcp6 */
#ifndef TCP_DOMAIN
# define TCP_DOMAIN AF_UNSPEC
#endif


/* TCP */
/* private */
/* types */
typedef struct _AppTransportPlugin TCP;

typedef struct _TCPSocket
{
	TCP * tcp;
	AppTransportClient * client;

	int fd;
	struct sockaddr * sa;
	socklen_t sa_len;

	/* input queue */
	char * bufin;
	size_t bufin_cnt;
	/* output queue */
	char * bufout;
	size_t bufout_cnt;
} TCPSocket;

struct _AppTransportPlugin
{
	AppTransportPluginHelper * helper;
	AppTransportMode mode;

	struct addrinfo * ai;
	struct addrinfo * aip;

	union
	{
		struct
		{
			/* for servers */
			int fd;
			TCPSocket ** clients;
			size_t clients_cnt;
		} server;

		/* for clients */
		TCPSocket client;
	} u;
};


/* constants */
#define INC 1024

#include "common.h"
#include "common.c"


/* protected */
/* prototypes */
/* plug-in */
static TCP * _tcp_init(AppTransportPluginHelper * helper, AppTransportMode mode,
		char const * name);
static void _tcp_destroy(TCP * tcp);

static int _tcp_client_send(TCP * tcp, AppMessage * message);
static int _tcp_server_send(TCP * tcp, AppTransportClient * client,
		AppMessage * message);

/* useful */
static int _tcp_error(char const * message);

/* servers */
static int _tcp_server_add_client(TCP * tcp, TCPSocket * client);
static int _tcp_server_remove_client(TCP * tcp, TCPSocket * client);

/* sockets */
static int _tcp_socket_init(TCPSocket * tcpsocket, int domain, int flags,
		TCP * tcp);
static void _tcp_socket_init_fd(TCPSocket * tcpsocket, TCP * tcp, int fd,
		struct sockaddr * sa, socklen_t sa_len);
static TCPSocket * _tcp_socket_new_fd(TCP * tcp, int fd, struct sockaddr * sa,
		socklen_t sa_len);
static void _tcp_socket_delete(TCPSocket * tcpsocket);
static void _tcp_socket_destroy(TCPSocket * tcpsocket);

static int _tcp_socket_queue(TCPSocket * tcpsocket, Buffer * buffer);

/* callbacks */
static int _tcp_callback_accept(int fd, TCP * tcp);
static int _tcp_callback_connect(int fd, TCP * tcp);
static int _tcp_socket_callback_read(int fd, TCPSocket * tcpsocket);
static int _tcp_socket_callback_write(int fd, TCPSocket * tcpsocket);


/* public */
/* constants */
/* plug-in */
AppTransportPluginDefinition transport =
{
	"TCP",
	NULL,
	_tcp_init,
	_tcp_destroy,
	_tcp_client_send,
	_tcp_server_send
};


/* protected */
/* functions */
/* plug-in */
/* tcp_init */
static int _init_client(TCP * tcp, char const * name, int domain);
static int _init_server(TCP * tcp, char const * name, int domain);

static TCP * _tcp_init(AppTransportPluginHelper * helper, AppTransportMode mode,
		char const * name)
{
	TCP * tcp;
	int res;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%u, \"%s\")\n", __func__, mode, name);
#endif
	if((tcp = object_new(sizeof(*tcp))) == NULL)
		return NULL;
	memset(tcp, 0, sizeof(*tcp));
	tcp->helper = helper;
	switch((tcp->mode = mode))
	{
		case ATM_CLIENT:
			res = _init_client(tcp, name, TCP_DOMAIN);
			break;
		case ATM_SERVER:
			res = _init_server(tcp, name, TCP_DOMAIN);
			break;
		default:
			res = -error_set_code(-EINVAL, "%s",
					"Unknown transport mode");
			break;
	}
	/* check for errors */
	if(res != 0)
	{
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() => %d (%s)\n", __func__, res,
				error_get(NULL));
#endif
		_tcp_destroy(tcp);
		return NULL;
	}
	return tcp;
}

static int _init_client(TCP * tcp, char const * name, int domain)
{
#ifdef DEBUG
	struct sockaddr_in * sa;
#endif

	tcp->u.client.tcp = tcp;
	tcp->u.client.fd = -1;
	/* obtain the remote address */
	if((tcp->ai = _init_address(name, domain, 0)) == NULL)
		return -1;
	/* connect to the remote host */
	for(tcp->aip = tcp->ai; tcp->aip != NULL; tcp->aip = tcp->aip->ai_next)
	{
		tcp->u.client.fd = -1;
		/* initialize the client socket */
		if(_tcp_socket_init(&tcp->u.client, tcp->aip->ai_family,
					O_NONBLOCK, tcp) != 0)
			continue;
#ifdef DEBUG
		if(tcp->aip->ai_family == AF_INET)
		{
			sa = (struct sockaddr_in *)tcp->aip->ai_addr;
			fprintf(stderr, "DEBUG: %s() connect(%s:%u)\n", __func__,
					inet_ntoa(sa->sin_addr),
					ntohs(sa->sin_port));
		}
		else
			fprintf(stderr, "DEBUG: %s() connect(%d)\n", __func__,
					tcp->aip->ai_family);
#endif
		if(connect(tcp->u.client.fd, tcp->aip->ai_addr,
					tcp->aip->ai_addrlen) != 0)
		{
			if(errno != EINPROGRESS)
			{
				close(tcp->u.client.fd);
				tcp->u.client.fd = -1;
				_tcp_error("socket");
				continue;
			}
			event_register_io_write(tcp->helper->event,
					tcp->u.client.fd,
					(EventIOFunc)_tcp_callback_connect,
					tcp);
			event_loop(tcp->helper->event);
		}
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() connect() => %d\n", __func__,
				tcp->u.client.fd);
#endif
		if(tcp->u.client.fd >= 0)
			break;
	}
	if(tcp->aip == NULL)
		return -1;
	/* listen for any incoming message */
	event_register_io_read(tcp->helper->event, tcp->u.client.fd,
			(EventIOFunc)_tcp_socket_callback_read, &tcp->u.client);
	/* write pending messages if any */
	if(tcp->u.client.bufout_cnt > 0)
	{
		event_register_io_write(tcp->helper->event, tcp->u.client.fd,
				(EventIOFunc)_tcp_socket_callback_write,
				&tcp->u.client);
		event_loop(tcp->helper->event);
	}
	return 0;
}

static int _init_server(TCP * tcp, char const * name, int domain)
{
	TCPSocket tcpsocket;
#ifdef DEBUG
	struct sockaddr_in * sa;
#endif

	tcp->u.server.fd = -1;
	/* obtain the local address */
	if((tcp->ai = _init_address(name, domain, AI_PASSIVE)) == NULL)
		return -1;
	for(tcp->aip = tcp->ai; tcp->aip != NULL; tcp->aip = tcp->aip->ai_next)
	{
		/* create the socket */
		if(_tcp_socket_init(&tcpsocket, tcp->aip->ai_family, O_NONBLOCK,
					tcp) != 0)
			continue;
		/* XXX ugly */
		tcp->u.server.fd = tcpsocket.fd;
		/* accept incoming connections */
#ifdef DEBUG
		if(tcp->aip->ai_family == AF_INET)
		{
			sa = (struct sockaddr_in *)tcp->aip->ai_addr;
			fprintf(stderr, "DEBUG: %s() %s (%s:%u)\n", __func__,
					"bind()", inet_ntoa(sa->sin_addr),
					ntohs(sa->sin_port));
		}
		else
			fprintf(stderr, "DEBUG: %s() %s %d\n", __func__,
					"bind()", tcp->aip->ai_family);
#endif
		if(bind(tcp->u.server.fd, tcp->aip->ai_addr,
					tcp->aip->ai_addrlen) != 0)
		{
			_tcp_error("bind");
			close(tcp->u.server.fd);
			tcp->u.server.fd = -1;
			continue;
		}
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() %s\n", __func__, "listen()");
#endif
		if(listen(tcp->u.server.fd, SOMAXCONN) != 0)
		{
			_tcp_error("listen");
			close(tcp->u.server.fd);
			tcp->u.server.fd = -1;
			continue;
		}
		event_register_io_read(tcp->helper->event, tcp->u.server.fd,
				(EventIOFunc)_tcp_callback_accept, tcp);
		break;
	}
	return (tcp->aip != NULL) ? 0 : -1;
}


/* tcp_destroy */
static void _destroy_client(TCP * tcp);
static void _destroy_server(TCP * tcp);

static void _tcp_destroy(TCP * tcp)
{
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s()\n", __func__);
#endif
	switch(tcp->mode)
	{
		case ATM_CLIENT:
			_destroy_client(tcp);
			break;
		case ATM_SERVER:
			_destroy_server(tcp);
			break;
	}
	if(tcp->ai != NULL)
		freeaddrinfo(tcp->ai);
	object_delete(tcp);
}

static void _destroy_client(TCP * tcp)
{
	_tcp_socket_destroy(&tcp->u.client);
}

static void _destroy_server(TCP * tcp)
{
	size_t i;

	for(i = 0; i < tcp->u.server.clients_cnt; i++)
		_tcp_socket_delete(tcp->u.server.clients[i]);
	free(tcp->u.server.clients);
	if(tcp->u.server.fd >= 0)
		close(tcp->u.server.fd);
}


/* tcp_client_send */
static int _tcp_client_send(TCP * tcp, AppMessage * message)
{
	int ret;
	Buffer * buffer;

	if(tcp->mode != ATM_CLIENT)
		return -error_set_code(1, "%s", "Not a client");
	/* send the message */
	if((buffer = buffer_new(0, NULL)) == NULL)
		return -1;
	if((ret = appmessage_serialize(message, buffer)) == 0
			&& (ret = _tcp_socket_queue(&tcp->u.client,
					buffer)) == 0)
		event_loop(tcp->helper->event);
	buffer_delete(buffer);
	return ret;
}


/* tcp_server_send */
static int _tcp_server_send(TCP * tcp, AppTransportClient * client,
		AppMessage * message)
{
	size_t i;
	TCPSocket * s;
	Buffer * buffer;

	if(tcp->mode != ATM_SERVER)
		return -error_set_code(1, "%s", "Not a server");
	/* lookup the client */
	for(i = 0; i < tcp->u.server.clients_cnt; i++)
	{
		s = tcp->u.server.clients[i];
		if(s->client == client)
			break;
	}
	if(i == tcp->u.server.clients_cnt)
		return -error_set_code(1, "%s", "Unknown client");
	/* send the message */
	if((buffer = buffer_new(0, NULL)) == NULL)
		return -1;
	if(appmessage_serialize(message, buffer) != 0
			|| _tcp_socket_queue(s, buffer) != 0)
	{
		buffer_delete(buffer);
		return -1;
	}
	return 0;
}


/* useful */
/* tcp_error */
static int _tcp_error(char const * message)
{
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(\"%s\")\n", __func__, message);
#endif
	return error_set_code(-errno, "%s%s%s",
			(message != NULL) ? message : "",
			(message != NULL) ? ": " : "", strerror(errno));
}


/* servers */
/* tcp_server_add_client */
static int _tcp_server_add_client(TCP * tcp, TCPSocket * client)
{
	TCPSocket ** p;
#ifndef NI_MAXHOST
# define NI_MAXHOST 256
#endif
	char host[NI_MAXHOST];
	char const * name = host;
	const int flags = NI_NUMERICSERV;

	if((p = realloc(tcp->u.server.clients, sizeof(*p)
					* (tcp->u.server.clients_cnt + 1)))
			== NULL)
		return -1;
	tcp->u.server.clients = p;
	/* XXX may not be instant */
	if(getnameinfo(client->sa, client->sa_len, host, sizeof(host), NULL, 0,
				NI_NAMEREQD | flags) != 0
			&& getnameinfo(client->sa, client->sa_len, host,
				sizeof(host), NULL, 0, NI_NUMERICHOST | flags)
			!= 0)
		name = NULL;
	if((client->client = tcp->helper->client_new(tcp->helper->transport,
					name)) == NULL)
		return -1;
	tcp->u.server.clients[tcp->u.server.clients_cnt++] = client;
	return 0;
}


/* tcp_server_remove_client */
static int _tcp_server_remove_client(TCP * tcp, TCPSocket * client)
{
	size_t i;

	for(i = 0; i < tcp->u.server.clients_cnt; i++)
		if(tcp->u.server.clients[i] == client)
			break;
	if(i == tcp->u.server.clients_cnt)
		return -1;
	tcp->u.server.clients[i] = NULL;
	_tcp_socket_delete(client);
	for(i = tcp->u.server.clients_cnt;
			i > 0 && tcp->u.server.clients[i - 1] == NULL;)
		tcp->u.server.clients_cnt = --i;
	return 0;
}


/* sockets */
/* tcp_socket_init */
static int _tcp_socket_init(TCPSocket * tcpsocket, int domain, int flags,
		TCP * tcp)
{
	int f;

	if((tcpsocket->fd = socket(domain, SOCK_STREAM, 0)) < 0)
		return -_tcp_error("socket");
	_tcp_socket_init_fd(tcpsocket, tcp, tcpsocket->fd, NULL, 0);
	/* set the socket flags */
	if(flags != 0)
	{
		if((f = fcntl(tcpsocket->fd, F_GETFL)) == -1)
			return -_tcp_error("fcntl");
		if((f & flags) != flags)
			if(fcntl(tcpsocket->fd, F_SETFL, f | flags) == -1)
				return -_tcp_error("fcntl");
	}
#ifdef TCP_NODELAY
	/* do not wait before sending any traffic */
	f = 1;
	setsockopt(fd, SOL_SOCKET, TCP_NODELAY, &f, sizeof(f));
#endif
	return 0;
}


/* tcp_socket_init_fd */
static void _tcp_socket_init_fd(TCPSocket * tcpsocket, TCP * tcp, int fd,
		struct sockaddr * sa, socklen_t sa_len)
{
	tcpsocket->tcp = tcp;
	tcpsocket->client = NULL;
	tcpsocket->fd = fd;
	tcpsocket->sa = sa;
	tcpsocket->sa_len = sa_len;
	tcpsocket->bufin = NULL;
	tcpsocket->bufin_cnt = 0;
	tcpsocket->bufout = NULL;
	tcpsocket->bufout_cnt = 0;
}


/* tcp_socket_new_fd */
static TCPSocket * _tcp_socket_new_fd(TCP * tcp, int fd, struct sockaddr * sa,
		socklen_t sa_len)
{
	TCPSocket * tcpsocket;

	if((tcpsocket = object_new(sizeof(*tcpsocket))) == NULL)
		return NULL;
	_tcp_socket_init_fd(tcpsocket, tcp, fd, sa, sa_len);
	return tcpsocket;
}


/* tcp_socket_delete */
static void _tcp_socket_delete(TCPSocket * tcpsocket)
{
	_tcp_socket_destroy(tcpsocket);
	object_delete(tcpsocket);
}


/* tcp_socket_destroy */
static void _tcp_socket_destroy(TCPSocket * tcpsocket)
{
	TCP * tcp = tcpsocket->tcp;
	AppTransportPluginHelper * helper = tcp->helper;

	helper->client_delete(helper->transport, tcpsocket->client);
	free(tcpsocket->sa);
	if(tcpsocket->fd >= 0)
	{
		event_unregister_io_read(tcpsocket->tcp->helper->event,
				tcpsocket->fd);
		event_unregister_io_write(tcpsocket->tcp->helper->event,
				tcpsocket->fd);
		close(tcpsocket->fd);
	}
	free(tcpsocket->bufin);
	free(tcpsocket->bufout);
}


/* tcp_socket_queue */
static int _tcp_socket_queue(TCPSocket * tcpsocket, Buffer * buffer)
{
	uint32_t len;
	char * p;
	Variable * v;
	Buffer * b = NULL;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, tcpsocket->fd);
#endif
	/* serialize the buffer */
	v = variable_new(VT_BUFFER, buffer);
	b = buffer_new(0, NULL);
	if(v == NULL || b == NULL || variable_serialize(v, b, 0) != 0)
	{
		if(v != NULL)
			variable_delete(v);
		if(b != NULL)
			buffer_delete(b);
		return -1;
	}
	variable_delete(v);
	len = buffer_get_size(b);
	/* FIXME queue the serialized buffer directly as a message instead */
	if((p = realloc(tcpsocket->bufout, tcpsocket->bufout_cnt + len))
			== NULL)
	{
		buffer_delete(b);
		return -1;
	}
	tcpsocket->bufout = p;
	memcpy(&p[tcpsocket->bufout_cnt], buffer_get_data(b), len);
	/* register the callback if necessary */
	if(tcpsocket->bufout_cnt == 0)
		event_register_io_write(tcpsocket->tcp->helper->event,
				tcpsocket->fd,
				(EventIOFunc)_tcp_socket_callback_write,
				tcpsocket);
	tcpsocket->bufout_cnt += len;
	buffer_delete(b);
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d) => %d\n", __func__, tcpsocket->fd, 0);
#endif
	return 0;
}


/* callbacks */
/* tcp_callback_accept */
static int _accept_client(TCP * tcp, int fd, struct sockaddr * sa,
		socklen_t sa_len);

static int _tcp_callback_accept(int fd, TCP * tcp)
{
	struct sockaddr * sa;
	socklen_t sa_len = tcp->aip->ai_addrlen;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, fd);
#endif
	/* check parameters */
	if(tcp->u.server.fd != fd)
		return -1;
	if((sa = malloc(sa_len)) == NULL)
		/* XXX this may not be enough to recover */
		sa_len = 0;
	if((fd = accept(fd, sa, &sa_len)) < 0)
	{
		free(sa);
		return _tcp_error("accept");
	}
	if(_accept_client(tcp, fd, sa, sa_len) != 0)
	{
		/* just close the connection and keep serving */
		/* FIXME report error */
		close(fd);
		free(sa);
	}
#ifdef DEBUG
	else
		fprintf(stderr, "DEBUG: %s() %d\n", __func__, fd);
#endif
	return 0;
}

static int _accept_client(TCP * tcp, int fd, struct sockaddr * sa,
		socklen_t sa_len)
{
	TCPSocket * tcpsocket;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, fd);
#endif
	if((tcpsocket = _tcp_socket_new_fd(tcp, fd, sa, sa_len)) == NULL)
		return -1;
	if(_tcp_server_add_client(tcp, tcpsocket) != 0)
	{
		/* XXX workaround for a double-close() */
		tcpsocket->fd = -1;
		_tcp_socket_delete(tcpsocket);
		return -1;
	}
	event_register_io_read(tcp->helper->event, tcpsocket->fd,
			(EventIOFunc)_tcp_socket_callback_read, tcpsocket);
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d) => 0\n", __func__, fd);
#endif
	return 0;
}


/* tcp_callback_connect */
static int _tcp_callback_connect(int fd, TCP * tcp)
{
	int ret;
	socklen_t s = sizeof(ret);

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, fd);
#endif
	/* check parameters */
	if(tcp->u.client.fd != fd)
		return -1;
	/* obtain the connection status */
	if(getsockopt(fd, SOL_SOCKET, SO_ERROR, &ret, &s) != 0)
	{
		error_set_code(-errno, "%s: %s", "getsockopt", strerror(errno));
		close(fd);
		tcp->u.client.fd = -1;
		/* FIXME report error */
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() %s\n", __func__, strerror(errno));
#endif
		ret = -1;
	}
	else if(ret != 0)
	{
		/* the connection failed */
		error_set_code(-ret, "%s: %s", "connect", strerror(ret));
		close(fd);
		tcp->u.client.fd = -1;
		/* FIXME report error */
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() %s\n", __func__, strerror(ret));
#endif
		ret = -1;
	}
	else
		/* deregister this callback */
		ret = 1;
	if(tcp->mode == ATM_CLIENT)
		event_loop_quit(tcp->helper->event);
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s() => %d\n", __func__, ret);
#endif
	return ret;
}


/* tcp_socket_callback_read */
static int _socket_callback_message(TCPSocket * tcpsocket,
		AppMessage ** message);
static void _socket_callback_read_client(TCPSocket * tcpsocket,
		AppMessage * message);
static void _socket_callback_read_server(TCPSocket * tcpsocket,
		AppMessage * message);
static int _socket_callback_recv(TCPSocket * tcpsocket);

static int _tcp_socket_callback_read(int fd, TCPSocket * tcpsocket)
{
	AppMessage * message;
	int res;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, fd);
#endif
	/* check parameters */
	if(tcpsocket->fd != fd)
		return -1;
	if(_socket_callback_recv(tcpsocket) != 0)
		return -1;
	while((res = _socket_callback_message(tcpsocket, &message)) > 0)
	{
		switch(tcpsocket->tcp->mode)
		{
			case ATM_CLIENT:
				_socket_callback_read_client(tcpsocket,
						message);
				break;
			case ATM_SERVER:
				_socket_callback_read_server(tcpsocket,
						message);
				break;
		}
		appmessage_delete(message);
	}
	if(res < 0)
	{
		_tcp_server_remove_client(tcpsocket->tcp, tcpsocket);
		return -1;
	}
	return 0;
}

static int _socket_callback_message(TCPSocket * tcpsocket,
		AppMessage ** message)
{
	int ret = 0;
	size_t size;
	Variable * variable;
	Buffer * buffer;
	uint32_t u32;

	/* XXX assumes the serialized size is sizeof(u32) */
	if((size = tcpsocket->bufin_cnt) < sizeof(u32))
		return 0;
	/* obtain the message size */
	if((variable = variable_new_deserialize_type(VT_UINT32, &size,
					tcpsocket->bufin)) == NULL)
		return -1;
	variable_get_as(variable, VT_UINT32, &u32);
	variable_delete(variable);
	/* verify the message size */
	if(u32 > APPMESSAGE_MAX_SIZE)
		return -error_set_code(E2BIG, "Message too big (%#x)", u32);
	/* XXX assumes the serialized size is u32 */
	if(u32 > tcpsocket->bufin_cnt)
		return 0;
	/* deserialize the data as a buffer (containing a message) */
	size = tcpsocket->bufin_cnt;
	if((variable = variable_new_deserialize_type(VT_BUFFER, &size,
					tcpsocket->bufin)) == NULL)
		return -1;
	tcpsocket->bufin_cnt -= size;
	memmove(tcpsocket->bufin, &tcpsocket->bufin[size],
			tcpsocket->bufin_cnt);
	if((variable_get_as(variable, VT_BUFFER, &buffer)) == 0)
	{
		*message = appmessage_new_deserialize(buffer);
		buffer_delete(buffer);
		ret = (*message != NULL) ? 1 : -1;
	}
	variable_delete(variable);
	return ret;
}

static void _socket_callback_read_client(TCPSocket * tcpsocket,
		AppMessage * message)
{
	AppTransportPluginHelper * helper = tcpsocket->tcp->helper;

	helper->receive(helper->transport, message);
}

static void _socket_callback_read_server(TCPSocket * tcpsocket,
		AppMessage * message)
{
	AppTransportPluginHelper * helper = tcpsocket->tcp->helper;

	helper->client_receive(helper->transport, tcpsocket->client,
			message);
}

static int _socket_callback_recv(TCPSocket * tcpsocket)
{
	const size_t inc = INC;
	ssize_t ssize;
	char * p;

	if((p = realloc(tcpsocket->bufin, tcpsocket->bufin_cnt + inc)) == NULL)
		return -1;
	tcpsocket->bufin = p;
	if((ssize = recv(tcpsocket->fd, &tcpsocket->bufin[tcpsocket->bufin_cnt],
					inc, 0)) < 0)
	{
		error_set_code(-errno, "%s", strerror(errno));
		close(tcpsocket->fd);
		tcpsocket->fd = -1;
		/* FIXME report error */
		return -1;
	}
	else if(ssize == 0)
	{
#ifdef DEBUG
		fprintf(stderr, "DEBUG: %s() recv() => %ld\n", __func__, ssize);
#endif
		close(tcpsocket->fd);
		tcpsocket->fd = -1;
		/* FIXME report transfer clean shutdown */
		return -1;
	}
	tcpsocket->bufin_cnt += ssize;
	return 0;
}


/* tcp_socket_callback_write */
static int _tcp_socket_callback_write(int fd, TCPSocket * tcpsocket)
{
	ssize_t ssize;

#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s(%d)\n", __func__, fd);
#endif
	/* check parameters */
	if(tcpsocket->fd != fd)
		return -1;
	if((ssize = send(tcpsocket->fd, tcpsocket->bufout,
					tcpsocket->bufout_cnt, 0)) < 0)
	{
		/* XXX report error (and reconnect) */
		error_set_code(-errno, "%s", strerror(errno));
		close(tcpsocket->fd);
		tcpsocket->fd = -1;
		return -1;
	}
	else if(ssize == 0)
	{
		close(tcpsocket->fd);
		tcpsocket->fd = -1;
		/* XXX report transfer interruption (and reconnect) */
		return -error_set_code(-errno, "%s", strerror(errno));
	}
#ifdef DEBUG
	fprintf(stderr, "DEBUG: %s() send() => %ld\n", __func__, ssize);
#endif
	/* XXX use a sliding cursor instead (and then queue the next message) */
	memmove(tcpsocket->bufout, &tcpsocket->bufout[ssize],
			tcpsocket->bufout_cnt - ssize);
	tcpsocket->bufout_cnt -= ssize;
	/* unregister the callback if there is nothing left to write */
	if(tcpsocket->bufout_cnt == 0)
	{
		if(tcpsocket->tcp->mode == ATM_CLIENT)
			event_loop_quit(tcpsocket->tcp->helper->event);
		return 1;
	}
	return 0;
}