/* ************************************************************************** */
/* ************** thread for receiving messages from server ***************** */
/* ************************************************************************** */

#include <malloc.h>
#include "MPSClient_defs.h"
#include "MPSClient_vars.h"

int ReadCreateRet()
{
	u_long signal_ptr, midptr, message_id;
	int ret;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &midptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &message_id) ) return -1;

/* fill operation variables */
	if ( midptr != 0 ) *( (u_long *) midptr) = message_id;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

int ReadDestroyRet()
{
	u_long signal_ptr;
	int ret;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

int ReadEndRet()
{
	u_long signal_ptr;
	int ret;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

int ReadRecvRet()
{
	u_long signal_ptr, mlenptr, midptr, pidptr, message_id, process_id, message_ptr, mlength;
	char *mtext;
	int ret;

	mtext = NULL;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &message_ptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &mlenptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &midptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &pidptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &message_id) ) return -1;

	if ( !td_read_u_long(conn_socket, &process_id) ) return -1;

	if ( !td_read_u_long(conn_socket, &mlength) ) return -1;

	if ( mlength != 0)
	{
		mtext = (char *) malloc(mlength);

		if ( mtext == NULL )
		{
			return -3;
		}

		if ( !td_read_char_array(conn_socket, mtext, mlength) )
		{
			free(mtext);
			return -4;
		}
	}

/* fill operation variables */
	if ( mlenptr != 0 ) *( (u_long *) mlenptr) = mlength;
	if ( midptr != 0 ) *( (u_long *) midptr) = message_id;
	if ( pidptr != 0 ) *( (u_long *) pidptr) = process_id;

	*( (char **) message_ptr) = mtext;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

int ReadSendRet()
{
	u_long signal_ptr;
	int ret;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

int ReadStartRet()
{
	u_long signal_ptr, mlenptr, midptr, message_id, message_ptr, mlength;
	char *mtext;
	int ret;

	mtext = NULL;

/* read operation result */
	if ( !td_read_u_long(conn_socket, &signal_ptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &message_ptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &mlenptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &midptr) ) return -1;

	if ( !td_read_u_long(conn_socket, &message_id) ) return -1;

	if ( !td_read_u_long(conn_socket, &mlength) ) return -1;

	if ( mlength != 0)
	{
		mtext = (char *) malloc(mlength);

		if ( mtext == NULL )
		{
			return -3;
		}

		if ( !td_read_string(conn_socket, mtext, mlength) )
		{
			free(mtext);
			return -4;
		}
	}

/* fill operation variables */
	if ( mlenptr != 0 ) *( (u_long *) mlenptr) = mlength;
	if ( midptr != 0 ) *( (u_long *) midptr) = message_id;

	*( (char **) message_ptr) = mtext;

/* signal semaphore */
	if ( signal_ptr != 0 )
	{
		ret = post_thread_semaphore( (sem_t *) signal_ptr );
		if (ret) return -2;
	}

	return 0;
}

void *client_thread(void *data)
{
	int select_return, socket_error;
	struct timeval select_timeout;
	fd_set read_sockets, except_sockets;
	u_char OpTp;


/* wait for initialization of data */
	pthread_mutex_lock(&thread_mutex);
	pthread_mutex_unlock(&thread_mutex);

	while ( !exit_client )
	{
		select_timeout.tv_sec = 0;
		select_timeout.tv_usec = 100000;
		FD_ZERO(&read_sockets);

		FD_SET(conn_socket, &read_sockets);

		except_sockets = read_sockets;

		select_return = select(conn_socket + 1, &read_sockets, NULL, &except_sockets, &select_timeout);
		if ( select_return == -1)
		{
			pthread_mutex_lock(&send_mutex);

			exit_client = 1;

			pthread_mutex_unlock(&send_mutex);
/*			raise(SIGINT);*/
			post_waiting_thread_semaphores();
			break;
		}
		if ( select_return == 0) continue;


		socket_error = 1;
/* if there is data to read, then read and execute operation*/
		if ( !FD_ISSET(conn_socket, &except_sockets) )
		{
			if ( td_read_u_char(conn_socket, &OpTp) )
			{
				switch (OpTp)
				{
					case OpTp_END_RET:
						if ( ReadEndRet() ) break;

						socket_error = 0;
						break;
					case OpTp_START_RET:
						if ( ReadStartRet() ) break;

						socket_error = 0;
						break;
					case OpTp_CREATE_RET:
						if ( ReadCreateRet() ) break;

						socket_error = 0;
						break;
					case OpTp_DESTROY_RET:
						if ( ReadDestroyRet() ) break;

						socket_error = 0;
						break;
					case OpTp_SEND_RET:
						if ( ReadSendRet() ) break;

						socket_error = 0;
						break;
					case OpTp_RECV_RET:
						if ( ReadRecvRet() ) break;

						socket_error = 0;
						break;
				}
			}
		}

/* if error then disconnect client */
		if ( socket_error != 0 )
		{
			pthread_mutex_lock(&send_mutex);

			exit_client = 1;

			pthread_mutex_unlock(&send_mutex);
/*			raise(SIGINT);*/
			post_waiting_thread_semaphores();
			break;
		}


	}

	pthread_mutex_lock(&thread_mutex);

	thread_created = 0;

	if ( exit_semaphore_waiting )
	{
		sem_post(&exit_semaphore);
	}
	pthread_mutex_unlock(&thread_mutex);

	pthread_exit(NULL);

/* to avoid warning */
	return NULL;
}

