PeerStreamer

#include <arpa/inet.h>

#include <netinet/in.h>

#include <sys/socket.h>

#include <fcntl.h>

#include <event2/event.h>

#include <stdlib.h>

#include <unistd.h>

#include <stdint.h>

#include <string.h>

#include <netdb.h>

#include <math.h>

#include "util/udpSocket.h"

#include "util/stun.h"

#include "transmissionHandler.h"

#define LOG_MODULE "[ml] "

#include "grapes_log.h"

/**************************** START OF INTERNALS ***********************/

/*

 * a pointer to a libevent instance

 */

struct event_base *base;

/*

 * define the nr of connections the messaging layer can handle

 */

#define CONNECTBUFSIZE 10000

/*

 * define the nr of data that can be received parallel

 */

#define RECVDATABUFSIZE 10000

/*

 * define an array for message multiplexing

 */

#define MSGMULTIPLEXSIZE 127

/*

 * global variables

 */

/*

 * define a buffer of pointers to connect structures

 */

connect_data *connectbuf[CONNECTBUFSIZE];

/*

 * define a pointer buffer with pointers to recv_data structures

 */

recvdata *recvdatabuf[RECVDATABUFSIZE];

/*

 * define a pointer buffer for message multiplexing

 */

receive_data_cb recvcbbuf[MSGMULTIPLEXSIZE];

/*

 * stun server address

 */

struct sockaddr_in stun_server;

/*

 * receive timeout

 */

struct timeval recv_timeout;

/*

 * boolean NAT traversal successful if true

 */

boolean NAT_traversal;

/*

 * file descriptor for local socket

 */

evutil_socket_t socketfd;

/*

 * local socketID

 */

socket_ID local_socketID;

socketID_handle loc_socketID = &local_socketID;

/*

 * callback function pointers

 */

/*

 * monitoring module callbacks

 */

get_recv_pkt_inf_cb get_Recv_pkt_inf_cb = NULL;

get_send_pkt_inf_cb get_Send_pkt_inf_cb = NULL;

set_monitoring_header_pkt_cb set_Monitoring_header_pkt_cb = NULL;

get_recv_data_inf_cb get_Recv_data_inf_cb = NULL;

get_send_data_inf_cb get_Send_data_inf_cb = NULL;

set_monitoring_header_data_cb set_Monitoring_header_data_cb = NULL;

/*

 * connection callbacks

 */

receive_connection_cb receive_Connection_cb = NULL;

connection_failed_cb failed_Connection_cb = NULL;

/*

 * local socketID callback

 */

receive_localsocketID_cb receive_SocketID_cb;

/*

 * boolean that defines if received data is transmitted to the upper layer

 * via callback or via upper layer polling

 */

boolean recv_data_callback;

/*

 * helper function to get rid of a warning

 */

int min(int a, int b) {

	if (a > b) return b;

	return a;

}

void register_recv_localsocketID_cb(receive_localsocketID_cb local_socketID_cb)

{

	if (local_socketID_cb == NULL)

		printf("Register receive_localsocketID_cb: NULL ptr \n");

	else

	receive_SocketID_cb = local_socketID_cb;

}

void keep_connection_alive(const int connectionID)

{

    // to be done with the NAT traversal

    // send a message over the wire

    printf("\n");

}

void unsetStunServer()

{

	stun_server.sin_addr.s_addr = 0;

}

bool isStunDefined()

{

	return stun_server.sin_addr.s_addr;

}

void send_msg(int con_id, int msg_type, char* msg, int msg_len, bool truncable, send_params * sParams) {

	socketaddrgen udpgen;

	bool retry;

	int pkt_len, offset;

	struct iovec iov[4];

	char h_pkt[MON_HEADER_SPACE];

	char h_data[MON_HEADER_SPACE];

	struct msg_header msg_h;

	iov[0].iov_base = &msg_h;

	iov[0].iov_len = MSG_HEADER_SIZE;

	msg_h.local_con_id = con_id;

	msg_h.remote_con_id = connectbuf[con_id]->external_connectionID;

	msg_h.msg_type = msg_type;

	msg_h.msg_seq_num = connectbuf[con_id]->seqnr++;

	iov[1].iov_len = iov[2].iov_len = 0;

	iov[1].iov_base = h_pkt;

	iov[2].iov_base = h_data;

	if (connectbuf[con_id]->internal_connect)

		udpgen = connectbuf[con_id]->external_socketID.internal_addr;

	else

		udpgen = connectbuf[con_id]->external_socketID.external_addr;

	do{

		offset = 0;

		retry = false;

		// Monitoring layer hook

		if(set_Monitoring_header_data_cb != NULL) {

			iov[2].iov_len = ((set_Monitoring_header_data_cb) (&(connectbuf[con_id]->external_socketID), msg_type));

		}

		msg_h.len_mon_data_hdr = iov[2].iov_len;

		if(get_Send_data_inf_cb != NULL && iov[2].iov_len != 0) {

			mon_data_inf sd_data_inf;

			sd_data_inf.remote_socketID = &(connectbuf[con_id]->external_socketID);

			sd_data_inf.buffer = msg;

			sd_data_inf.bufSize = msg_len;

			sd_data_inf.msgtype = msg_type;

			sd_data_inf.monitoringDataHeader = iov[2].iov_base;

			sd_data_inf.monitoringDataHeaderLen = iov[2].iov_len;

			sd_data_inf.priority = sParams->priority;

			sd_data_inf.padding = sParams->padding;

			sd_data_inf.confirmation = sParams->confirmation;

			sd_data_inf.reliable = sParams->reliable;

			memset(&sd_data_inf.arrival_time, 0, sizeof(struct timeval));

			(get_Send_data_inf_cb) ((void *) &sd_data_inf);

		}

		do {

			if(set_Monitoring_header_pkt_cb != NULL) {

				iov[1].iov_len = (set_Monitoring_header_pkt_cb) (&(connectbuf[con_id]->external_socketID), msg_type);

			}

			pkt_len = min(connectbuf[con_id]->pmtusize - iov[2].iov_len - iov[1].iov_len - iov[0].iov_len, msg_len - offset) ;

			iov[3].iov_len = pkt_len;

			iov[3].iov_base = msg + offset;

			//fill header

			msg_h.len_mon_packet_hdr = iov[1].iov_len;

			msg_h.offset = offset;

			msg_h.msg_length = truncable ? pkt_len : msg_len;

			//monitoring layer hook

			if(get_Send_pkt_inf_cb != NULL && iov[1].iov_len) {

				mon_pkt_inf pkt_info;

				pkt_info.remote_socketID = &(connectbuf[con_id]->external_socketID);

				pkt_info.buffer = msg + offset;

				pkt_info.bufSize = pkt_len;

				pkt_info.msgtype = msg_type;

				pkt_info.dataID = connectbuf[con_id]->seqnr;

				pkt_info.offset = offset;

				pkt_info.datasize = msg_len;

				pkt_info.monitoringHeaderLen = iov[1].iov_len;

				pkt_info.monitoringHeader = iov[1].iov_base;

				pkt_info.ttl = -1;

				memset(&(pkt_info.arrival_time),0,sizeof(struct timeval));

				(get_Send_pkt_inf_cb) ((void *) &pkt_info);

			}

			switch(sendPacket(socketfd, iov, 4, &udpgen.udpaddr)) {

				case MSGLEN:

					connectbuf[con_id]->pmtusize = pmtu_decrement(connectbuf[con_id]->pmtusize);

					connectbuf[con_id]->delay = true;

					retry = true;

					offset = msg_len; // exit the while

					break;

				case FAILURE:

					offset = msg_len; // exit the while

					break;

				case OK:

					//update

					offset += pkt_len + iov[2].iov_len;

					//transmit data header only in the first packet

					iov[2].iov_len = 0;

					break;

			}

		} while(offset != msg_len + msg_h.len_mon_data_hdr && !truncable);

	} while(retry);

}

void send_conn_msg(int con_id, int buf_size, int command_type)

{

	send_params sparams;

	if(connectbuf[con_id]->ctrl_msg_buf == NULL)

		connectbuf[con_id]->ctrl_msg_buf = malloc(MAX);

	if(connectbuf[con_id]->ctrl_msg_buf == NULL) {

		error("ML: can not allocate memory for connection message");

		return;

	}

	struct conn_msg *msg_header = (struct conn_msg*) connectbuf[con_id]->ctrl_msg_buf;

	msg_header->comand_type = command_type;

	msg_header->pmtu_size = connectbuf[con_id]->pmtusize;

	memcpy(&(msg_header->sock_id), loc_socketID, sizeof(socket_ID));

	send_msg(con_id, ML_CON_MSG, connectbuf[con_id]->ctrl_msg_buf, buf_size, true, &sparams);

}

void

recv_conn_msg(struct msg_header *msg_h, char *msgbuf, int bufsize)

{

	struct conn_msg *con_msg;

	int free_con_id, con_id;

	time_t now = time(NULL);

	double timediff = 0.0;

	// Monitoring layer hook

	if(get_Recv_data_inf_cb != NULL && msg_h->len_mon_data_hdr != 0) {

		// update pointer to the real data

		msgbuf += msg_h->len_mon_data_hdr;

		bufsize -= msg_h->len_mon_data_hdr;

		con_msg = (struct conn_msg *)msgbuf;

		mon_data_inf recv_data_inf;

		recv_data_inf.remote_socketID = &(con_msg->sock_id);

		recv_data_inf.buffer = msgbuf;

		recv_data_inf.bufSize = bufsize;

		recv_data_inf.msgtype = msg_h->msg_type;

		recv_data_inf.monitoringDataHeaderLen = msg_h->len_mon_data_hdr;

		recv_data_inf.monitoringDataHeader = msg_h->len_mon_data_hdr ? msgbuf : NULL;

		gettimeofday(&recv_data_inf.arrival_time, NULL);

		recv_data_inf.firstPacketArrived = true;

		recv_data_inf.recvFragments = 1;

		recv_data_inf.priority = false;

		recv_data_inf.padding = false;

		recv_data_inf.confirmation = false;

		recv_data_inf.reliable = false;

		// send data recv callback to monitoring module

		(get_Recv_data_inf_cb) ((void *) &recv_data_inf);

	} else

		con_msg = (struct conn_msg *) msgbuf;

	// check the connection command type

	switch (con_msg->comand_type) {

		/*

		* if INVITE: enter a new socket make new entry in connect array

		* send an ok

		*/

		case INVITE:

			debug("ML: received INVITE");

			/*

			* check if another connection for the external connectionID exist

			* that was established within the last 2 seconds

			*/

			free_con_id = -1;

			for (con_id = 0; con_id < CONNECTBUFSIZE; con_id++) {

				if (connectbuf[con_id] != NULL) {

					if (mlCompareSocketIDs(&(connectbuf[con_id]->external_socketID), &(con_msg->sock_id)) == 0) {

						timediff = difftime(now, connectbuf[con_id]->starttime);

						if (timediff < 2)

							break;

					}

				} else if(free_con_id == -1)

					free_con_id = con_id;

			}

			if (con_id == CONNECTBUFSIZE) {

				// create an entry in the connecttrybuf

				if(free_con_id == -1) {

					error("ML: no new connect_buf available");

					return;

				}

				connectbuf[free_con_id] = (connect_data *) malloc(sizeof(connect_data));

				memset(connectbuf[free_con_id],0,sizeof(connect_data));

				connectbuf[free_con_id]->connection_head = connectbuf[free_con_id]->connection_last = NULL;

				connectbuf[free_con_id]->starttime = time(NULL);

				memcpy(&(connectbuf[free_con_id]->external_socketID), &(con_msg->sock_id), sizeof(socket_ID));

				connectbuf[free_con_id]->pmtusize = con_msg->pmtu_size;

				connectbuf[free_con_id]->external_connectionID = msg_h->local_con_id;

				connectbuf[free_con_id]->internal_connect =

					!compare_external_address_socketIDs(&(con_msg->sock_id), loc_socketID);

				con_id = free_con_id;

			}

			if(connectbuf[con_id]->status <= CONNECT) {

				//update status and send back answer

				connectbuf[con_id]->status = CONNECT;

				send_conn_msg(con_id, con_msg->pmtu_size, CONNECT);

			}

			break;

		case CONNECT:

			debug("ML: received CONNECT");

			if(msg_h->remote_con_id != -1 && connectbuf[msg_h->remote_con_id] == NULL) {

				error("ML: received CONNECT for unexistent connection");

				return;

			}

			/*

			* check if the connection status is not already 1 or 2

			*/

			if (connectbuf[msg_h->remote_con_id]->status == INVITE) {

				// set the external connectionID

				connectbuf[msg_h->remote_con_id]->external_connectionID = msg_h->local_con_id;

				// change status con_msg the connection_data

				connectbuf[msg_h->remote_con_id]->status = READY;

				// change pmtusize in the connection_data

				connectbuf[msg_h->remote_con_id]->pmtusize = con_msg->pmtu_size;

				// send the READY

				send_conn_msg(msg_h->remote_con_id, con_msg->pmtu_size, READY);

				if (receive_Connection_cb != NULL)

					(receive_Connection_cb) (msg_h->remote_con_id, NULL);

				// call all registered callbacks

				while(connectbuf[msg_h->remote_con_id]->connection_head != NULL) {

					struct receive_connection_cb_list *temp;

					temp = connectbuf[msg_h->remote_con_id]->connection_head;

					(temp->connection_cb) (msg_h->remote_con_id, temp->arg);

					connectbuf[msg_h->remote_con_id]->connection_head = temp->next;

					free(temp);

				}

				connectbuf[msg_h->remote_con_id]->connection_head =

					connectbuf[msg_h->remote_con_id]->connection_last = NULL;

			} else

				// send the READY

				send_conn_msg(msg_h->remote_con_id, con_msg->pmtu_size, READY);

			debug("ML: active connection established");

			break;

			/*

			* if READY: find the entry in the connection array set the

			* connection active change the pmtu size

			*/

		case READY:

			debug("ML: received READY");

			if(connectbuf[msg_h->remote_con_id] == NULL) {

				error("ML: received READY for unexistent connection");

				return;

			}

			/*

			* checks if the connection is not already established

			*/

			if (connectbuf[msg_h->remote_con_id]->status == CONNECT) {

				// change status of the connection

				connectbuf[msg_h->remote_con_id]->status = 2;

				// change pmtusize

				connectbuf[msg_h->remote_con_id]->pmtusize = con_msg->pmtu_size;

				if (receive_Connection_cb != NULL)

					(receive_Connection_cb) (msg_h->remote_con_id, NULL);

				while(connectbuf[msg_h->remote_con_id]->connection_head != NULL) {

					struct receive_connection_cb_list *temp;

					temp = connectbuf[msg_h->remote_con_id]->connection_head;

					(temp->connection_cb) (msg_h->remote_con_id, temp->arg);

					connectbuf[msg_h->remote_con_id]->connection_head = temp->next;

					free(temp);

				}

				connectbuf[msg_h->remote_con_id]->connection_head =

					connectbuf[msg_h->remote_con_id]->connection_last = NULL;

				debug("ML: passiv connection established");

			}

			break;

	}

}

void recv_stun_msg(char *msgbuf, int recvSize)

{

	/*

	* create empty stun message struct

	*/

	StunMessage resp;

	memset(&resp, 0, sizeof(StunMessage));

	/*

	* parse the message

	*/

	int returnValue = 0;

	returnValue = recv_stun_message(msgbuf, recvSize, &resp);

	if (returnValue == 0) {

		/*

		* read the reflexive Address into the local_socketID

		*/

		struct sockaddr_in reflexiveAddr;

		reflexiveAddr.sin_family = AF_INET;

		reflexiveAddr.sin_addr.s_addr = htonl(resp.mappedAddress.ipv4.addr);

		reflexiveAddr.sin_port = htons(resp.mappedAddress.ipv4.port);

		socketaddrgen reflexiveAddres;

		reflexiveAddres.udpaddr = reflexiveAddr;

		local_socketID.external_addr = reflexiveAddres;

		NAT_traversal = true;

		// callback to the upper layer indicating that the socketID is now

		// ready to use

		(receive_SocketID_cb) (&local_socketID, 0);

	}

}

//done

void recv_timeout_cb(int fd, short event, void *arg)

{

	int recv_id = (long) arg;

	debug("ML: recv_timeout_cb called");

	if (recvdatabuf[recv_id] == NULL) {

		return;

	}

	if(recvdatabuf[recv_id]->status == ACTIVE) {

		//TODO make timeout at least a DEFINE

		struct timeval timeout = { 4, 0 };

		recvdatabuf[recv_id]->status = INACTIVE;

		event_base_once(base, -1, EV_TIMEOUT, recv_timeout_cb,

			arg, &timeout);

		return;

	}

	if(recvdatabuf[recv_id]->status == INACTIVE) {

		// Monitoring layer hook

		if(get_Recv_data_inf_cb != NULL) {

			mon_data_inf recv_data_inf;

			recv_data_inf.remote_socketID =

					&(connectbuf[recvdatabuf[recv_id]->connectionID]->external_socketID);

			recv_data_inf.buffer = recvdatabuf[recv_id]->recvbuf;

			recv_data_inf.bufSize = recvdatabuf[recv_id]->bufsize;

			recv_data_inf.msgtype = recvdatabuf[recv_id]->msgtype;

			recv_data_inf.monitoringDataHeaderLen = recvdatabuf[recv_id]->monitoringDataHeaderLen;

			recv_data_inf.monitoringDataHeader = recvdatabuf[recv_id]->monitoringDataHeaderLen ?

				recvdatabuf[recv_id]->recvbuf : NULL;

			gettimeofday(&recv_data_inf.arrival_time, NULL);

			recv_data_inf.firstPacketArrived = recvdatabuf[recv_id]->firstPacketArrived;

			recv_data_inf.recvFragments = recvdatabuf[recv_id]->recvFragments;

			recv_data_inf.priority = false;

			recv_data_inf.padding = false;

			recv_data_inf.confirmation = false;

			recv_data_inf.reliable = false;

			// send data recv callback to monitoring module

			(get_Recv_data_inf_cb) ((void *) &recv_data_inf);

		}

		// Get the right callback

		receive_data_cb receive_data_callback = recvcbbuf[recvdatabuf[recv_id]->msgtype];

		recv_params rParams;

		rParams.nrMissingBytes = recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->arrivedBytes;

		rParams.recvFragments = recvdatabuf[recv_id]->recvFragments;

		rParams.msgtype = recvdatabuf[recv_id]->msgtype;

		rParams.connectionID = recvdatabuf[recv_id]->connectionID;

		rParams.remote_socketID =

			&(connectbuf[recvdatabuf[recv_id]->connectionID]->external_socketID);

		rParams.firstPacketArrived = recvdatabuf[recv_id]->firstPacketArrived;

		(receive_data_callback) (recvdatabuf[recv_id]->recvbuf + recvdatabuf[recv_id]->monitoringDataHeaderLen, recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->monitoringDataHeaderLen,

			recvdatabuf[recv_id]->msgtype, &rParams);

		//clean up

		free(recvdatabuf[recv_id]->recvbuf);

		free(recvdatabuf[recv_id]);

		recvdatabuf[recv_id] = NULL;

	}

}

// process a single recv data message

void recv_data_msg(struct msg_header *msg_h, char *msgbuf, int bufsize)

{

	debug("transmissionHandler: received data message called");

	int recv_id, free_recv_id = -1;

	if(connectbuf[msg_h->remote_con_id] == NULL) {

		debug("ML: Received a message not related to any opened connection!");

		return;

	}

	// check if a recv_data exist and enter data

	for (recv_id = 0; recv_id < RECVDATABUFSIZE; recv_id++)

		if (recvdatabuf[recv_id] != NULL) {

			if (msg_h->remote_con_id == recvdatabuf[recv_id]->connectionID &&

					msg_h->msg_seq_num == recvdatabuf[recv_id]->seqnr)

						break;

		} else

			if(free_recv_id == -1)

				free_recv_id = recv_id;

	if(recv_id == RECVDATABUFSIZE) {

		//no recv_data found: create one

		recv_id = free_recv_id;

		recvdatabuf[recv_id] = (recvdata *) malloc(sizeof(recvdata));

		memset(recvdatabuf[recv_id], 0, sizeof(recvdata));

		recvdatabuf[recv_id]->connectionID = msg_h->remote_con_id;

		recvdatabuf[recv_id]->seqnr = msg_h->msg_seq_num;

		recvdatabuf[recv_id]->monitoringDataHeaderLen = msg_h->len_mon_data_hdr;

		recvdatabuf[recv_id]->bufsize = msg_h->msg_length + msg_h->len_mon_data_hdr;

		recvdatabuf[recv_id]->recvbuf = (char *) malloc(recvdatabuf[recv_id]->bufsize);

		/*

		* read the timeout data and set it

		*/

		recvdatabuf[recv_id]->timeout_value.tv_sec = recv_timeout.tv_sec;

		recvdatabuf[recv_id]->timeout_value.tv_usec = recv_timeout.tv_usec;

		recvdatabuf[recv_id]->recvID = recv_id;

		recvdatabuf[recv_id]->starttime = time(NULL);

		recvdatabuf[recv_id]->msgtype = msg_h->msg_type;

		// fill the buffer with zeros

		memset(recvdatabuf[recv_id]->recvbuf, 0, msg_h->msg_length);

	}

	if (msg_h->offset == 0)

		recvdatabuf[recv_id]->firstPacketArrived = 1;

	// increment fragmentnr

	recvdatabuf[recv_id]->recvFragments++;

	// increment the arrivedBytes

	recvdatabuf[recv_id]->arrivedBytes += bufsize;

	// enter the data into the buffer

	memcpy(recvdatabuf[recv_id]->recvbuf + msg_h->offset, msgbuf, bufsize);

	//TODO very basic checkif all fragments arrived: has to be reviewed

	if(recvdatabuf[recv_id]->arrivedBytes == recvdatabuf[recv_id]->bufsize)

		recvdatabuf[recv_id]->status = COMPLETE; //buffer full -> msg completly arrived

	else

		recvdatabuf[recv_id]->status = ACTIVE;

	if (recv_data_callback) {

		if(recvdatabuf[recv_id]->status == COMPLETE) {

			// Monitoring layer hook

			if(get_Recv_data_inf_cb != NULL) {

				mon_data_inf recv_data_inf;

				recv_data_inf.remote_socketID =

					 &(connectbuf[recvdatabuf[recv_id]->connectionID]->external_socketID);

				recv_data_inf.buffer = recvdatabuf[recv_id]->recvbuf;

				recv_data_inf.bufSize = recvdatabuf[recv_id]->bufsize;

				recv_data_inf.msgtype = recvdatabuf[recv_id]->msgtype;

				recv_data_inf.monitoringDataHeaderLen = recvdatabuf[recv_id]->monitoringDataHeaderLen;

				recv_data_inf.monitoringDataHeader = recvdatabuf[recv_id]->monitoringDataHeaderLen ?

					recvdatabuf[recv_id]->recvbuf : NULL;

				gettimeofday(&recv_data_inf.arrival_time, NULL);

				recv_data_inf.firstPacketArrived = recvdatabuf[recv_id]->firstPacketArrived;

				recv_data_inf.recvFragments = recvdatabuf[recv_id]->recvFragments;

				recv_data_inf.priority = false;

				recv_data_inf.padding = false;

				recv_data_inf.confirmation = false;

				recv_data_inf.reliable = false;

				// send data recv callback to monitoring module

				(get_Recv_data_inf_cb) ((void *) &recv_data_inf);

			}

			// Get the right callback

			receive_data_cb receive_data_callback = recvcbbuf[msg_h->msg_type];

			if (receive_data_callback) {

				recv_params rParams;

				rParams.nrMissingBytes = recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->arrivedBytes;

				rParams.recvFragments = recvdatabuf[recv_id]->recvFragments;

				rParams.msgtype = recvdatabuf[recv_id]->msgtype;

				rParams.connectionID = recvdatabuf[recv_id]->connectionID;

				rParams.remote_socketID =

					&(connectbuf[recvdatabuf[recv_id]->connectionID]->external_socketID);

				rParams.firstPacketArrived = recvdatabuf[recv_id]->firstPacketArrived;

				(receive_data_callback) (recvdatabuf[recv_id]->recvbuf + recvdatabuf[recv_id]->monitoringDataHeaderLen, recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->monitoringDataHeaderLen,

					recvdatabuf[recv_id]->msgtype, (void *) &rParams);

			} else {

			    fprintf(stderr,"ML: Warning: callback not initialized for this message type: %d!\n",msg_h->msg_type);

			}

			//clean up

			free(recvdatabuf[recv_id]->recvbuf);

			free(recvdatabuf[recv_id]);

			recvdatabuf[recv_id] = NULL;

		} else { // not COMPLETE

			//start time out

			//TODO make timeout at least a DEFINE

			struct timeval timeout = { 4, 0 };

			event_base_once(base, -1, EV_TIMEOUT, &recv_timeout_cb, (void *) (long)recv_id, &timeout);

		}

	}

}

//done

void pmtu_timeout_cb(int fd, short event, void *arg)

{

	debug("ML: pmtu timeout called");

	int con_id = (long) arg;

	pmtu new_pmtusize;

	struct timeval timeout;

	if(connectbuf[con_id] == NULL) {

		error("ML: pmtu timeout called on non existing con_id");

		return;

	}

	if(connectbuf[con_id]->status == READY) {

		// nothing to do anymore

		return;

	}

	timeout = connectbuf[con_id]->timeout_value;

	if(connectbuf[con_id]->delay || connectbuf[con_id]->trials == MAX_TRIALS - 1) {

		double delay = timeout.tv_sec + timeout.tv_usec / 1000000.0;

		delay = delay * 5;

		timeout.tv_sec = floor(delay);

		timeout.tv_usec = fmod(delay, 1.0) * 1000000.0;

		if(connectbuf[con_id]->delay) {

			connectbuf[con_id]->delay = false;

			goto reschedule;

		}

	}

	if(connectbuf[con_id]->trials == MAX_TRIALS) {

		// decrement the pmtu size

		connectbuf[con_id]->pmtusize = pmtu_decrement(connectbuf[con_id]->pmtusize);

		connectbuf[con_id]->trials = 0;

	}

	//error in PMTU discovery?

	if (connectbuf[con_id]->pmtusize == ERROR) {

		if (connectbuf[con_id]->internal_connect == true) {

			//as of now we tried directly connecting, now let's try trough the NAT

			connectbuf[con_id]->internal_connect = false;

			connectbuf[con_id]->pmtusize = MAX;

		} else {

			//nothing to do we have to give up

			error("ML: Could not create connection with connectionID %i!",con_id);

			// envoke the callback for failed connection establishment

			if(failed_Connection_cb != NULL)

				(failed_Connection_cb) (con_id, NULL);

			// delete the connection entry

			mlCloseConnection(con_id);

			return;

		}

	}

	//retry with new pmtu size

	connectbuf[con_id]->trials++;

	send_conn_msg(con_id, connectbuf[con_id]->pmtusize, connectbuf[con_id]->status);

reschedule:

	/* reschedule */

	event_base_once(base, -1, EV_TIMEOUT, pmtu_timeout_cb, (void *) (long)con_id, &timeout);

}

/*

 * decrements the mtu size

 */

pmtu pmtu_decrement(pmtu pmtusize)

{

	pmtu pmtu_return_size;

	switch(pmtusize) {

	case MAX:

		return DSL;

	case DSL:

		return DSLMEDIUM;

	case DSLMEDIUM:

		return DSLSLIM;

	case DSLSLIM:

		return BELOWDSL;

	case BELOWDSL:

		return MIN;

	default:

		return ERROR;

	}

}

void pmtu_error_cb_th(char *msg, int msglen)

{

	debug("ML: pmtu_error callback called msg_size: %d",msglen);

	//TODO debug

	return;

    char *msgbufptr = NULL;

    int msgtype;

    int connectionID;

    pmtu pmtusize;

    pmtu new_pmtusize;

    int dead = 0;

    // check the packettype

    msgbufptr = &msg[0];

    // check the msgtype

    msgbufptr = &msg[1];

    memcpy(&msgtype, msgbufptr, 4);

    if (msgtype == 0) {

	// get the connectionID

	msgbufptr = &msg[5];

	memcpy(&connectionID, msgbufptr, 4);

	int msgtype_c = connectbuf[connectionID]->status;

//	pmtusize = connectbuf[connectionID]->pmtutrysize;

	if (msgtype_c != msgtype) {

	    dead = 1;

	}

    } else if (msgtype == 1) {

	// read the connectionID

	msgbufptr = &msg[9];

	memcpy(&connectionID, msgbufptr, 4);

	int msgtype_c = connectbuf[connectionID]->status;

//	pmtusize = connectbuf[connectionID]->pmtutrysize;

	if (msgtype_c != msgtype) {

	    dead = 1;

	}

    }

    // decrement the pmtu size

    new_pmtusize = pmtu_decrement(pmtusize);

//    connectbuf[connectionID]->pmtutrysize = new_pmtusize;

    if (new_pmtusize == ERROR) {

		fprintf(stderr,

			"transmissionHandler:  Could not create connection with connectionID %i !\n",

			connectionID);

		if(failed_Connection_cb != NULL)

			(failed_Connection_cb) (connectionID, NULL);

		// set the message type to a non existent message

		msgtype = 2;

		// delete the connection entry

		 mlCloseConnection(connectionID);

	}

    if (msgtype == 0 && dead != 1) {

	// stop the timeout event

	// timeout_del(connectbuf[connectionID]->timeout);

	/*

	 * libevent2

	 */

	// event_del(connectbuf[connectionID]->timeout);

	// create and send a connection message

// 	create_conn_msg(new_pmtusize, connectionID,

// 			&local_socketID, INVITE);

//	send_conn_msg(connectionID, new_pmtusize);

	// set a timeout event for the pmtu discovery

	// timeout_set(connectbuf[connectionID]->timeout,pmtu_timeout_cb,(void

	// *)&connectionID);

	// timeout_add(connectbuf[connectionID]->timeout,&connectbuf[connectionID]->timeout_value);

	/*

	 * libevent2

	 */

	struct event *ev;

	ev = evtimer_new(base, pmtu_timeout_cb,

			 (void *) connectbuf[connectionID]);

	// connectbuf[connectionID]->timeout = ev;

	event_add(ev, &connectbuf[connectionID]->timeout_value);

    } else if (msgtype == 1 && dead != 1) {

	// stop the timeout event

	// timeout_del(connectbuf[connectionID]->timeout);

	/*

	 * libevent2

	 */

	// printf("still here 11 \n");

	// printf("ev %d \n",connectbuf[connectionID]->timeout);

	// event_del(connectbuf[connectionID]->timeout );

	// evtimer_del(connectbuf[connectionID]->timeout );

// 	// create and send a connection message

// 	create_conn_msg(new_pmtusize,

// 			connectbuf[connectionID]->connectionID,

// 			NULL, CONNECT);

	//send_conn_msg(connectionID, new_pmtusize);

	// set a timeout event for the pmtu discovery

	// timeout_set(connectbuf[connectionID]->timeout,pmtu_timeout_cb,(void

	// *)&connectionID);

	// timeout_add(connectbuf[connectionID]->timeout,&connectbuf[connectionID]->timeout_value);

	/*

	 * libevent2

	 */

	// struct event *ev;

	// ev = evtimer_new(base,pmtu_timeout_cb, (void

	// *)connectbuf[connectionID]);

	// connectbuf[connectionID]->timeout = ev;

	// event_add(ev,&connectbuf[connectionID]->timeout_value);

    }

}

/*

 * what to do once a packet arrived if it is a conn packet send it to

 * recv_conn handler if it is a data packet send it to the recv_data

 * handler

 */

//done --

void recv_pkg(int fd, short event, void *arg)

{

	debug("ML: recv_pkg called");

	struct msg_header *msg_h;

	char msgbuf[MAX];

	char *bufptr = msgbuf;

	int ttl;

	struct sockaddr_in recv_addr;

	pmtu recvSize = MAX;

	int msg_size;

	recvPacket(fd, msgbuf, &recvSize, &recv_addr, pmtu_error_cb_th, &ttl);

	// check if it is not just an ERROR message

	if(recvSize < 0)

		return;

	// @TODO check if this simplistic STUN message recognition really always works, probably not

	unsigned short stun_bind_response = 0x0101;

	unsigned short * msgspot = (unsigned short *) msgbuf;

	if (*msgspot == stun_bind_response) {

		debug("ML: recv_pkg: parse stun message called");

		recv_stun_msg(msgbuf, recvSize);

		return;

	}

	msg_h = (struct msg_header *) msgbuf;

	bufptr += MSG_HEADER_SIZE + msg_h->len_mon_packet_hdr;

	msg_size = recvSize - MSG_HEADER_SIZE - msg_h->len_mon_packet_hdr;

	if(get_Recv_pkt_inf_cb != NULL && msg_h->len_mon_packet_hdr != 0) {

		mon_pkt_inf msginfNow;

		msginfNow.monitoringHeaderLen = msg_h->len_mon_packet_hdr;

		msginfNow.monitoringHeader = msg_h->len_mon_packet_hdr ? &msgbuf[0] + MSG_HEADER_SIZE : NULL;

		//TODO rethink this ...

		if(msg_h->msg_type == ML_CON_MSG) {

			struct conn_msg *c_msg = (struct conn_msg *) bufptr;

			msginfNow.remote_socketID = &(c_msg->sock_id);

		}

		else if(connectbuf[msg_h->remote_con_id] == NULL) {

			error("ML: received pkg called with non existent connection");

			return;

		} else

			msginfNow.remote_socketID = &(connectbuf[msg_h->remote_con_id]->external_socketID);

		msginfNow.buffer = bufptr;

		msginfNow.bufSize = recvSize;

		msginfNow.msgtype = msg_h->msg_type;

		msginfNow.ttl = ttl;

		msginfNow.dataID = msg_h->msg_seq_num;

		msginfNow.offset = msg_h->offset;

		msginfNow.datasize = msg_h->msg_length;

		gettimeofday(&msginfNow.arrival_time, NULL);

		(get_Recv_pkt_inf_cb) ((void *) &msginfNow);

	}

	switch(msg_h->msg_type) {

		case ML_CON_MSG:

			debug("ML: received conn pkg");

			recv_conn_msg(msg_h, bufptr, msg_size);

			break;

		default:

			if(msg_h->msg_type < 127) {

				debug("ML: received data pkg");

				recv_data_msg(msg_h, bufptr, msg_size);

				break;

			}

			debug("ML: unrecognised msg_type");

			break;

	}

}

/*

 * compare the external IP address of two socketIDs

 */

int

compare_external_address_socketIDs(socketID_handle sock1, socketID_handle sock2)

{

	if( sock1->external_addr.udpaddr.sin_addr.s_addr == sock2->external_addr.udpaddr.sin_addr.s_addr)

		return 0;

	return 1;

}

/*

 * the timeout of the NAT traversal

 */

void nat_traversal_timeout(int fd, short event, void *arg)

{

	if (NAT_traversal == false) {

		debug("ML: NAT traversal request resend");

		send_stun_request(socketfd, &stun_server);

		/*

		* enter a NAT traversal timeout that takes care of retransmission

		*/

		struct event *ev1;

		struct timeval timeout_value_NAT_traversal = { 1, 0 };

		ev1 = evtimer_new(base, nat_traversal_timeout, NULL);

		event_add(ev1, &timeout_value_NAT_traversal);

		if(receive_SocketID_cb)

			(receive_SocketID_cb) (&local_socketID, 2);

	}

}

unsigned long resolve(const char *ipaddr)

{

	struct hostent *h = gethostbyname(ipaddr);

	if (!h) {

		error("Unable to resolve hostname %s\n", ipaddr);

	exit(-1);

    }

    unsigned long *addr = (unsigned long *) (h->h_addr);

    return *addr;

}

/*

 * returns a handle to the socketID struct the ipaddr can be a null

 * pointer. Then all available ipaddr on the machine are choosen.

 */

void create_socket(const int port, const char *ipaddr)

{

	struct sockaddr_in udpaddr;

	udpaddr.sin_family = AF_INET;

	if (ipaddr == NULL) {

		/*

		* try to guess the local IP address

		*/

		const char *ipaddr_iface = mlAutodetectIPAddress();

		if (ipaddr_iface) {

			udpaddr.sin_addr.s_addr = inet_addr(ipaddr_iface);

		} else {

			udpaddr.sin_addr.s_addr = INADDR_ANY;

		}

	} else {

		udpaddr.sin_addr.s_addr = inet_addr(ipaddr);

	}

	udpaddr.sin_port = htons(port);

	socketaddrgen udpgen;

	udpgen.udpaddr = udpaddr;

	local_socketID.internal_addr = udpgen;

	socketfd = createSocket(port, ipaddr);

	struct event *ev;

	ev = event_new(base, socketfd, EV_READ | EV_PERSIST, recv_pkg, NULL);

	event_add(ev, NULL);

	if (isStunDefined()) {

		/*

		* send the NAT traversal STUN request

		*/

		 send_stun_request(socketfd, &stun_server);

		/*

		* enter a NAT traversal timeout that takes care of retransmission

		*/

		struct event *ev1;

		struct timeval timeout_value_NAT_traversal = { 2, 0 };

		ev1 = evtimer_new(base, nat_traversal_timeout, NULL);

		event_add(ev1, &timeout_value_NAT_traversal);

		NAT_traversal = false;

	} else {

		/*

		* Assume we have accessibility and copy internal address to external one

		*/

		local_socketID.external_addr = local_socketID.internal_addr;

		NAT_traversal = true; // @TODO: this is not really NAT traversal, but a flag that init is over

		// callback to the upper layer indicating that the socketID is now

		// ready to use

		(receive_SocketID_cb) (&local_socketID, 0);

	}

}

/**************************** END OF INTERNALS ***********************/

void mlInit(bool recv_data_cb,struct timeval timeout_value,const int port,const char *ipaddr,const int stun_port,const char *stun_ipaddr,receive_localsocketID_cb local_socketID_cb,void *arg){

	base = (struct event_base *) arg;

	recv_data_callback = recv_data_cb;

	mlSetRecvTimeout(timeout_value);

	if (stun_ipaddr) {

		 mlSetStunServer(stun_port, stun_ipaddr);

	} else {

	}

	register_recv_localsocketID_cb(local_socketID_cb);

	create_socket(port, ipaddr);

void mlRegisterGetRecvPktInf(get_recv_pkt_inf_cb recv_pkt_inf_cb){

	if (recv_pkt_inf_cb == NULL)

		printf("Register get_recv_pkt_inf_cb failed: NULL ptr  \n");

	else

		get_Recv_pkt_inf_cb = recv_pkt_inf_cb;

	if (send_pkt_inf_cb == NULL)

		printf("Register get_send_pkt_inf_cb: NULL ptr  \n");

	else

		get_Send_pkt_inf_cb = send_pkt_inf_cb;

void mlRegisterSetMonitoringHeaderPktCb(set_monitoring_header_pkt_cb monitoring_header_pkt_cb ){

	if (monitoring_header_pkt_cb == NULL)

		printf("Register set_monitoring_header_pkt_cb: NULL ptr  \n");

	else

		set_Monitoring_header_pkt_cb = monitoring_header_pkt_cb;

void mlRegisterGetRecvDataInf(get_recv_data_inf_cb recv_data_inf_cb){

	if (recv_data_inf_cb == NULL)

		printf("Register get_recv_data_inf_cb: NULL ptr  \n");

	else

		get_Recv_data_inf_cb = recv_data_inf_cb;

void mlRegisterGetSendDataInf(get_send_data_inf_cb  send_data_inf_cb){

	if (send_data_inf_cb == NULL)

		printf("Register get_send_data_inf_cb: NULL ptr  \n");

	else

		get_Send_data_inf_cb = send_data_inf_cb;

void mlRegisterSetMonitoringHeaderDataCb(set_monitoring_header_data_cb monitoring_header_data_cb){

	if (monitoring_header_data_cb == NULL)

		printf("Register set_monitoring_header_data_cb : NULL ptr  \n");

	else

	set_Monitoring_header_data_cb = monitoring_header_data_cb;

void mlRegisterRecvConnectionCb(receive_connection_cb recv_conn_cb){

	if (recv_conn_cb == NULL)

		printf("Register receive_connection_cb: NULL ptr  \n");

	else

		receive_Connection_cb = recv_conn_cb;

void mlRegisterErrorConnectionCb(connection_failed_cb conn_failed){

	if (conn_failed == NULL)

		printf("Register connection_failed_cb: NULL ptr  \n");

	else

	failed_Connection_cb = conn_failed;

void mlRegisterRecvDataCb(receive_data_cb data_cb,unsigned char msgtype){

    if (msgtype > 126) {

	printf

	    ("transmissionHandler: Could not register recv_data callback. Msgtype is greater then 126 \n");

    }

    if (data_cb == NULL) {

	printf("Register receive data callback: NUll ptr \n ");

    } else {

	recvcbbuf[msgtype] = data_cb;

    }

void mlCloseSocket(socketID_handle socketID){

	free(socketID);

int mlOpenConnection(socketID_handle external_socketID,receive_connection_cb connection_cb,void *arg){

	int con_id;

	if (external_socketID == NULL) {

		error("ML: cannot open connection: one of the socketIDs is NULL");

		return -1;

	}

	if (NAT_traversal == false) {

		error("ML: cannot open connection: NAT traversal for socketID still in progress");

		return -1;

	}

	if (connection_cb == NULL) {

		error("ML: cannot open connection: connection_cb is NULL");

		return -1;

	}

	// check if that connection already exist

	con_id = mlConnectionExist(external_socketID, false);

	if (con_id >= 0) {

		// if so check if it is ready to use

		if (connectbuf[con_id]->status == READY) {

				// if so use the callback immidiatley

				(connection_cb) (con_id, arg);

		// otherwise just write the connection cb and the arg pointer

		// into the connection struct

		} else {

			struct receive_connection_cb_list *temp;

			temp = malloc(sizeof(struct receive_connection_cb_list));

			temp->next = NULL;

			temp->connection_cb = connection_cb;

			temp->arg = arg;

			if(connectbuf[con_id]->connection_last != NULL) {

				connectbuf[con_id]->connection_last->next = temp;

				connectbuf[con_id]->connection_last = temp;

			} else

				connectbuf[con_id]->connection_last = connectbuf[con_id]->connection_head = temp;

		}

		return con_id;

	}

	// make entry in connection_establishment array

	for (con_id = 0; con_id < CONNECTBUFSIZE; con_id++) {

		if (connectbuf[con_id] == NULL) {

			connectbuf[con_id] = (connect_data *) malloc(sizeof(connect_data));

			memset(connectbuf[con_id],0,sizeof(connect_data));

			connectbuf[con_id]->starttime = time(NULL);

			memcpy(&connectbuf[con_id]->external_socketID, external_socketID, sizeof(socket_ID));

			connectbuf[con_id]->pmtusize = MAX;

			connectbuf[con_id]->status = INVITE;

			connectbuf[con_id]->seqnr = 0;

			connectbuf[con_id]->internal_connect = !compare_external_address_socketIDs(external_socketID, &local_socketID);

			/*

			* timeout values for the pmtu discovery

			*/

			connectbuf[con_id]->timeout_value.tv_sec = 15;

			connectbuf[con_id]->timeout_value.tv_usec = 0;

			connectbuf[con_id]->connectionID = con_id;

			connectbuf[con_id]->connection_head = connectbuf[con_id]->connection_last = malloc(sizeof(struct receive_connection_cb_list));

			connectbuf[con_id]->connection_last->next = NULL;

			connectbuf[con_id]->connection_last->connection_cb = connection_cb;

			connectbuf[con_id]->connection_last->arg = arg;

			connectbuf[con_id]->external_connectionID = -1;

			break;

		}

	} //end of for

	if (con_id == CONNECTBUFSIZE) {

		error("ML: Could not open connection: connbuffer full");

		return -1;

	}

	// create and send a connection message

	send_conn_msg(con_id, MAX, INVITE);

	struct event *ev;

	ev = evtimer_new(base, pmtu_timeout_cb, (void *) (long)con_id);

	event_add(ev, &connectbuf[con_id]->timeout_value);

	return con_id;

}

void mlCloseConnection(const int connectionID){

	// remove it from the connection array

	if(connectbuf[connectionID]) {

		if(connectbuf[connectionID]->ctrl_msg_buf)

			free(connectbuf[connectionID]->ctrl_msg_buf);

		free(connectbuf[connectionID]);

		connectbuf[connectionID] = NULL;

	}

void mlSendData(const int connectionID,char *sendbuf,int bufsize,unsigned char msgtype,send_params *sParams){

	if (sParams == NULL) {

		error("ML: send data failed: send_params is a NULL ptr");

		return;

	}

	if (connectionID < 0) {

		error("ML: send data failed: connectionID does not exist");

		return;

	}

	if (connectbuf[connectionID] == NULL) {

		error("ML: send data failed: connectionID does not exist");

		return;

	}

	if (connectbuf[connectionID]->status != READY) {

	    error("ML: send data failed: connection is not active");

	    return;

	}

	send_msg(connectionID, msgtype, sendbuf, bufsize, false, sParams);

int mlSendAllData(const int connectionID,send_all_data_container *container,int nr_entries,unsigned char msgtype,send_params *sParams){

    if (nr_entries < 1 || nr_entries > 5) {

	printf

	    ("send_all_data ERROR: nr_enties is not between 1 and 5 \n ");

	return 0;

    } else {

	if (nr_entries == 1) {

		mlSendData(connectionID, container->buffer_1,

		      container->length_1, msgtype, sParams);

	    return 1;

	} else if (nr_entries == 2) {

	    int buflen = container->length_1 + container->length_2;

	    char buf[buflen];

	    memcpy(buf, container->buffer_1, container->length_1);

	    memcpy(&buf[container->length_1], container->buffer_2,

		   container->length_2);

	    mlSendData(connectionID, buf, buflen, msgtype, sParams);

	    return 1;

	} else if (nr_entries == 3) {

	    int buflen =

		container->length_1 + container->length_2 +

		container->length_3;

	    char buf[buflen];

	    memcpy(buf, container->buffer_1, container->length_1);

	    memcpy(&buf[container->length_1], container->buffer_2,

		   container->length_2);

	    memcpy(&buf[container->length_2], container->buffer_3,

		   container->length_3);

	    mlSendData(connectionID, buf, buflen, msgtype, sParams);

	    return 1;

	} else if (nr_entries == 4) {

	    int buflen =

		container->length_1 + container->length_2 +

		container->length_3 + container->length_4;

	    char buf[buflen];

	    memcpy(buf, container->buffer_1, container->length_1);

	    memcpy(&buf[container->length_1], container->buffer_2,

		   container->length_2);

	    memcpy(&buf[container->length_2], container->buffer_3,

		   container->length_3);

	    memcpy(&buf[container->length_3], container->buffer_4,

		   container->length_4);

	    mlSendData(connectionID, buf, buflen, msgtype, sParams);

	    return 1;

	} else {

	    int buflen =

		container->length_1 + container->length_2 +

		container->length_3 + container->length_4 +

		container->length_5;

	    char buf[buflen];

	    memcpy(buf, container->buffer_1, container->length_1);

	    memcpy(&buf[container->length_1], container->buffer_2,

		   container->length_2);

	    memcpy(&buf[container->length_2], container->buffer_3,

		   container->length_3);

	    memcpy(&buf[container->length_3], container->buffer_4,

		   container->length_4);

	    memcpy(&buf[container->length_4], container->buffer_5,

		   container->length_5);

	    mlSendData(connectionID, buf, buflen, msgtype, sParams);

	    return 1;

	}

    }

int mlRecvData(const int connectionID,char *recvbuf,int *bufsize,recv_params *rParams){

	//TODO yet to be converted

	return 0;

#if 0

	if (rParams == NULL) {

		error("ML: recv_data failed: recv_params is a NULL ptr");

		return 0;

    } else {

	printf("transmissionhandler: recv data called \n");

	int i = 0;

	int returnValue = 0;

	double timeout = (double) recv_timeout.tv_sec;

	time_t endtime = time(NULL);

	for (i = 0; i < RECVDATABUFSIZE; i++) {

	    if (recvdatabuf[i] != NULL) {

		if (recvdatabuf[i]->connectionID == connectionID) {

		    printf("transmissionhandler: recv data has entry  \n");

		    double timepass = difftime(endtime, recvdatabuf[i]->starttime);

		    // check if the specified connection has data and it

		    // is complete

		    // check the data seqnr