/* ************************************************************************** */
/* ************** functions for sending operations to server **************** */
/* ************************************************************************** */

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

#if defined(WAIT_FOR_WRITE)
	void WaitForWrite(int socket)
	{
		fd_set write_sockets;

		FD_ZERO(&write_sockets);
		FD_SET(socket, &write_sockets);

		select(socket + 1, NULL, &write_sockets, NULL, NULL);
	}
#else
	#define WaitForWrite(x)
#endif

#define UNLOCK_AND_RETURN(x, y) { pthread_mutex_unlock(x); return y; }
#define UNLOCK_RAISE_AND_RETURN(x, y, z) { pthread_mutex_unlock(x); raise(y); return z; }

int mps_create(char *mtext, u_long mlength, u_long *message_id_ptr)
{
	thread_semaphore *thread_sem;
	u_long signal_ptr, midptr;
	u_char OpTp;

	*message_id_ptr = 0;

/* prepare variables */
	midptr = (u_long) message_id_ptr;
	OpTp = OpTp_CREATE;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &midptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &mlength, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( mlength != 0 )
	{
		if ( !td_write_char_array(conn_socket, mtext, mlength, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -4)
	}

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}

int mps_destroy(u_long message_id)
{
	thread_semaphore *thread_sem;
	u_long signal_ptr;
	u_char OpTp;

/* prepare variables */
	OpTp = OpTp_DESTROY;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &message_id, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}

int mps_end()
{
	thread_semaphore *thread_sem;
	u_long signal_ptr;
	u_char OpTp;

/* prepare variables */
	OpTp = OpTp_END;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}

int mps_recv(char **mtext_ptr, u_long *mlength_ptr, u_long *message_id_ptr, u_long *process_id_ptr, u_long num_process_ids, u_long *process_ids, char *filter)
{
	thread_semaphore *thread_sem;
	u_long message_ptr, signal_ptr, mlenptr, midptr, pidptr, process_index, flength;
	u_char OpTp;

	*mtext_ptr = NULL;
	*message_id_ptr = 0;

/* prepare variables */
	message_ptr = (u_long) mtext_ptr;
	mlenptr = (u_long) mlength_ptr;
	midptr = (u_long) message_id_ptr;
	pidptr = (u_long) process_id_ptr;
	flength = (filter == NULL) ? (0): (strlen(filter));
	OpTp = OpTp_RECV;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &message_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &mlenptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &midptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &pidptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &num_process_ids, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( num_process_ids != 0 )
	{
		for (process_index = 0; process_index < num_process_ids; process_index++)
		{
			if ( !td_write_u_long(conn_socket, &(process_ids[process_index]), &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -5)
		}
	}

	if ( !td_write_u_long(conn_socket, &flength, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( flength != 0 )
	{
		if ( !td_write_char_array(conn_socket, filter, flength, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -4)
	}

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}

int mps_send(u_long message_id, u_long num_process_ids, u_long *process_ids)
{
	thread_semaphore *thread_sem;
	u_long signal_ptr, process_index;
	u_char OpTp;

/* prepare variables */
	OpTp = OpTp_SEND;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &message_id, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &num_process_ids, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( num_process_ids != 0 )
	{
		for (process_index = 0; process_index < num_process_ids; process_index++)
		{
			if ( !td_write_u_long(conn_socket, &(process_ids[process_index]), &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -5)
		}
	}

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}

int mps_send_async(u_long message_id, u_long num_process_ids, u_long *process_ids)
{
	u_long signal_ptr, process_index;
	u_char OpTp;

/* prepare variables */
	signal_ptr = (u_long) NULL;
	OpTp = OpTp_SEND;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &message_id, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &num_process_ids, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( num_process_ids != 0 )
	{
		for (process_index = 0; process_index < num_process_ids; process_index++)
		{
			if ( !td_write_u_long(conn_socket, &(process_ids[process_index]), &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -5)
		}
	}

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	pthread_mutex_unlock(&send_mutex);

	return exit_client;
}

int mps_start(char **mtext_ptr, u_long *mlength_ptr, u_long *message_id_ptr)
{
	thread_semaphore *thread_sem;
	u_long message_ptr, signal_ptr, mlenptr, midptr;
	u_char OpTp;

	*mtext_ptr = NULL;
	*message_id_ptr = 0;

/* prepare variables */
	message_ptr = (u_long) mtext_ptr;
	mlenptr = (u_long) mlength_ptr;
	midptr = (u_long) message_id_ptr;
	OpTp = OpTp_START;

/* write operation */
	WaitForWrite(conn_socket);

	pthread_mutex_lock(&send_mutex);

	if ( exit_client ) UNLOCK_RAISE_AND_RETURN(&send_mutex, SIGINT, exit_client)

	thread_sem = create_thread_semaphore();
	if (thread_sem == NULL) UNLOCK_AND_RETURN(&send_mutex, -1)

/* prepare variables 2 */
	signal_ptr = (u_long) &(thread_sem->semaphore);

	if ( !td_write_u_char(conn_socket, &OpTp, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -2)

	if ( !td_write_u_long(conn_socket, &signal_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &message_ptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &mlenptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_write_u_long(conn_socket, &midptr, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -3)

	if ( !td_flush(conn_socket, &td_buffer) ) UNLOCK_AND_RETURN(&send_mutex, -6)

	thread_sem->is_waiting = 1;
	pthread_mutex_unlock(&send_mutex);

/* wait for result */
	sem_wait(&(thread_sem->semaphore));

	return exit_client;
}
