sysdis-project/central-manager/lib/local/handler.c

/**************************
*   Handler Dependency    *
***************************
* Designed & Developed by *
*     Adrien Marquès      *
*     <xdrm-brackets>     *
***************************
*   doowap31@gmail.com    *
**************************/
#include "handler.h"






/* Gestion d'une connexion PLANE
*
* @THREADABLE_SOCKET<void*>			SOCKET de la connexion client
*
* @history
* [1] Initialisation des variables
* @loop
*   (2) Attente de requête
*   (3) Gestion de la requête
*   (4) Envoi de la réponse
* [5] On notifie que l'avion est déconnecté (crash)
* [6] On libère la mémoire
* [7] Fermeture de la connection (SOCKET)
* [n] Arrêt du THREAD
* 	1. On met à jour "activeManagers"
* 	2. On arrête le THREAD
*
*/
void* managePlane(void* THREADABLE_ARGS){

	/* [1] Initialisation des variables
	=========================================================*/
	/* 1. Variables utiles */
	int          read, i, index, pindex; // compteurs
	char         buffer[MAX_BUF_LEN];    // buffer
	struct plane data;                   // données de l'avion
	int          SOCKET;                 // Copie de la socket (évite les conflits de références)
	char         tmpFlags;


	/* 2. On récupère les arguments */
	struct handler_arg* arg = THREADABLE_ARGS;
	memcpy(&SOCKET, &arg->socket, sizeof(int));
	if( DEBUGMOD&HDR ) printf("===== managePlane(%p, %p, %d, %p) =====\n", (void*) arg->TCPManagers, (void*) arg->activeTCPManagers, arg->socket, (void*) arg->sgca);


	/* 3. On récupère le rang du thread parmi les "managers" */
	index = -1;
	for( i = 0 ; i < MAX_TCP_THR ; i++ )
		if( arg->TCPManagers[i] == pthread_self() ){ index = i; break; }

	// Erreur de thread
	if( index == -1 ){
		if( DEBUGMOD&THR ) printf("{tcp_com}(%d)             Unknown thread index. Aborting\n", index);
		pthread_exit(NULL);
	}


	pindex = -1; // on ne connait pas encore l'avion

	while( 1 ){



		/* (2) Récupération de la requête
		---------------------------------------------------------*/
		/* 1. On lit sur la socket */
		read = recv(SOCKET, buffer, MAX_BUF_LEN, 0);

		/* 2.1. Si erreur reception (-1:erreur, 0:fermeture client propre) */
		if( read <= 0 ){
			if( DEBUGMOD&BUF ) printf("{tcp_com}(%d)             read: %d -> must exit thread\n", index, read);
			break;
		}

		/* 2.2. Si message trop court */
		if( read != PLANE_LEN+1 ){
			if( DEBUGMOD&BUF ) printf("{tcp_com}(%d)             read: %d (expected: %d)\n", index, read, (int) (PLANE_LEN+1) );
			continue;
		}


		/* 3. On parse la requête (indianness: network order) */
		memcpy(&tmpFlags,  buffer+sizeof(char)*0+sizeof(int)*0, sizeof(char)   );
		memcpy(&data.code, buffer+sizeof(char)*1+sizeof(int)*0, sizeof(char)*6 );
		memcpy(&data.x,    buffer+sizeof(char)*7+sizeof(int)*0, sizeof(int)    );
		memcpy(&data.y,    buffer+sizeof(char)*7+sizeof(int)*1, sizeof(int)    );
		memcpy(&data.z,    buffer+sizeof(char)*7+sizeof(int)*2, sizeof(int)    );
		memcpy(&data.spd,  buffer+sizeof(char)*7+sizeof(int)*4, sizeof(int)    );
		memcpy(&data.cap,  buffer+sizeof(char)*7+sizeof(int)*3, sizeof(int)    );

		/* 4. Gestion de l'indianness */
		data.x      = ntohl(data.x);
		data.y      = ntohl(data.y);
		data.z      = ntohl(data.z);
		data.cap    = ntohl(data.cap);
		data.spd    = ntohl(data.spd);
		data.online = 1;



		/* (3) Gestion de la requête -> enregistrement
		---------------------------------------------------------*/
		/* 1. On cherche le rang de l'avion si on l'a pas déja */
		if( pindex <= -1 ){

			/* 2. On regarde si l'avion existe */
			for( i = 0 ; i < arg->sgca->n ; i++ ){

				// Si l'avion existe -> on passe à la suite
				if( strcmp(arg->sgca->unit[i].data.code, data.code) == 0 ){
					pindex = i;
					break;
				}

			}

		}

		/* 3. Création si n'existe pas */
		if( pindex <= -1 ){

			// On ajoute une entrée à data + socket
			pindex                         = arg->sgca->n;
			arg->sgca->n++;
			arg->sgca->unit                = (struct context_unit*) realloc(arg->sgca->unit,   sizeof(struct context_unit)*arg->sgca->n + 1);
			arg->sgca->unit[pindex].socket = SOCKET;
			arg->sgca->unit[pindex].thrId  = 0xf0;
			printf("{tcp_com}(%d)             plane '%s' (#%d) created\n", index, data.code, pindex);

		}

		/* 4. On copie/met à jour les valeurs */
		if( tmpFlags != 0x10 ){ // gestion réponse ctrlTerm
			arg->sgca->unit[pindex].flags = tmpFlags;                               // on met à jour les flags
			while( arg->sgca->unit[pindex].thrId == 0xf0 );                         // on attend le numéro de thread
			pthread_kill(arg->UDPManagers[arg->sgca->unit[pindex].thrId], SIGCONT); // on réveille le thread
			arg->sgca->unit[pindex].thrId = 0xf0;                                   // on laisse la place pour qqn d'autre
		}
		memcpy(&arg->sgca->unit[pindex].data, &data, sizeof(struct plane));

		if( DEBUGMOD&COM ) printf("{tcp_com}(%d)             stored (%d)'%s': {x = %d; y = %d; z = %d; cap = %d; spd = %d}\n", index, pindex, arg->sgca->unit[pindex].data.code, arg->sgca->unit[pindex].data.x, arg->sgca->unit[pindex].data.y, arg->sgca->unit[pindex].data.z, arg->sgca->unit[pindex].data.cap, arg->sgca->unit[pindex].data.spd);

	}


	/* [5] On notifie que l'avoin est déconnecté (crash)
	=========================================================*/
	/* 1. On notifie le crash */
	arg->sgca->unit[pindex].data.online = 0;
	arg->sgca->unit[pindex].socket      = -1;





	/* [6] On libère la mémoire
	=========================================================*/



	/* [7] Fermeture de la connection (SOCKET)
	=========================================================*/
	printf("{tcp_com}(%d)             Closing communication socket\n", index);
	close(SOCKET);



	/* [n] Arrêt du THREAD
	============================================================================*/
	/* 1. On met à jour "activeManagers" */
	if( index != -1 )
		arg->activeTCPManagers[index] = 0;

	/* 2. On arrête le THREAD */
	if( DEBUGMOD&THR ) printf("{tcp_com}(%d)             freed\n", index);
	pthread_exit(NULL);
}




/* Gestion d'une connexion TERMINAL
*
* @THREADABLE_SOCKET<void*>			SOCKET de la connexion client
*
* @history
* [1] Initialisation des variables
*
* @loop
*   [2] Récupération des données
*   [3] Construction de la requête
*   [4] Envoi de la requête
*   [5] Timeout
*
* [6] On vide les buffers
* [7] Fermeture de la connection (SOCKET)
* [n] Arrêt du THREAD
* 	1. On récupère le rang dans les "managers"
* 	2. On met à jour "activeManagers"
* 	3. On arrête le THREAD
*
*/
void* manageViewTerm(void* THREADABLE_ARGS){

	/* [1] Initialisation des variables
	=========================================================*/
	/* 1. Initialisation des variables */
	struct sockaddr_in clientInfo;
	socklen_t          len;
	char               loop = 1;
	size_t             buflen;
	int                sent;                 // compteurs d'envoi
	int                i, index = -1;        // Compteurs globaux
	char*              buffer   = malloc(1); // Buffer d'envoi

	/* 2. On récupère les arguments */
	struct handler_arg* arg = THREADABLE_ARGS;

	/* 3. On récupère le rang dans les "managers" */
	for( i = 0 ; i < MAX_UDP_THR ; i++ )
		if( arg->UDPManagers[i] == pthread_self() ){ index = i; break; }

	printf("{udp_vterm}{com}(%d)      starting terminal routine (rate: %d sec)\n", index, PUBL_TIMEOUT);


	/* 4. Attente d'un client */
	len = sizeof(struct sockaddr_in);
	if( recvfrom(arg->socket, buffer, MAX_BUF_LEN*sizeof(char), 0, (struct sockaddr*) &clientInfo, &len) < 0 ){
		printf("{udp_vterm}{com}(%d)      No terminal detected, exiting\n", index);
		loop = 0;
	}

	if( loop ) printf("{udp_cterm}{com}(%d)      Terminal connected\n", index);


	while( loop ){


		/* [2] Récupération des données
		=========================================================*/
		getPlaneData(&buffer, &buflen, arg->sgca);


		/* [3] Envoi de la requête
		=========================================================*/
		/* 2. Envoi */
		len = sizeof(struct sockaddr_in);
		sent = sendto(arg->socket, buffer, buflen + 1, 0, (struct sockaddr*) &clientInfo, len);

		/* 3. Gestion erreur */
		if( sent <= 0 ){
			printf("{udp_vterm}{com}(%d)      Unable to send data\n", index);
			break;
		}


		/* [4] Réception feedback
		=========================================================*/
		/* 1. Réception feedback (0x10) */
		len = sizeof(struct sockaddr_in);
		sent = recvfrom(arg->socket, buffer, 1, 0, (struct sockaddr*) &clientInfo, &len);

		/* 2. Gestion erreur (erreur ou mauvais feedback != 0x10) */
		if( sent <= 0 || buffer[0] != 0x10 ){
			printf("{udp_vterm}{com}(%d)      Unable to recv feedback -> exiting\n", index);
			break;
		}


		/* [4] Timeout
		=========================================================*/
		sleep(PUBL_TIMEOUT);

	}


	/* [n] Arrêt du THREAD
	============================================================================*/
	/* 1. On met à jour "activeManagers" */
	if( index != -1 )
		arg->activeUDPManagers[index] = 0;

	/* 2. On ferme la socket + libère la mémoire */
	close(arg->socket);
	free(buffer);


	/* 3. On arrête le THREAD */
	if( DEBUGMOD&THR ) printf("{udp_vterm}{com}(%d)      freed\n", index);
	pthread_exit(NULL);
}





/* Gestion d'une connexion TERMINAL
*
* @THREADABLE_SOCKET<void*>			SOCKET de la connexion client
*
* @history
* [1] Initialisation des variables
* [2] Récupération de la requête
* [3] Traitement de la requête
* [4] Création de la réponse
* [5] Envoi de la réponse
* [6] On vide les buffers
* [7] Fermeture de la connection (SOCKET)
* [n] Arrêt du THREAD
* 	1. On récupère le rang dans les "managers"
* 	2. On met à jour "activeManagers"
* 	3. On arrête le THREAD
*
*/
void* manageCtrlTerm(void* THREADABLE_ARGS){

	/* [1] Initialisation des variables
	=========================================================*/
	/* 1. Initialisation des variables */
	struct sockaddr_in clientInfo;
	socklen_t          len;
	size_t             dataLen;
	char               loop = 1, update = 0, fbk = 0;
	int                count, last;                   // compteurs d'envoi
	int                i, index = -1;                 // Compteurs globaux
	int                pindex;                        // index of the current plane
	char               buffer[MAX_BUF_LEN];           // Buffer d'envoi
	char*              dataBuffer = malloc(1);
	struct term_req    request;                       // Requête
	char               flags;


	/* 2. On récupère les arguments */
	struct handler_arg* arg = THREADABLE_ARGS;

	/* 3. On récupère le rang dans les "managers" */
	for( i = 0 ; i < MAX_UDP_THR ; i++ )
		if( arg->UDPManagers[i] == pthread_self() ){ index = i; break; }

	printf("{udp_cterm}{com}(%d)      starting terminal routine\n", index);

	/* 4. Attente d'un client */
	len = sizeof(struct sockaddr_in);
	if( recvfrom(arg->socket, buffer, MAX_BUF_LEN*sizeof(char), 0, (struct sockaddr*) &clientInfo, &len) < 0 ){
		printf("{udp_cterm}{com}(%d)      No terminal detected, exiting\n", index);
		loop = 0;
	}else
		printf("{udp_cterm}{com}(%d)      Terminal connected\n", index);


	while( loop ){

		pindex = -1;

		/* [2] Récupération de la requête
		=========================================================*/
		/* 1. On lit sur la socket */
		len = sizeof(struct sockaddr_in);
		bzero(buffer, sizeof(char)*MAX_BUF_LEN);
		count = recvfrom(arg->socket, buffer, MAX_BUF_LEN, 0, (struct sockaddr*) &clientInfo, &len);

		/* 2. Si erreur reception */
		if( count <= 0 ) // because of timeout or error
			continue;

		if( count < TERMREQ_LEN ){
			send(arg->socket, "\0\0", sizeof(char)*2, 0);
			if( DEBUGMOD&BUF ) printf("{udp_cterm}{com}(%d)      Error receiving request\n", index);
			continue;
		}

		/* 3. On désérialise la requête*/
		bzero(&request, sizeof(struct term_req));
		count  = 0;    last = sizeof(char);   memcpy(&request.flags,       buffer+count, last );
		count += last; last = sizeof(char)*6; memcpy(&request.update.code, buffer+count, last );
		count += last; last = sizeof(int);    memcpy(&request.update.x,    buffer+count, last );
		count += last; last = sizeof(int);    memcpy(&request.update.y,    buffer+count, last );
		count += last; last = sizeof(int);    memcpy(&request.update.z,    buffer+count, last );
		count += last; last = sizeof(int);    memcpy(&request.update.cap,  buffer+count, last );
		count += last; last = sizeof(int);    memcpy(&request.update.spd,  buffer+count, last );

		printf("{udp_cterm}{com}(%d)      Plane req { flags: %d; plane: #%s { z=%d; cap=%d; speed=%d } }\n", index, request.flags, request.update.code, ntohl(request.update.z), ntohl(request.update.cap), ntohl(request.update.spd));


		/* [3] Gestion de la requête
		=========================================================*/
		/* 1. On vérifie qu'il y a bien un update demandé (ou DATA) */
		update = ( request.flags&TERMREQ_ALT || request.flags&TERMREQ_CAP || request.flags&TERMREQ_SPD );
		fbk    = request.flags&TERMREQ_FBK;
		if( !( update || fbk ) ){
			send(arg->socket, "\x00\0", sizeof(char)*2, 0);
			printf("{udp_cterm}{com}(%d)      Invalid flag, passing\n", index);
			continue;
		}


		/* [4] Gestion de la mise à jour de valeurs
		=========================================================*/
		while( update ){

			/* (1) Recherche de l'avion
			---------------------------------------------------------*/
			/* 1. On cherche l'avion par code */
			for( i = 0 ; i < arg->sgca->n ; i++ ){

				/* 2. Si l'avion existe et online -> on passe à la suite */
				if( strcmp(arg->sgca->unit[i].data.code, request.update.code) == 0 && arg->sgca->unit[i].data.online == 1 ){
					pindex = i;
					break;
				}

			}



			/* 3. Si on a pas trouvé -> on quitte l'udpdate */
			if( pindex <= -1 || arg->sgca->unit[pindex].socket <= -1 ){
				printf("{udp_cterm}{com}(%d)      Plane unknown or unreachable, passing\n", index);
				request.flags = 0x10;
				break;
			}

			/* (2) Transfert de la requête à l'avion
			---------------------------------------------------------*/
			/* 1. On prépare la requête */
			bzero(buffer, sizeof(char)*MAX_BUF_LEN);

			/* 2. On remplit le buffer d'envoi */
			count  = 0;    last = sizeof(char); memcpy(buffer+count, &request.flags,      last );
			count += last; last = sizeof(int);  memcpy(buffer+count, &request.update.z,   last );
			count += last; last = sizeof(int);  memcpy(buffer+count, &request.update.cap, last );
			count += last; last = sizeof(int);  memcpy(buffer+count, &request.update.spd, last );
			count += last;

			/* 3. On envoie la requête à l'avion */
			if( send(arg->sgca->unit[pindex].socket, buffer, count/sizeof(char), 0) <= 0 ){
				printf("{udp_cterm}{com}(%d)      Cannot send request to plane\n", index);
				request.flags = 0x10;
				break;
			}


			/* (3) Gestion de la réponse de l'avion
			---------------------------------------------------------*/
			// Attente du flag de réponse
			arg->sgca->unit[pindex].thrId = index;
			sleep(2);
			flags = arg->sgca->unit[pindex].flags; // on récupère les flags
			arg->sgca->unit[pindex].flags = 0x10;  // on reset les flags

			if( flags == 0x10 ){
				printf("{udp_cterm}{com}(%d)      Cannot get response from plane %d (%d)\n", index, count, (int) PLANE_LEN);
				request.flags = 0x10;
				break;
			}

			printf("{udp_cterm}{com}(%d)      Plane res { flags: %d }\n", index, flags);

			/* 2. Gestion de la validation de l'update */
			if( !(flags&TERMREQ_ALT) && request.flags&TERMREQ_ALT )
				request.flags -= TERMREQ_ALT;

			if( !(flags&TERMREQ_CAP) && request.flags&TERMREQ_CAP )
				request.flags -= TERMREQ_CAP;

			if( !(flags&TERMREQ_SPD) && request.flags&TERMREQ_SPD )
				request.flags -= TERMREQ_SPD;

			/* 3. Si pas de demande des données -> on pré-remplie la requête */
			if( !fbk ){
				dataBuffer    = realloc(dataBuffer, sizeof(char)*2 );
				dataBuffer[0] = request.flags;
				dataBuffer[1] = 0;
				dataLen = 2;
			}

			break;

		}

		if( fbk ){

			/* [5] Données des avions
			=========================================================*/
			/* 1. On récupère les données */
			getPlaneData(&dataBuffer, &dataLen, arg->sgca);

			/* 2. Si update, on mixe les flags */
			if( update )
				dataBuffer[0] |= request.flags;

		}


		/* [6] Réponse au terminal
		=========================================================*/
		printf("{udp_cterm}{com}(%d)      Sending response { flags: %d; n: %d }\n", index, dataBuffer[0], dataBuffer[1]);
		len = sizeof(struct sockaddr_in);
		if( sendto(arg->socket, dataBuffer, dataLen, 0, (struct sockaddr*) &clientInfo, len) < 0 )
			printf("{udp_cterm}{com}(%d)      Cannot answer to terminal\n", index);


	}


	/* [n] Arrêt du THREAD
	============================================================================*/
	/* 1. On met à jour "activeManagers" */
	if( index != -1 )
		arg->activeUDPManagers[index] = 0;

	/* 2. On ferme la socket */
	close(arg->socket);
	free(dataBuffer);

	/* 3. On arrête le THREAD */
	if( DEBUGMOD&THR ) printf("{udp_cterm}{com}(%d)      freed\n", index);
	pthread_exit(NULL);
}




















void getPlaneData(char** pBuffer, size_t* pLen, struct context* pContext){
	/* [1] Initialisation des variables
	=========================================================*/
	/* 1. Initialisation des variables */
	int                count, last, i, nb; // compteurs
	struct term_res    response;               // Requête


	/* [1] Récupération des données
	=========================================================*/
	/* 1. On initialise les variables utiles */
	response.flags = TERMREQ_FBK;
	nb             = pContext->n;
	response.n     = 0; // nombre d'avions online

	response.data  = malloc( nb * sizeof(struct plane) + 1 );

	/* 2. On récupère la liste des avions (network byte order) */
	for( i = 0 ; i < nb ; i++ ){

		memcpy(&response.data[i].code, &pContext->unit[i].data.code, sizeof(char)*6);
		response.data[i].x      = htonl( pContext->unit[i].data.x   );
		response.data[i].y      = htonl( pContext->unit[i].data.y   );
		response.data[i].z      = htonl( pContext->unit[i].data.z   );
		response.data[i].cap    = htonl( pContext->unit[i].data.cap );
		response.data[i].spd    = htonl( pContext->unit[i].data.spd );
		response.data[i].online =        pContext->unit[i].data.online;

		// Incrément du compte si online
		response.n += response.data[i].online;

	}


	/* [3] Construction de la requête
	=========================================================*/
	/* 1. Initialisation du buffer */
	*pBuffer = realloc(*pBuffer, sizeof(char)*2 + PLANE_LEN * response.n + sizeof(char));

	/* 2. Copie des données globales */
	count  = 0;    last = sizeof(char); memcpy(*pBuffer+count, &response.flags, last);
	count += last; last = sizeof(char); memcpy(*pBuffer+count, &response.n,     last);

	/* 3. Copie des données des avions */
	for( i = 0 ; i < nb ; i++ ){

		// Ignore les avions offline
		if( !response.data[i].online )
			continue;

		count += last; last = sizeof(char)*6; memcpy(*pBuffer+count, &response.data[i].code, last);
		count += last; last = sizeof(int);    memcpy(*pBuffer+count, &response.data[i].x,    last);
		count += last; last = sizeof(int);    memcpy(*pBuffer+count, &response.data[i].y,    last);
		count += last; last = sizeof(int);    memcpy(*pBuffer+count, &response.data[i].z,    last);
		count += last; last = sizeof(int);    memcpy(*pBuffer+count, &response.data[i].cap,  last);
		count += last; last = sizeof(int);    memcpy(*pBuffer+count, &response.data[i].spd,  last);
	}

	*pLen = (count+last) / sizeof(char);
	free(response.data);

}