rtes_messenger/src/listener.c

/*!
 * \file    listener.c
 * \brief   Listener related functionality
 *
 * \author Christos Choutouridis AEM:8997 <cchoutou@ece.auth.gr>
 */

#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>

#include "listener.h"

char_t rx_buffer[4*MSG_TEXT_SIZE];     //!< receive buffer

/*!
 * Incoming message handling
 * @param   dev      Pointer to sender/client's device
 * @param   buffer   Buffer to message
 * @param   size     Size of message
 */
static void listen_handler (devAEM_t dev, char_t* buffer, size_t size) {
   msg_t msg;
   msg_init (&msg);
   msg.sender = dev;
   if (cMsg_parse (&msg.cMsg, buffer, size) == MSG_ERROR)   // parse the message
      return;
   log_debug("Debug: Message parsed\n");

   msgList_acquire ();
   mIter_t myCopy = msgList_find (&msgList, &msg);          // try to find message in msgList
   if (myCopy == -1) {
   // We don't have a copy, accept and store it
      mIter_t slot = msgList_add (&msgList, &msg);
      msgList_release ();
      statsUpdateIn (&msg, false);  // message process
      log_msg_in (&msg);            // log

      // Processing...
      devList_acquire();
      dIter_t d = devList_getIter (dev);
      dIter_t f = devList_getIter (msg.cMsg.from);
      devList_release();

      msgList_acquire ();
      // Do not echo message to sender, he already has it
      msgList.m[slot].recipients[d] = true;
      // don't push back message to creator, he already has it
      msgList.m[slot].recipients[f] = true;
      msgList_release ();
   }
   else {
   // We have a copy
      msgList_release ();
      statsUpdateIn (&msg, true);   // message process
      log_debug ("Debug: Duplicate message from: %d\n", msg.sender);
   }
}

/*!
 * Listener. This function normally never returns. If it does it is to
 * report unrecoverable error.
 * @return  Error creating server side socket
 */
static status_t listener(void) {
   int srvSock;
   sockaddr_in_t srvAddPort;

   // Create socket
   memset(&srvAddPort, 0, sizeof(srvAddPort));
   srvAddPort.sin_family= AF_INET;
   srvAddPort.sin_port= htons(settings.port);
   srvAddPort.sin_addr.s_addr = htonl(INADDR_ANY);

   // try to create tcp socket
   if ((srvSock = socket (PF_INET, SOCK_STREAM, 0)) == -1) {
      log_error ("Error: Can not create socket\n");
      goto _list_error1;
   }
   log_debug("Debug: Socket for listening created\n");

   // Try to bind socket
   if(bind(srvSock, (sockaddr_t *) &srvAddPort, sizeof(srvAddPort)) == -1) {
      log_error ("Error: Can not bind socket to port %d\n", settings.port);
      goto _list_error0;
   }
   // try to setup listener. We use DEVICE_LIST_SIZE for pending request
   if (listen (srvSock, DEVICE_LIST_SIZE) == -1) {
      log_error ("Error: Can not enable socket\n");
      goto _list_error0;
   }
   log_debug("Debug: Listening on [0.0.0.0], port %d\n", settings.port);

   while (1) {
      sockaddr_in_t clntAddr;
      socklen_t clntLen= sizeof(clntAddr);
      int clntSock;
      int rcvBytes;
      // block in accept and get client's connection socket
      if ((clntSock = accept (srvSock, (sockaddr_t *)&clntAddr, &clntLen)) == -1) {
         continue;
      }
      // make a device from clients address
      devAEM_t dev = addr2devAEM (ntohl(clntAddr.sin_addr.s_addr));
      devIP_t ip  = AEM2ip (dev);
      log_debug (
         "Debug: Connection from %u.%u.%u.%u port:%u received\n",
         ip.A, ip.B, ip.C, ip.D, clntAddr.sin_port
      );

      // Block in receive, after the connection is accepted
      if ((rcvBytes = recv(clntSock, rx_buffer, sizeof(rx_buffer), 0)) < 0) {
         log_error ("Error: Fail to receive from socket\n");
         close (clntSock);
         continue;
      }
      rx_buffer[rcvBytes] = '\0';   // add null termination to buffer
      close(clntSock);              // close the client socket

      // call listen handler for the device and message
      log_debug (
         "Debug: Message from %u.%u.%u.%u port:%u received\n",
         ip.A, ip.B, ip.C, ip.D, clntAddr.sin_port
      );
      listen_handler (dev, rx_buffer, rcvBytes);
   }

   // Local error return handling
_list_error0:
   close(srvSock);      // server socket clean up
   log_debug ("Debug: Socket for listening closed\n");
_list_error1:
   return MSG_ERROR;    // report back
}


/*!
 * pthread listener wrapper
 * @param ptr  Not used
 */
void* pthListener(void* ptr) {
   (void)&ptr;    // use parameter
   if (listener() == MSG_ERROR)
      exit(1);    // this is a deal breaker sorry.
   return NULL;
}