Yesterday I was asked to help out on another project that was trying to read some messages from an MQSeries queue. For this particular project, they were using C# and the MQ API rather than BizTalk, because they were dealing with “grouped” messages. Grouped messages were a completely new concept to me. Basically, the gist of it is that you have 1 – N physical messages that make up one logical message. Each physical message has a particular flag set to indicate it is part of a group and contains a sequence number to indicate it’s position within the group. Additionally, the message has a Group ID property that indicates the particular logical message group it belongs to. For example, a grouped message may look like this:
Here we have 5 separate physical messages that are all part of the same logical message.
So, the project I was asked to help out with had been having some problems reading these logical messages. Basically, when a message that was part of a logical group was read off of a queue, it would have this garbage header data at the beginning of the message. The screenshot below shows the content of a message dumped to the Immediate window in Visual Studio, with the garbage header highlighted in red.
I’m a novice when it comes to MQ and in fact my only experience has been using MQ via BizTalk, I’d never looked at the API before. I dove in and started writing some code to see if I could read these messages and avoid the garbage data. After considerable trial and error, I came up with the following which seems to work quite well:
1: MQGetMessageOptions options = new MQGetMessageOptions
2: {
3: Options = MQC.MQGMO_LOGICAL_ORDER | MQC.MQGMO_ALL_MSGS_AVAILABLE | MQC.MQGMO_SYNCPOINT,
4: MatchOptions = MQC.MQMO_NONE
5: };
6:
7: MQQueueManager qManager = new MQQueueManager("<your queue mgr>",
8: "<your channel name>",
9: "<your server name>");
10: MQQueue queue = qManager.AccessQueue("<your queue name>",
11: MQC.MQOO_INPUT_AS_Q_DEF + MQC.MQOO_FAIL_IF_QUIESCING);
12:
13: MQMessage message = new MQMessage
14: {
15: Format = MQC.MQFMT_STRING,
16: Version = MQC.MQMD_VERSION_2
17: };
18:
19: using (StreamWriter writer = new StreamWriter(Guid.NewGuid().ToString() + ".txt"))
20: {
21: do
22: {
23: queue.Get(message, options);
24: options.MatchOptions = MQC.MQMO_MATCH_GROUP_ID;
25: string content = message.ReadString(message.DataLength);
26: Console.WriteLine(content);
27: writer.WriteLine(content);
28: }
29: while (options.GroupStatus != MQC.MQGS_LAST_MSG_IN_GROUP);
30: }
31: queue.Close();
32: qManager.Disconnect();
It seems that the key to getting rid of the garbage data at the beginning of the message was settings MQMessage.Version to MQC.MQMD_VERSION_2.
Obviously a big concern would be what happens if you’re reading a grouped message and encounter an exception before you’ve reached the last segment. Since reading a message from a queue causes that message to be removed from the queue, this could be a real problem. What you need is a way to read segmented messages in a transactional manner. This is what the MQC.MQGMO_SYNCPOINT flag gets you, all calls to MQQueue.Get will be treated as a single unit of work that is either committed or rolled back.