agsMSMQ v1.1 - A Message Queue Implementation

Agus Kurniawan

4.80/5 (5 votes)

Dec 20, 2001

2 min read

131684

3535

Class wrapper for Message Queue (MSMQ)

Introduction

This class wrapper to encapsulate Message Queue.

Features
API reference
Usage in application

Features

Compatible with class MFC
Ease and simple to use
Ease to track error

API Reference

This class agsMSMQ has features in header file like here :

 struct MQmsg{
    CString label;
    CString msg;
 };

 class agsMSMQ  
 {
 public:
    BOOL closeMSMQ();
    MQmsg getMessageQ();
    int getTotalMessage();
    HRESULT getMSMQmsg();
    BOOL PrepapeGetMessage(CString srv,CString path,int modeMSMQ);
    BOOL SendPrivateQ(CString serv, CString pathmsg,CString labelmsg,CString msg);
    BOOL SendPublicQ(CString serv, CString pathmsg,CString labelmsg,CString msg);
    BOOL DeletePrivateQ(CString serv, CString pathmsg);
    BOOL DeletePublicQ(CString serv, CString pathmsg);
    BOOL CreatePrivateQ(CString serv, CString pathmsg,CString label);
    BOOL CreatePublicQ(CString serv, CString pathmsg,CString label);
    CString getErrorMsg();
    agsMSMQ();
    virtual ~agsMSMQ();

 private:
    void InitialMQMSG();
    void InitialPropidBody(CString label,CString msg);
    void InitialPropidQ();
    CString m_sError;
    void InitialMQQueue();
    DWORD            dwIDprop;
    DWORD            dwBufferLength;
    WCHAR            wszBuffer[250];
    CString            m_sPathQ;
    CString            m_sMSMQserv;
    CString            m_sLabelQ;
    unsigned short        wszPath[256];
    unsigned short        wszLabel[1024];
    DWORD            dwDestFormatLength;
    WCHAR            wszDestFormat[256];
    unsigned char        pMessage[2048];
    DWORD            dwAccessMode;
    DWORD            dwShareMode;
    DWORD            dwReaction; 
    int            m_nMessage;
    MQmsg            msgQ;
    

 protected:
    MQQUEUEPROPS        QProps;
    MQMSGPROPS        QPropMsg;
    MQPROPVARIANT        QPropVar[NUMBER_OF_PROPERTIES];    
    QUEUEPROPID        QPropID[NUMBER_OF_PROPERTIES];
    MSGPROPID        QMsg[NUMBER_OF_PROPERTIES];
    HRESULT            QStatus[NUMBER_OF_PROPERTIES];
    HRESULT            hr;
    HANDLE            hrTrack;
    HANDLE            hQue;
    PSECURITY_DESCRIPTOR    pSecurity;
    
 };

Here's explanation of functions :

BOOL CreatePublicQ(CString serv, CString pathmsg,CString label)Use to create public queue on server queue serv with name of queue pathmsg and label queue label

BOOL CreatePrivateQ(CString serv, CString pathmsg,CString label) Use to create private queue on server queue serv with name of queue pathmsg and label queue label.

BOOL SendPublicQ(CString serv, CString pathmsg,CString labelmsg,CString msg); Use to send public message queue on serv queue server, pathmsg path queue with message label labelmsg and content message msg.

BOOL SendPrivateQ(CString serv, CString pathmsg,CString labelmsg,CString msg); Use to send private message queue on serv queue server, pathmsg path queue with message label labelmsg and content message msg.

BOOL PrepapeGetMessage(CString srv,CString path,int modeMSMQ); Prepare to get message queue on serv with path queue path, modeMSMQ parameter identify type of queue. if value of modeMSMQ parameter is 0 then it's public queue. Otherwise, if value of modeMSMQ is 1, then it's private queue.

MQmsg getMessageQ(void)
Get data queue message. MQmsg is a structure :
```
 struct MQmsg{
    CString label;
    CString msg;
 };
```

int getTotalMessage(void)Get number of messages.

HRESULT getMSMQmsg(void)Use to check error or not while it's getting queue message.

BOOL closeMSMQ(void)Use to close queue server.

CString getErrorMsg(void)Use to get error message.

Usage in Application

This's GUI to implement agsMSMQ class:

Don't forget to put header file (#include "agsMSMQ.h") on the top of your application. Beside that, you must add library (mqrt.lib) into your project like this picture below:

Create public queue:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.CreatePublicQ(m_sServer,m_sPath,m_sLabel))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Create Public Queue succesfully");

Create private queue:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.CreatePrivateQ(m_sServer,m_sPath,m_sLabel))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Create Private Queue succesfully");

Delete public queue:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.DeletePublicQ(m_sServer,m_sPath))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Delete Public Queue succesfully");

Delete private queue:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.DeletePrivateQ(m_sServer,m_sPath))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Delete Private Queue succesfully");

Send public queue message:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.SendPublicQ(m_sServer,m_sPath,m_sMessageLabel,m_sMessage))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Send Public Message succesfully");

Send private queue message:

agsMSMQ mque;    

UpdateData();
CWaitCursor wait;

if(!mque.SendPrivateQ(m_sServer,m_sPath,m_sMessageLabel,m_sMessage))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

AfxMessageBox("Send Public Message succesfully");

Get all public queue message:

agsMSMQ mque;    
int current;
MQmsg msg;
HRESULT hr;

UpdateData();
CWaitCursor wait;
m_cMessage.DeleteAllItems();
if(!mque.PrepareGetMessage(m_sServer,m_sPath,0))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

msg    = mque.getMessageQ();
current = m_cMessage.InsertItem(0,msg.label);
m_cMessage.SetItemText(current,1, msg.msg);    
do
{
        hr    = mque.getMSMQmsg(); 
        if(hr!=0) break;
        msg        = mque.getMessageQ();
        current = m_cMessage.InsertItem(0,msg.label);
        m_cMessage.SetItemText(current,1, msg.msg);        
            
        
}while(hr==0);
    
if(!mque.closeMSMQ())
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

if(mque.getTotalMessage()==0)
{
    AfxMessageBox("No message in Message Queue");
else
{
    CString mseg;
    mseg.Format("Get %d message(s) from public queue successfully",
                 mque.getTotalMessage());
    AfxMessageBox(mseg);        
}

Get all private queue message:

agsMSMQ mque;    
int current;
MQmsg msg;
HRESULT hr;

UpdateData();
CWaitCursor wait;
m_cMessage.DeleteAllItems();
if(!mque.PrepareGetMessage(m_sServer,m_sPath,1))
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

msg        = mque.getMessageQ();
current = m_cMessage.InsertItem(0,msg.label);
m_cMessage.SetItemText(current,1, msg.msg);    

do
{
    hr    = mque.getMSMQmsg();
    if(hr!=0) break;

    msg = mque.getMessageQ();
    current = m_cMessage.InsertItem(0,msg.label);
    m_cMessage.SetItemText(current,1, msg.msg);
}while(hr==0);
    
if(!mque.closeMSMQ())
{
    AfxMessageBox(mque.getErrorMsg());
    return;
}

if(mque.getTotalMessage()==0)
{
    AfxMessageBox("No message in Message Queue");
}else
{
    CString mseg;
    mseg.Format("Get %d message(s) from private queue successfully",
                 mque.getTotalMessage());
    AfxMessageBox(mseg);
}

Reference

Platform SDK for Message Queue (MSMQ).

History

Version 1.0 First release : November 16, 2001

Version 1.1 Modified on PrepareGetMessage() and getMSMQmsg() function