/* $Id: messages.c,v 1.63 2004/11/15 20:59:45 graziano Exp $ */


#include "config_portability.h"

#include <stddef.h>
#include <string.h>

#include "diagnostic.h"
#include "protocol.h"
#include "messages.h"
#include "osutil.h"
#include "timeouts.h"

static void *lock = NULL;			/* local mutex */

/*
 * Info on registered listeners.  #message# is the message for which #listener#
 * is registered; #image# the message image.  Note that, since we provide no
 * way to terminate listening for messages, we can simply expand the list by
 * one every time a new listener is registered.
 */
typedef struct {
	MessageType message;
	const char *image;
	ListenFunction listener;
} ListenerInfo;

static ListenerInfo *listeners = NULL;
static unsigned listenerCount = 0;

/*
 * Returns 1 or 0 depending on whether or not format conversion is required for
 * data with the format described by the #howMany#-long array #description#.
 *
 *  I believe is thread safe (DataSize & DifferentFOrmat are thread safe)
 */
static int
ConversionRequired(	const DataDescriptor *description,
			size_t howMany) {
	int i;

	if(DataSize(description, howMany, HOST_FORMAT) !=
			DataSize(description, howMany, NETWORK_FORMAT)) {
		return 1;
	}

	for(i = 0; i < howMany; i++) {
		if(description[i].type == STRUCT_TYPE) {
			if(ConversionRequired(description[i].members, description[i].length)) {
				return 1;
			}
		} else if(DifferentFormat(description[i].type))
			return 1;
	}

	return 0;
}


/* it should be thread safe (all the conversions routines should be
 * thread safe).
 * */
int
RecvData(	Socket sd,
		void *data,
		const DataDescriptor *description,
		size_t howMany,
		double timeOut) {

	void *converted;
	int convertIt;
	void *destination;
	int recvResult;
	size_t totalSize = DataSize(description, howMany, NETWORK_FORMAT);
	double start, recvTO;
	IPAddress addr;

	start = 0;			/* avoid annoying warnings */
	converted = NULL;
	convertIt = ConversionRequired(description, howMany);

	if(convertIt) {
		converted = MALLOC(totalSize);
		if(converted == NULL) {
			FAIL1("RecvData: memory allocation of %d bytes failed\n", totalSize);
		}
		destination = converted;
	} else {
		destination = data;
	}

	/* use adaptive timeouts? */
	if (timeOut < 0) {
		addr = Peer(sd);		/* save one call */
		recvTO = GetTimeOut(RECV, addr, totalSize);
		start = MicroTime();
	} else { 
		/* whatever the user wants */
		recvTO  = timeOut;
	}

	recvResult = RecvBytes(sd, destination, totalSize, recvTO);

	if (timeOut < 0) {
		/* give a feedback for the automatic timeout */
		SetTimeOut(RECV, addr, (MicroTime()-start)/1000000, recvResult, (recvResult != totalSize));
	}

	if (recvResult == totalSize) {
		if(DifferentOrder() || convertIt)
			ConvertData(data, destination, description, 
				howMany, NETWORK_FORMAT);

		if(converted != NULL)
			free(converted);
	}
	return (recvResult == totalSize);
}


/* It should be thread safe (just read headerDescriptor[Lenght] and
 * RecvByte is thread safe) */
int
RecvMessage(Socket sd,
            MessageType message,
            size_t *dataSize,
            double timeOut) {
	char *garbage;
	size_t tmp;
	double when, recvTO;
	MessageHeader header;

	if (!RecvHeader(sd,
			&header,
			timeOut)) {
		FAIL("RecvMessage: no message received\n");
	}

	if(header.message != message) {
		/* we need to clean up: let's try do it within the right
		 * timeout */
		when = CurrentTime() + timeOut;

		garbage = MALLOC(2048);
		if (garbage == NULL) {
			FAIL("RecvMessage: out of memory!");
		}
		while(header.dataSize > 0) {
			/* if we passed the timeout quit the clean up */
			recvTO = when - CurrentTime();
			if (when <= 0) {
				break;
			}

			/* if we time out let's drop the socket */
			if (header.dataSize > sizeof(garbage)) {
				tmp = sizeof(garbage);
			} else {
				tmp = header.dataSize;
			}
			if (RecvBytes(sd, garbage, tmp, recvTO)) {
				DROP_SOCKET(&sd);
				WARN("RecvMessage: timeout on receiving non-handled message: dropping socket\n");
				break;
			}
			header.dataSize -= sizeof(garbage);
		}
		free(garbage);

		FAIL1("RecvMessage: unexpected message %d received\n", header.message);
	}
	*dataSize = header.dataSize;
	return(1);
}

/* it should be thread safe */
int
RecvMessageAndDatas(Socket sd,
                    MessageType message,
                    void *data1,
                    const DataDescriptor *description1,
                    size_t howMany1,
                    void *data2,
                    const DataDescriptor *description2,
                    size_t howMany2,
                    double timeOut) {
	size_t dataSize;

	if (RecvMessage(sd, message, &dataSize, timeOut) != 1) {
		/* failed to receive message: errors already printed out
		 * by RecvMessage() */
		return 0;
	}

	if(data1 != NULL) {
		if(!RecvData(sd, data1, description1, howMany1, timeOut)) {
			FAIL("RecvMessageAndDatas: data receive failed\n");
		}
	}

	if(data2 != NULL) {
		if(!RecvData(sd, data2, description2, howMany2, timeOut)) {
			FAIL("RecvMessageAndDatas: data receive failed\n");
		}
	}

	return(1);
}

/* 
 * waits for timeOut seconds for incoming messages and calls the
 * appropriate (registered) listener function.
 */
int
ListenForMessages(double timeOut) {
	MessageHeader header;
	int i;
	Socket sd;
	char *name;

	if(!IncomingRequest(timeOut, &sd)) {
		return 0;
	}

	/* let's use the adaptive timeouts on receiving the header */
	if (!RecvHeader(	sd,
				&header,
				-1)) {
		/* Likely a connection closed by the other side.  There
		 * doesn't seem to be any reliable way to detect this,
		 * and, for some reason, select() reports it as a
		 * connection ready for reading.  */
		DROP_SOCKET(&sd);
		return 0;
	}

	for(i = 0; i < listenerCount; i++) {
		if(listeners[i].message == header.message) {
			name = PeerName_r(sd);
			if (name != NULL) {
				INFO4("Received %s message from %s:%d on %d\n", listeners[i].image, name, PeerNamePort(sd), sd);
				FREE(name);
			} else {
				INFO2("Received %s message from on %d\n", listeners[i].image, sd);
			}
			listeners[i].listener(&sd, header);
			break;
		}
	}

	if(i == listenerCount) {
		name = PeerName_r(sd);
		if (name != NULL) {
			WARN3("Unknown message %d received from %s on %d\n", header.message, name, sd);
			FREE(name);
		} else {
			WARN2("Unknown message %d received on %d\n", header.message, sd);
		}
		DROP_SOCKET(&sd);
	}

	return 1;
}



/* regsiters the functions which should be called upon the receive of the
 * messageType message. Should be thread safe */
void
RegisterListener(MessageType message,
                 const char *image,
                 ListenFunction listener) {
	int i;

	if (!GetNWSLock(&lock)) {
		ERROR("RegisterListener: couldn't obtain the lock\n");
	}

	/* if there is an old handler, remove it */
	for (i = 0; i < listenerCount; i++) {
		if (listeners[i].message == message) {
			/* found it */
			break;
		}
	}

	/* did we find an old slot? */
	if (i >= listenerCount) {
		/* nope: we need more space */
		listeners = REALLOC(listeners, (listenerCount+1)*sizeof(ListenerInfo));
		if (listener == NULL) {
			ABORT("RegisterListener: out of memory\n");
		}
		i = listenerCount;	/* point to the free slot */
		listenerCount++;	/* we have an extra listener */
	}

	/* we register the listener in slot i */
	listeners[i].message = message;
	listeners[i].image = image;
	listeners[i].listener = listener;
	ReleaseNWSLock(&lock);
}

/* unregister the handler for message #type# */
void
UnregisterListener(MessageType type) {
	int i;

	if (!GetNWSLock(&lock)) {
		ERROR("RegisterListener: couldn't obtain the lock\n");
	}

	/* look if we have it registered */
	for (i = 0; i < listenerCount; i++) {
		if (listeners[i].message == type) {
			/* found it */
			break;
		}
	}

	if (i >= listenerCount) {
		/* we didn't find it */
		LOG1("UnregisterListener: no listener for type %d\n", type);
	} else {
		/* remove the listener */
		for (; i < (listenerCount - 1); i++) {
			listeners[i].message = listeners[i+1].message;
			listeners[i].image = listeners[i+1].image;
			listeners[i].listener = listeners[i+1].listener;
		}

		/* free the extra memory */
		listenerCount--;
		listeners = REALLOC(listeners, (listenerCount)*sizeof(ListenerInfo));
		if (listeners == NULL) {
			ABORT("UnregisterListener: realloc failed\n");
		}
	}

	ReleaseNWSLock(&lock);
}


/* it should be thread safe (Convert*, SendBytes, DataSize abd
 * DifferentOrder are thread safe) */
int
SendData(	Socket sd,
		const void *data,
		const DataDescriptor *description,
		size_t howMany,
		double timeOut) {
	void *converted;
	int sendResult;
	const void *source;
	size_t totalSize;
	double start, sendTO;
	IPAddress addr;

	converted = NULL;
	start = 0;
	totalSize = DataSize(description, howMany, NETWORK_FORMAT);

	if(DifferentOrder() || ConversionRequired(description, howMany)) {
		converted = MALLOC(totalSize);
		if(converted == NULL) {
			FAIL("SendData: memory allocation failed\n");
		}
		/* make memory checkers happy */
		memset(converted, 0, totalSize);

		ConvertData(converted, data, description, howMany, HOST_FORMAT);
		source = converted;
	} else {
		source = data;
	}

	/* use adaptive timeouts? */
	if (timeOut < 0) {
		addr = Peer(sd);		/* save one call */
		sendTO = GetTimeOut(SEND, addr, totalSize);
		start = MicroTime();
	} else {
		/* whatever the user wants */
		sendTO = timeOut;
	}

	sendResult = SendBytes(sd, source, totalSize, sendTO);

	if (timeOut < 0) {
		/* give a feedback for the automatic timeout */
		SetTimeOut(SEND, addr, (MicroTime()-start)/1000000, sendResult, (sendResult != totalSize));
	}

	FREE(converted);

	return (sendResult == totalSize);
}

/* it should be thread safe (SendData, DataSize are thread safe) */
int
SendMessageAndDatas(Socket sd,
                    MessageType message,
                    const void *data1,
                    const DataDescriptor *description1,
                    size_t howMany1,
                    const void *data2,
                    const DataDescriptor *description2,
                    size_t howMany2,
                    double timeOut) {
	MessageHeader header;

	header.version = NWS_VERSION;
	header.message = message;
	header.dataSize = 0;
	if(data1 != NULL)
		header.dataSize += DataSize(description1, howMany1, NETWORK_FORMAT);
	if(data2 != NULL)
		header.dataSize += DataSize(description2, howMany2, NETWORK_FORMAT);

	if(!SendData(sd,
			&header,
			headerDescriptor,
			headerDescriptorLength,
			timeOut)) {
		FAIL("SendMessageAndDatas: header send failed \n");
	}
	if((data1 != NULL) && !SendData(sd, data1, description1, howMany1, timeOut)) {
		FAIL("SendMessageAndDatas: data1 send failed\n");
	}
	if((data2 != NULL) && !SendData(sd, data2, description2, howMany2, timeOut)) {
		FAIL("SendMessageAndDatas: data2 send failed\n");
	}
	return 1;
}

/*
 * reads the NWS header associated with in incoming message and returns
 * it in #header#.  returns 0 if the read fails, 1 otherwise.
 *
 * it should be thread safe (RecvData is thread safe and header* are only
 * read) 
 */
int RecvHeader(	Socket sd, 
		MessageHeader *h, 
		double tout) {

	/* sanity check */
	if (sd == NO_SOCKET || h == NULL) {
		ERROR("RecvHeader: wrong parameters\n");
		return 0;
	}

	return RecvData(sd, 
			(void *)h, 
			headerDescriptor, 
			headerDescriptorLength, 
			tout);
}