examples/PIPS/opencmessagequeuelibraryex/engine/src/msgqcreate.c

#include <errno.h>
#include <sys/msg.h>
#include <sys/sem.h>
#include "MsgQInternal.h"


/********************************************************************************
* MsgQCreate
* Description: Creates a message queue
* Inputs:
*  qName   : ULONG that represents the queue name
*  maxMsgs : ULONG that represents maximum size of messages 
*********************************************************************************/
EXPORT_C int MsgQCreate(ULONG qName, ULONG maxMsgs, ULONG qOptions, int* err) {
        int         qId;
        int         semId;
        int         semName ;
        struct msqid_ds  qStatus ;
        struct semid_ds  sStatus ;
        
        union semun {
                int              val;
                struct semid_ds* buf;
                ushort_t*                array;
        } arg ;


        int          hashInstErr;

        if (MsgQTableLookup(qName) != NULL) {
                *err = OK;
                return (OK);
        }
        else {
                if((qOptions == MSG_Q_FIFO) || (qOptions == MSG_Q_PRIORITY)) {
                        // Set msg queue options to  FIFO and create the message queue
                        qOptions= MSG_Q_FIFO ;
                        if((qId = msgget((key_t) qName ,IPC_CREAT | 0666 | IPC_EXCL )) >=0 ) {
                                //  set msg queue parameter max # bytes in queue
                                if(  msgctl(qId,IPC_STAT,&qStatus) == 0  )          
                                if( qStatus.msg_qbytes > (maxMsgs * MAX_MSG_LEN) ) {
                                        qStatus.msg_qbytes = maxMsgs * MAX_MSG_LEN ;              
                                        if( msgctl(qId,IPC_SET,&qStatus) < 0) {
                                                // delete message queue on error 
                                                msgctl(qId,IPC_RMID,0);
                                                *err = errno;
                                                return(ERROR);
                                        }
                                }
                                // create semaphore
                                semName= (key_t) qName;
                                if((semId = semget(semName, 1, IPC_CREAT | 0666 |IPC_EXCL )) >= 0 ) {
                                        // set the semaphore value
                                        arg.buf = &sStatus;              
                                        arg.val = 1;
                                        semctl(semId , 0, SETVAL, arg) ;
                                        
                                        //install queue data in hash table
                                        if(InstallMsqQTable(qName, qId, semId, &hashInstErr) == OK) {
                                                AddToMsgQTable(qName);
                                                *err = OK;                
                                                return (OK);
                                        }
                                        else {
                                                //delete semaphore on error
                                                semctl(semId,0,IPC_RMID,0) ;
                                                *err = hashInstErr;
                                        }
                                }
                                else {
                                        // delete message queue on error
                                        msgctl(qId,IPC_RMID,0);
                                        *err = errno;
                                }
                        }
                        else {
                                *err = errno;
                        }       
                }
                else
                        *err = KMsgQLibParamErr;

                return(ERROR);
        }
}

---