rtes_messenger/src/core.c

/*!
 * \file core.c
 *
 * \author Christos Choutouridis AEM:8997 <cchoutou@ece.auth.gr>
 */


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <pthread.h>

#include "core.h"

pthread_mutex_t lock_msgList;

//! Helper API
//! @{

#define _ADHOC_SUBNET(A, B, C, D)   (((A)<<24) | ((B)<<16) | ((C)<<8) | (D))

device_t addr2device (uint32_t in_addr) {
   device_t dev = {
      .id   = (in_addr & 0x000000FF) + ((in_addr >> 8) & 0x000000FF) * 100,
      .next = NULL
   };
   return dev;
}

uint32_t device2addr (const device_t* dev) {
   uint32_t add  = _adhoc_subnet;
            add |= (dev->id % 100) & 0x000000FF;
            add |= ((dev->id / 100) & 0x000000FF) << 8;
   return add;
}

device_t ip2device (devIP_t* ip) {
   device_t dev = {
      .id = ip->C*100 + ip->D,
      .next = NULL
   };
   return dev;
}

devIP_t device2ip (const device_t* dev) {
   devIP_t ip = {
      .A=0, .B=0, .C=dev->id/100, .D=dev->id%100
   };
   return ip;
}

devIP_t addr2ip (uint32_t in_addr) {
   devIP_t ip = {
      .A=0, .B=0,
      .C=(in_addr >> 8) & 0x000000FF,
      .D=in_addr & 0x000000FF
   };
   return ip;
}
//! @}


//! cMsg_t API
//! @{

/*!
 * Parse an incoming message
 *
 * @param cMsg       Pointer to cMsg object to store the parsed data
 * @param rawMsg     Pointer to raw message
 * @param size       The size f raw message buffer
 * @return           The status of the operation
 *    @arg  MSG_OK      Success
 *    @arg  MSG_ERROR   Parse failure, incoming message format error
 */
status_t cMsg_parse (cMsg_t* cMsg, char_t* rawMsg, size_t size) {

   // Check message integrity
   if (size > MSG_TEXT_SIZE)
      return MSG_ERROR;
   int d =0;
   for (size_t i =0; rawMsg[i] && i<size; ++i) {
      d += (rawMsg[i] == MSG_DELIMITER) ? 1:0;
   }
   if (d != 3)
      return MSG_ERROR;

   // Get message
   strcpy(cMsg->msg, rawMsg);

   // Parse message
   char_t del[2] = {MSG_DELIMITER, '\0'};
   char_t* tok;

   tok = strtok (cMsg->msg, del);
   cMsg->from = atoi (tok);

   tok = strtok(NULL, del);
   cMsg->to = atoi (tok);

   tok = strtok(NULL, del);
   cMsg->ts = atoll (tok);

   tok = strtok(NULL, del);
   cMsg->text = tok - cMsg->msg;

   return MSG_OK;
}

/*! getter for cMsg_t member fromAEM */
uint32_t cMsg_getFrom(cMsg_t* cMsg) { return cMsg->from; }
/*! getter for cMsg_t member toAEM */
uint32_t cMsg_getTo(cMsg_t* cMsg)   { return cMsg->to; }
/*! getter for cMsg_t member fromAEM */
uint64_t cMsg_getTs(cMsg_t* cMsg)   { return cMsg->ts; }
/*! getter for payload text member */
char_t*  cMsg_getText(cMsg_t* cMsg) { return (char_t*)& cMsg->msg[cMsg->text]; }

/*!
 * Predicate to check core message equality
 * @param m1   Pointer to message 1
 * @param m2   Pointer to message 2
 * @return     Equality result (true, false)
 */
bool cMsg_equal (cMsg_t* m1, cMsg_t* m2) {
   if (m1->from != m2->from)  return false;
   if (m1->to != m2->to)      return false;
   if (m1->ts != m2->ts)      return false;
   if (strncmp (cMsg_getText(m1), cMsg_getText(m2), sizeof(m1->msg)))
      return false;
   return true;
}
//! @}

/*!
 * Create a message list for our application.
 */
msgList_t msgList;


//! msgList API
//! @{

/*! Macro helper to saturate increased values */
#define _top_saturate(test, apply, value) do { \
   if (test >= value)   apply = value; \
} while (0)

/*! Macro helper to saturate decreased values */
#define _btm_saturate(test, apply, value) do { \
   if (test < value)   apply = value; \
while (0)

/*!
 * @brief   msgList iterator pre-increment in the msg_t direction
 *
 * This iterator force a ring buffer behavior. This function takes pointer
 * to iterator to alter but return the altered value so it can be directly
 * used in expressions
 *
 * @param it   Pointer to iterator to increase
 * @return     The iterator values
 */
static mIter_t _preInc(mIter_t* it) {
   if (++*it >= MSG_LIST_SIZE)  *it = 0;
   return *it;
}

/*!
 * @brief   msgList iterator pre-decrement in the msg_t direction
 *
 * This iterator force a ring buffer behavior. This function takes pointer
 * to iterator to alter but return the altered value so it can be directly
 * used in expressions
 *
 * @param it   Pointer to iterator to decrease
 * @return     The iterator values
 */
static mIter_t _preDec(mIter_t* it) {
   if (--*it < 0)  *it = MSG_LIST_SIZE;
   return *it;
}

/*!
 * @brief Access devices in the device_t direction.
 *
 * This function returns the first device on the recipient list if there is one.
 *
 * @param this    The msgList object to work with
 * @param it      The iterator value [msg_t direction]
 * @return        Pointer to first device on the list, or NULL if the list is empty
 */
static device_t* devList_get (msgList_t* this, mIter_t it) {
   device_t* d = this->m[it].recipients;
   return (d) ? d->next : NULL;
}

/*!
 * Iterate through device list.
 *
 * This function return the next device on the recipient list if there is one.
 * [device direction]
 *
 * @param d    Pointer to current device
 * @return     Pointer to next device on the list, or NULL if the list is empty
 */
//static device_t* devList_getNext (device_t* d) {
//   return (d) ? d->next : NULL;
//}

/*!
 * @brief   Adds a new device on the recipients list
 *
 * @param this    The msgList object to work with
 * @param it      The iterator value [msg_t direction]
 * @param dev     Pointer to device to add [device direction]
 * @return        The status of the operation
 *    @arg  MSG_OK      Success
 *    @arg  MSG_ERROR   memory allocation error
 */
status_t devList_add (msgList_t* this, mIter_t it, device_t* dev) {
   pthread_mutex_lock (&lock_msgList);
   device_t* last = devList_get (this, it);
   while (last && last->next)
      last = last->next;
   last = (device_t*)malloc (sizeof(device_t));
   *last = *dev;
   pthread_mutex_unlock (&lock_msgList);
   return (last) ? MSG_OK : MSG_ERROR;
}

/*!
 * Frees up all allocated memory in a device list.
 *
 * @param this    The msgList object to work with
 * @param it      The iterator value [msg_t direction]
 */
static void devList_free (msgList_t* this, mIter_t it) {
   device_t* last = devList_get (this, it);
   while (last) {
      device_t* next = last->next;
      free(last);
      last = next;
   }
}

status_t msgList_init (msgList_t* msgList) {
   if (pthread_mutex_init(&lock_msgList, NULL) != 0) {
       fprintf (stderr, "Error %s: mutex init has failed\n", __FUNCTION__ );
       return MSG_ERROR;
   }
   memset((void*)msgList, 0, sizeof(msgList_t));
   msgList->last =-1;
   return MSG_OK;
}

/*!
 * Searches for a message in the message list.
 *
 * @param this    The msgList object to work with
 * @param msg     Pointer to message to search
 * @return        Iterator to message if found, or -1 if not
 */
mIter_t msgList_find (msgList_t* this, msg_t* msg) {
   pthread_mutex_lock (&lock_msgList);
   mIter_t it =this->last;    // get iterator
   // search from end to start to find msg, return on success
   for (size_t i=0 ; i < this->size ; ++i) {
      if (cMsg_equal (&this->m[it].cMsg, &msg->cMsg)) {
         pthread_mutex_unlock (&lock_msgList);
         return it;
      }
      _preDec(&it);
      // We start from the end as we think, its more possible
      // to find msg in the recent messages.
   }
   pthread_mutex_unlock (&lock_msgList);
   return (mIter_t)-1;     // fail to find
}

/*!
 * Add a new message in the message list
 *
 * @param this    The msgList object to work with
 * @param msg     Pointer to message
 */
void msgList_add (msgList_t* this, msg_t* msg) {
   pthread_mutex_lock (&lock_msgList);
   devList_free (this, _preInc(&this->last));               // free up possible previous recipients list
   this->m[this->last] = *msg;                              // store data
   pthread_mutex_unlock (&lock_msgList);
   _top_saturate(++this->size, this->size, MSG_LIST_SIZE);  // count the items
}

//! @}



static char_t* _frm_msg_io  = "dev=%d, message: from=%d, to=%d, timestamp=%lld, text=%s";
static char_t* _frm_msg_new = "new message: from=%d, to=%d, timestamp=%lld, text=%s";

pthread_mutex_t lock_stderr;
pthread_mutex_t lock_stdout;


status_t log_init (void) {
   if (pthread_mutex_init(&lock_stderr, NULL) != 0) {
       fprintf (stderr, "Error %s: mutex init has failed\n", __FUNCTION__ );
       return MSG_ERROR;
   }
   if (pthread_mutex_init(&lock_stdout, NULL) != 0) {
      fprintf (stderr, "Error %s: mutex init has failed\n", __FUNCTION__ );
      return MSG_ERROR;
   }
   return MSG_OK;
}

void log_msg_io (msg_t* msg) {
   if (settings.outLevel >= OUTLEVEL_1) {
      char_t head[16];
      strncpy (head, cMsg_getText(&msg->cMsg), sizeof(head));
      head[16] = 0;
      pthread_mutex_lock(&lock_stdout);
      fprintf (stdout, _frm_msg_io,
         msg->sender.id,
         cMsg_getFrom (&msg->cMsg),
         cMsg_getTo (&msg->cMsg),
         cMsg_getTs (&msg->cMsg),
         head
      );
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
   }
}

void log_msg_new (msg_t* msg) {
   if (settings.outLevel >= OUTLEVEL_1) {
      char_t head[16];
      strncpy (head, cMsg_getText(&msg->cMsg), sizeof(head));
      head[16] = 0;
      pthread_mutex_lock(&lock_stdout);
      fprintf (stdout, _frm_msg_new,
         cMsg_getFrom (&msg->cMsg),
         cMsg_getTo (&msg->cMsg),
         cMsg_getTs (&msg->cMsg),
         head
      );
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
   }
}

void log_debug (const char *fmt, ...) {
   if (settings.outLevel >= OUTLEVEL_2) {
      va_list ap;
      va_start(ap, fmt);
      pthread_mutex_lock(&lock_stdout);
      vfprintf (stdout, fmt, ap);
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
      va_end(ap);
   }
}

void log_error (const char *fmt, ...) {
   va_list ap;
   va_start(ap, fmt);
   pthread_mutex_lock(&lock_stderr);
   vfprintf (stderr, fmt, ap);
   fflush(stderr);
   pthread_mutex_unlock(&lock_stderr);
   va_end(ap);
}