Man Page mq_notify :
1) Уведомление о сообщении происходит только тогда, когда приходит новое сообщение и очередь ранее была пуста.Если очередь не была пуста во время вызова mq_notify (), то уведомление будет появляться только после того, как очередь очищена и появится новое сообщение.
2) Уведомление происходит один раз: после доставки уведомления регистрация уведомления удаляется, и другой процесс может зарегистрироваться для уведомления о сообщении.Если уведомленный процесс желает получить следующее уведомление, он может использовать mq_notify () для запроса дополнительного уведомления.
3) Для получения уведомления из очереди сообщений может быть зарегистрирован только один процесс.
Итак:
1) Очистить MsgQ для немедленного получения новых сообщений в процессе чтенияпосле регистрации в mq_notify ().
2) Перерегистрировать в функции уведомления для получения следующего сообщения.
3) Зарегистрировать только 1 процесс для получения сообщений от Q.
Вот простой код читателя очереди сообщений на C ++:
#include <iostream>
#include <mqueue.h>
#include <string.h>
#include <sstream>
#include <unistd.h>
#include <errno.h>
using namespace std;
#define MSG_Q_NAME "/MY_MSGQ_3"
static void /* Thread start function */
tfunc(union sigval sv)
{
mqd_t msq_id = *(static_cast<mqd_t*>(sv.sival_ptr));
struct mq_attr attr;
if(mq_getattr(msq_id, &attr) < 0)
{
cout << "Error in mq_getattr " << strerror(errno) << endl;
return;
}
// Reregister for new messages on Q
struct sigevent sev;
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = tfunc;
sev.sigev_notify_attributes = NULL;
sev.sigev_value.sival_ptr = sv.sival_ptr;
if (mq_notify(msq_id, &sev) < 0)
{
cout << "Error during Reregister in msq_notify : "
<< strerror(errno) << endl;
exit(EXIT_FAILURE);
}
// Read new message on the Q
char* arr = new char[attr.mq_msgsize];
memset(arr, 0, attr.mq_msgsize);
if(mq_receive(msq_id, arr, attr.mq_msgsize, 0) < 0)
{
if(errno != EAGAIN)
{
cout << "Error in mq_receive " << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
}
else
{
cout << "Msg rcvd " << arr << endl;
}
}
int main()
{
mqd_t msq_id = mq_open(MSG_Q_NAME, O_RDONLY | O_NONBLOCK | O_CREAT, 0666, 0);
if(msq_id == (mqd_t) -1)
{
cout << "Error on msg Q creation: " << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
// The process is registered for notification for new message on the Q
struct sigevent sev;
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = tfunc;
sev.sigev_notify_attributes = NULL;
sev.sigev_value.sival_ptr = &msq_id;
if (mq_notify(msq_id, &sev) < 0)
{
cout << "Error on msg Q notify : " << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
else
{
cout << "Notify for msg Q reception " << MSG_Q_NAME << endl;
}
// Man Page mq_notify: Message notification occurs only when a new
// message arrives and the queue was previously empty. If the queue was
// not empty at the time mq_notify() was called, then a notification will
// occur only after the queue is emptied and a new message arrives.
//
// So emptying the Q to recv new messages
ssize_t n = 0;
struct mq_attr attr;
if(mq_getattr(msq_id, &attr) < 0)
{
cout << "Error in mq_getattr " << strerror(errno) << endl;
exit(EXIT_FAILURE);
}
char* arr = new char[attr.mq_msgsize];
memset(arr, 0, attr.mq_msgsize);
while((n = mq_receive(msq_id, arr, attr.mq_msgsize, 0) >= 0))
{
cout << "Empty the Q. Msg rcvd " << arr << endl;
}
while(1)
;
mq_close(msq_id);
}
Вот простой код писателя Q:
#include <iostream>
#include <mqueue.h>
#include <string.h>
#include <sstream>
#include <unistd.h>
#include <errno.h>
#include <sys/stat.h>
using namespace std;
#define MSG_Q_NAME "/MY_MSGQ_3"
int main()
{
struct mq_attr attr;
memset(&attr, 0, sizeof attr);
attr.mq_msgsize = 8192;
attr.mq_flags = 0;
attr.mq_maxmsg = 10;
mqd_t msq_id = mq_open(MSG_Q_NAME, O_RDWR | O_CREAT | O_NONBLOCK,
0777, &attr);
if(msq_id == (mqd_t) -1)
{
cout << "Error on msg Q creation: " << strerror(errno) << endl;
exit(1);
}
// Write 5 msgs on message Q
for(int i = 0; i < 5; ++i)
{
stringstream s;
s << "My Msg " << i;
if(mq_send(msq_id, s.str().c_str(), strlen(s.str().c_str()), 0) < 0)
{
if(errno != EAGAIN)
{
cout << "Error on sending msg on MsgQ " << strerror(errno);
mq_close(msq_id);
exit(1);
}
}
else
{
cout << "Sent msg " << s.str() << endl;
}
sleep(1); // Easily see the received message in reader
}
mq_close(msq_id);
}