Using POSIX Message Queues

Code 

/* Includes */
#include < pthread.h >
#include < stdio.h >
#include < stdlib.h >
#include < unistd.h >
#include < mqueue.h >
#include < sys/stat.h >
#include < errno.h >
#include < string.h >


/* Threads */
static void *MainThread (void *);
static void *AnotherThread (void *);

/* Defines */
#define MAIN_QNAME "/MainQueue"
pthread_mutex_t wait_mutex;

int mq_test(void)
{
 pthread_t mainThread, anotherThread;

  
 printf ("Creating threads .. \n");
 pthread_create (&mainThread,    NULL, &MainThread,    NULL);

 pthread_create (&anotherThread, NULL, &AnotherThread, NULL);

 pthread_mutex_init (&wait_mutex, NULL);

 
 pthread_join (mainThread,    NULL);
 pthread_join (anotherThread, NULL);

 
 return 1;
}

#define BUFFER_SIZE 10000

/* Main thread .. Waiting for messages */
static void *MainThread (void *args)
{

 mqd_t queue_handle;
 char buffer[BUFFER_SIZE];
 int bytes_read;

 struct mq_attr msgq_attr;
 unsigned int sender;
 
 printf ("[MainThread] Inside main thread \n");

 
 // Let the other thread wait till I am ready! 
 pthread_mutex_lock (&wait_mutex);
 
 // Clear the buffer
 memset (buffer, 0, BUFFER_SIZE);

 
 // Detach the thread
 pthread_detach (pthread_self());
 
 // unlink the queue if it exisits - debug
 mq_unlink (MAIN_QNAME);

 
 printf ("[MainThread]Opening MQ \n"); 
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR | O_CREAT, S_IRWXU | S_IRWXG, NULL);

 if (queue_handle == -1)
 {
  perror ("[MainThread] Error Opening MQ: ");
  return 0;
 } 
 
 printf ("[MainThread] Waiting for messages ... \n"); 
 pthread_mutex_unlock (&wait_mutex);

 
 for (; ;)
 {
  bytes_read = mq_receive(queue_handle, buffer, BUFFER_SIZE, &sender);

  if (bytes_read == -1)
  {
   perror("[MainThread] Failed to recieve:");
   return 0;
  }

  else
  {
   printf ("[MainThread] Data: %s \n", buffer);   
   // Get the MQ attributes
   mq_getattr(queue_handle, &msgq_attr);
   printf("[MainThread] Queue \"%s\":\n"
      "\t- stores at most %ld messages\n"
      "\t- large at most %ld bytes each\n"
      "\t- currently holds %ld messages\n",
       MAIN_QNAME, 
       msgq_attr.mq_maxmsg, 
       msgq_attr.mq_msgsize, 
       msgq_attr.mq_curmsgs);   

   // Clear buffer and sleep to block for some time t and see 
   // if you get all the messages!

   memset(buffer, 0, BUFFER_SIZE); 
   sleep(5);
  }  
 } 

 mq_close (queue_handle);
}

#define MAX_SEND_BUFFER 70 

static void *AnotherThread (void *args)
{

 mqd_t queue_handle;
 char buffer[MAX_SEND_BUFFER];
 unsigned int msgprio = 1;

 int count;
 
 printf ("[AnotherThread] Inside Another thread \n");
 
 pthread_mutex_lock (&wait_mutex);

 
 queue_handle=  mq_open(MAIN_QNAME, O_RDWR);
 if (queue_handle == -1)
 {

  perror ("[AnotherThread] Error Opening MQ:");
  return 0;
 }
 
 for (count = 0; count < 100; count++)
 {

  snprintf (buffer, MAX_SEND_BUFFER, "Message %d from Another thread", count);

  if (0 != mq_send (queue_handle, buffer, strlen(buffer)+1, msgprio))
  {

   perror ("[AnotherThread] Sending:");
   mq_close (queue_handle); 
   pthread_mutex_unlock (&wait_mutex);

   return 0;
  }
 } 
 
 pthread_mutex_unlock (&wait_mutex); 
 mq_close (queue_handle); 
 
 
 return 0;
}

#ifdef _DEBUG_
int main (void)
{
 return mq_test(void);
}

#endif
on