msg_queue.c

linux下的消息队列的一些函数封装,包括消息队列的创建,发送消息,接受消息,清除消息等.

#include <include.h>

#include "timer.h"
#include "msg_queue.h"
#include "common.h"
#include "error.h"
#include "udplog.h"


int OpenMsgQueue (char * path)
{
	key_t	key;
	int		msgId;

	key = ftok (path, 'm');

	msgId = msgget (key, IPC_CREAT|0660);
	if (msgId < 0)
	{
		return -1;
	}
	return msgId;
}

/*
send a msg to ms queue
msgId, 	msg queue id
buf		send data buffer
len		send data length
return value:
	0	success
	-1	error
*/
int SendMsg (int msgId, long type, char *buf, int len)
{
	int	rlt;
	int	i;
	T_MsgBuf		msgBuf;

//udplog ("Send Type = %d", type);
	msgBuf.mtype = type;
	memcpy (msgBuf.mtext	, buf, len);

	for (i=0; i<SEND_RETRY; i++)
	{
		rlt = msgsnd (msgId, (struct msgbuf *)&msgBuf, len, IPC_NOWAIT);
		if (rlt == 0)
			return 0;
		else
		{
			if (errno == EAGAIN)		//msg queue is full, wait
			{
				usleep (10*1000);		//wait 10ms to retry
				continue;
			}
			else
				return -1;
		}
	}
	return -1;
}

/*
msgId	msg queue's ID
mtype	will receive msg's type
return value	0	has a msg in the msg queue
			-1 	no msg in it
*/
int IsMsgReady (int msgId, long mtype)
{
	if (msgrcv (msgId, NULL, 0, mtype, IPC_NOWAIT) == -1)
	{
		if (errno == E2BIG)
			return 0;
	}
	return -1;
}

/*
msgId	msg queue's ID
mtype	will receive msg's type
buf		the recieve data store buffer
length	want to receive length;
return  value:	receive length
			-1 error
		
*/
int ReceiveMsg (int msgId, long type, char * buf, int length)
{
	int rlt;
	T_MsgBuf		msgBuf;

//udplog ("Read Type = %d", type);

	if (IsMsgReady (msgId, type) < 0)
	{
		return 0;
	}

	msgBuf.mtype = type;
	
	rlt = msgrcv (msgId, (struct msgbuf *)&msgBuf, length, type, IPC_NOWAIT|MSG_NOERROR);

	if (rlt > 0)
	{
		memcpy (buf, msgBuf.mtext, rlt);
	}

	return rlt;
}

int SendAndReceiveMsg (int SendId, int RecvId, char *SendBuffer, int SendLen, int sendType, 
						char *ReceiveBuffer, int ReceiveLen, int ReceiveType, int TimeOut)
{
	T_Timer	time;
	int 	len;
	
	if (SendMsg (SendId, (long) sendType, SendBuffer, SendLen) < 0)
	{
	udplog_mem ("[%d]Send Msg Cmd[%d]=", SendBuffer, SendLen, getpid(), SendLen);
		udplog ("Send msg error:%s", strerror(errno));
		return E_SEND_MSG_FAILED;
	}
	if (TimeOut == 0)
		return 0;
	
	StartMsecTimer(&time, TimeOut);
	while (1)
	{
		if ( CheckTimeOut(&time))
		{
			udplog ("Receive msg time out");
			return E_READ_MSG_TIMEOUT;
		}
		len = ReceiveMsg (RecvId, ReceiveType, ReceiveBuffer, ReceiveLen);
		
		if (len < 0)
		{
			udplog ("Receive msg error:%s", strerror(errno));
			return E_READ_MSG_FAILED;
		}
		else if (len > 0)
		{
			break;
		}
		usleep(1000);
	}
//	udplog_mem ("[%d]Read Msg Cmd[%d]=", ReceiveBuffer, len, ReceiveType, len);
	return  len;
}

int ClearMsg (int msgId, int type)
{
	char sbuf[1000];

	if (type != 0)
		while(ReceiveMsg (msgId, type, sbuf, sizeof(sbuf)) > 0);
	return 0;
}