Windows Mobile Interprocess Communication with Message Queues in .NET

• Download source - 216.03 KB

Introduction

When the need arises to pass information between different programs, Windows Mobile and Windows CE offer a variety of technologies and solutions to do so. Information can be passed through shared storage locations such as the registry, files, or a database. For frequent communication of small messages, one can place messages on an applications message pump or use through message queues. Message Queues are in the same family of objects as events, semaphores, and mutexes; they are named kernel objects. Presently the .NET Framework does not support these objects directly. But through a few P/Invokes declarations, the functionality can easily be accessed. Within this article, I show how to interact with the message queue functionality.

My goal is not to do an exhaustive explanation of message queues, but to cover enough information for the readers to pick up the concept and carry on.

Prerequisites

This article builds upon the concepts of some other Windows CE kernel objects, namely the event, mutex, and semaphore. It also builds upon code that I presented within a recent article on Windows Mobile Native Thread Synchronization for .NET. Before reading through this article, please read the above article first; I extend the code from this previous article and you will want to have familiarity with it before reading this article.

Native Functions and Structures

There are a number of native functions that are central to this article. Detailed information on the functions can be found within the MSDN library. The functions are listed below:

• CloseMsgQueue
• CreateMsgQueue
• GetMsgQueueInfo
• OpenMsgQueue
• ReadMsgQueue
• WriteMsgQueue

• MSGQUEUEINFO
• MSGQUEUEOPTIONS

What is a Message Queue

At its simplest, a queue is an ordered list. An item can be added to the list or moved from the list. However one cannot add or remove items from arbitrary positions within the list. Items can only be removed from the beginning of the list and items can only be added to the end of the list. These rules on insertion and removal are often labelled as FIFO (First In First Out) or FCFS (First Come First Server). Windows CE devices offer two implementations for message queues. One implementation is a part of the operation system and is used for communication among processes on the same device. The other is an implementation of MSMQ and can be installed onto a Windows CE device and can be used for communicating with other machines. This paper is centered around the first of the two implementations.

A message queue can be specific to a process or shared among processes. In either case, a handle to a message queue allows either read-only or write-only access to the queue. You cannot both read and write to a message queue using the same handle. If you already have a handle to a message queue, you can create extra handles for reading or writing to the associated queue. This is especially important for unnamed queues.

As previously mentioned, the code in this article builds upon the code from a previous article. The relationship between the code in the previous article and the code within this article is visible below in the class hierarchy diagram. The classes in the dark blue (MessageQueue, MessageQueueReader, MessageQueueWriter) are the classes being added in this article.

Creating and Opening a Message Queue

A message queue is created or opened through the native function CreateMsgQueue. Like the synchronization objects in the previous article, the message queue must be assigned a name to be shared among processes. If multiple processes create message queues with the same name, then each process will receive a handle to the same message queue. The call to create the message queue must pass a MSGQUEOPTIONS structure to specify the maximum number of items that can be in the message queue, the maximum size of each message on the queue, and whether a read-only or write-only handle is being requested.

If your message queues are only being used to move information around to threads within the same process (in which case the message queue probably has no name), you will need to use the handle to the first message queue that you created to create another handle that attaches to the same queue using OpenMsgQueue. Without this function as you create new message handles, you have no way to specify that the handle you are creating should be attached to a previous queue that already exists.

Message queues are created by the operating system and are a system resource. You must be sure to free your handle to the message queue when you no longer have a need for it. You can free a message queue handle with CloseMsgQueue.

The Base Class

Like the system event, semaphore, and mutex message queues are waitable, can optionally have names, and are tied to resources through a handle that must be manually cleared. The SyncBase class from the previous article was already designed to hold a handle to a system resource. The layout of the class is displayed below:

public abstract class SyncBase : IDisposable
    {
        protected IntPtr _hSyncHandle;
        protected bool _firstInstance;

        public bool FirstInstance
        {
            get { return _firstInstance; }
        }

        public IntPtr SyncHandle
        {
            get { return _hSyncHandle; }
        }

        public static SyncBase WaitForMultipleObjects
		(int timeout, params SyncBase[] syncObjectList) {...}

        public bool SetEvent() {...}


        public static SyncBase WaitForMultipleObjects
		(params SyncBase[] syncObjectList) {...}

        public WaitObjectReturnValue Wait(int timeout) {...}

        public WaitObjectReturnValue Wait() {...}


        #region IDisposable Members

        public virtual void Dispose()
        {
            if(!_hSyncHandle.Equals(IntPtr.Zero))
                CoreDLL.CloseHandle(_hSyncHandle);
        }

        #endregion
    }

As with the system event, when code waits on a queue it will be blocked until the queue is in a signalled state. For a read-only handle to a queue, the signalled state means that there is data ready to be read from the queue. For a write-only handle, the signalled state means that there is enough room in the queue for more messages. The handle is no longer signalled once the queue is full. So the Wait methods defined in the base class are usable without modification.

The constructor must be defined and call CreateMsgQueue. The handle returned from the call will be saved in the member _hSyncHandle.

Constructors

While there are several message queue constructors, they all call one of two foundational constructors. The native function CreateMsgQueue is used in the constructor and it requires the MSGQUEUEOPTIONS parameter for the creation parameters. I've implemented the managed class MsgQueueOptions to reflect the structure and have wrapped the creation and initiation of instances of this class in the method GetMessageQueueOptions. There is a message queue setting that I've now allowed the caller. It is the dwFlags member of MsgQueueOptions. The options that can be set through this flag are MSGQUEUE_NOPRECOMMIT and MSGQUEUE_ALLOW_BROKEN. MSGQUEUE_NOPRECOMMIT which prevents the system from preallocating the memory needed for the message queue and causes memory to be allocated as needed. MSGQUEUE_ALLOW_BROKEN is used to allow a writer queue to be usable even if there is no reader queue on the other end. IF this option is not specified and if there is an attempt to write to the queue before a reader is attached to the queue, then the handle becomes immediately unusable (for that reason, I would never omit that flag. If the creation of the reader or writer results in the creation of the queue, then GetLastWin32Error will return SUCCESS (numerical value 0). Otherwise it returns ERROR_ALREADY_EXISTS. As long as a non-zero handle is returned by CreateMsgQueue, then the call was successful. I use GetLastWin32Error() to determine whether or not the first handle to the queue was just created. Here is the code to the base constructor:

internal MessageQueue(string name, bool readAccess, int maxItems, int itemSizeInBytes)
{
    MsgQueueOptions options = GetMessageQueueOptions
		(readAccess, maxItems, itemSizeInBytes);
    _hSyncHandle = CoreDLL.CreateMsgQueue(name, options);
    int lastError = Marshal.GetLastWin32Error();

    if (IntPtr.Zero.Equals(_hSyncHandle))
    {
        throw new ApplicationException(String.Format("Could not create or
		open message queue {0}.  LastWin32Error={1}", name, lastError));
    }
    _firstInstance = (0 == lastError);
}    

The other important constructor is to create a queue endpoint that is connected to an existing queue using another queue endpoint as an argument. If you are working with a queue that has no name, then this is the only way that you will be able to create another endpoint to the queue. The native function OpenMsgQueue is used to do this. Like CreateMsgQueue this method requires a MsgQueueOptions. The only arguments within the options structure that are used by OpenMsgQueue are dwSize and bReadAccess. The other constructor for the base message queue class follows:

internal MessageQueue(MessageQueue source, int maxCount, bool readOnly)
{
    _firstInstance = false;
    MsgQueueOptions options = GetMessageQueueOptions(readOnly, maxCount, 0);
    IntPtr processID = (IntPtr)Process.GetCurrentProcess().Id;
    _hSyncHandle = CoreDLL.OpenMsgQueue(processID, source._hSyncHandle, options);
    if (_hSyncHandle.Equals(IntPtr.Zero))
    {
        int errorCode = Marshal.GetLastWin32Error();
        QueueResult result = Win32ErrorCodeToQueueResult(errorCode);
        string errorMessage = String.Format("Error occurred opening
	   read message queue (Win32Error={0}, QueueResult={1}", errorCode, result);
        if (result == QueueResult.InvalidHandle)
        {
            CoreDLL.CloseMsgQueue(_hSyncHandle);
            _hSyncHandle = IntPtr.Zero;
        }
        throw new ApplicationException(errorMessage);
    }
}

This message queue implementation is disposable; when it is no longer being used its resources can be cleaned up by calling the dispose message.

Subclassing the Queue

When I was writing the wrapper for the message queue functionality, I considered throwing an exception if a developer tried to read from a write-only queue or write to a read-only queue. It made more sense to simply not allow a developer to perform such an invalid action. So I subclassed the MessageQueue class into two classes; MessageQueueReader and MessageQueueWriter. Each one either contains a set of methods for reading or writing to a message queue but not both. The constructors for these classes only call the base constructor with the readOnly argument set to true or false.

Writing to a Queue

The methods that are used for writing to a queue use WriteMsgQueue. The method will block the caller if there is no space on the queue in which to write a message. CreateMsgQueue accepts a parameter named dwTimeout for specifying how long a caller will wait before the write request is considered to be a failure. If this values is set to INFINITE (numerical value -1), then the call will block indefinitely until there is enough free space to perform the write.

Within my implementation of this wrapper, the developer can only pass information to be written to the queue in the form of an array of bytes. I thought about using generics to make the code more flexible but I found a few ways by which such an implementation is misused and abused. It is the burden of the developer to convert her/his data to a byte array. There are two write methods exposed in my implementation. The first accepts the message byte array and a timeout value. The second contains only the message bytes and assumes the timeout value to be INFINITE.

Reading from a Queue

The Read methods are reflections of the Write methods; the developer gives the method a byte buffer and optionally a timeout value. The call will block if there is nothing available to read and the timeout value controls how long the method will wait for a message before the read attempt is considered a failure.

Read and Write Results

Since failures in attempts to read from a queue or write to a queue are expected to be a part of the normal flow of execution, I have decided not to throw exceptions when a write request fails. There are performance implications with throwing exceptions so I try not to throw them unnecessarily. Instead these methods return a value of the enumeration type QueueResult. QueueResult.OK represents successful completion of a read or write request and the other values represent failure with a reason (such as the write request timing out).

Code Examples

Reader/Writer Client

The reader/writer client example creates a message queue and allows the user to add messages to the queue from the main (UI) thread and processes the messages from the queue on a separate thread. From the standpoint of the user experience, there's not much to look at. The interesting workings are all in the code.

/// <summary>
/// Waits on messages to be placed on the queue and displays them as they arrive
/// </summary>
void ReaderThread()
{
    using(_reader)
    {
        while (!_shuttingDown)
        {
            //The following call will block this thread until there is either a message
            //on the queue to read or the thread is being signalled to run to prepare
            //for program termination. Since the following call blocks the thread until
            //it is time to do work it is not subject to the same batter killing
            //affect of other similar looking code patterns
            //( http://tinyurl.com/6rxoc6 ).
            if (SyncBase.WaitForMultipleObjects(_readerWaitEvent, _reader) == _reader)
            {
                string msg;
                _reader.Read(out msg);  //Get the next message
                AppendMessage(msg);     //Display the thread to the user
            }
        }
    }
}

/// <summary>
/// Appends processed message to top of list box.
/// </summary>
/// <param name=""message""></param>
public void  AppendMessage(string message)
{
    //If this is called from a secondary thread then marshal it to
    //the primary thread.
    if (this.InvokeRequired)
    {
        this.Invoke(_appendDelegate, new object[] { message });
    }
    else
    {
        this.lstReceivedMessages.Items.Insert(0, message);
    }
}    

Writer Client

The Writer client works with the previous code example. It connects to the same queue as the previous code example and any messages that the user places on the queue will show up in the other program (if it is running). If you run the writer client by itself without starting the reader client, then the messages will accumulate on the queue until it is full. If you attempt to write a message to the queue while it is full, the request will block for 4 seconds and then return a timeout result.

Power Notification Queue

The power notification queue example has relevance to an article I posted on Windows Mobile Power Management. The program creates a queue reader and passes the handle to the reader queue in a call to the native function RequestPowerNotifications. The operating system will then write messages to the queue to notify the programs of changes in power state. The notifications are appended to beginning of a list view. Selecting an item in the list will cause the relevant power flags to be displayed at the bottom of the screen.

The messages that result from requesting power notifications are passed as structs, but the wrapper I have provided works with byte arrays. I've created a new queue type that inherits from the MessageQueueReader and provides an implementation of Read that returns power queue messages. The overloaded Read method reconstructs the PowerBroadcast structure.

public PowerBroadcast Read()
{
    PowerBroadcast retVal = new PowerBroadcast();
    int bytesRead;
    QueueResult result;
    result = Read(readBuffer, out bytesRead);
    if (QueueResult.OK == result)
    {
        int message = readBuffer[0] | readBuffer[1] << 8 |
			readBuffer[2] << 0x10 | readBuffer[3] << 0x18;
        int flags = readBuffer[4] | readBuffer[5] << 8 |
			readBuffer[6] << 0x10 | readBuffer[7] << 0x18;
        int length = readBuffer[8] | readBuffer[9] << 8 |
			readBuffer[10] << 0x10 | readBuffer[11] << 0x18;

        retVal.Message = (PowerBroadCastMessageType)message;
        retVal.Flags = (PowerBroadcastFlags)flags;
        retVal.Length = length;
        if ((length > 0)&&( (retVal.Message&PowerBroadCastMessageType.PBT_TRANSITION)
			==PowerBroadCastMessageType.PBT_TRANSITION))
        {
            retVal.SystemPowerState = TextEncoder.GetString(readBuffer,12,length);
        }
    }
    return retVal;
}       

Closing

I've covered the essentials of Windows messages queues and the information provided should be more than sufficient for more scenarios that require the use of message queues. But don't stop with this article. Continue to read on about message queues and the other named objects within the MSDN library.

History

• 16^th November, 2008 - Initial publication