How to Use MQSeries with Perl

Today I managed to finally get Perl to put and get messages to MQ Series. It’s something that I’ve been wanting to do for quite some time, but didn’t have the time or even MQ knowledge to do so.

This post is intended for those who, like me, aren’t MQSeries gurus and can’t make much of the documentation of the MQSeries module in CPAN. I hope it serves you well and in the end you will be able to use MQSeries with Perl.

Very VERY Brief introduction to MQSeries

MQ Series (recently renamed Websphere MQ) is a messaging software developed by IBM. There is a client and a server piece to it.

MQ Messages are transmitted between servers by means of queues. These queues are managed by Queue Managers. Queues can be configured to receive messages (input or request) and/or return messages (output or reply).

The installation of MQ series is beyond the scope of this post, but it’s important to know that if you need to communicate with a queue manager installed in a local machine, you have to use one set of APIs (the server APIs), and if you want to talk to a remote queue manager, you’ll use the client APIs. This is important for later on.

A MQ message is composed of 2 main parts: the MQ Message Descriptor (MQMD) and the Data portion. The MQMD portion works as a header where several fields can be set to tell the server how to handle the message. The fields we will be using in this tutorial are MsgId, CorrelId, MsgType, Expiry, ReplyToQ, and ReplyToQMgr.

MsgId: Each message being sent into the server gets a unique Message ID. They never repeat, even when the server is restarted.

CorrelId: The Correlation Id is how a reply message can be associated to a request message.

MsgType: The kind of data you will be sending through the queues. We will be using a simple string format.

Expiry: The amount of time that the server should keep the message in a queue.

ReplyToQMgr: The name of the Queue Manager where the reply must be sent to.

ReplyToQ: The name of the queue where the reply must be sent to.

You will find more information about the fields and MQ Series in general at the MQSeries Application Programming Reference. I couldn’t find a link to the hardcopy at Amazon, but you can consider getting this one.

Before trying to put or get messages, you will probably need to set the MQSERVER environment variable (for both Unix and Windows – don’t know how Mac works), since the client usually requires it. The value for MQSERVER is “CHANNEL_NAME/TCP/HOST_IP(PORT)”. There can be variances according to the settings of the host you’ll connect to, but this is the most common scenario.

MQSeries CPAN modules

CPAN has a very handy set of modules which allow us to communicate with MQSeries. The main module, MQSeries, exports all the funny named subroutines that you need to use to send and receive messages. If you don’t have working knowledge with C/C++ using MQ Series, then don’t even bother trying to use it. The Object Oriented modules (by the same author) are much simpler for the unseasoned MQ Series explorer (like myself).

First off, you will need to install the modules in your computer. If you’re working on windows, read this article to get it done. It’s simpler on Unix, but both cases require that you have the MQ Series client installed. You can get it from here

When you install MQSeries module from CPAN, you also get MQSeries::QueueManager, MQSeries::Queue, MQSeries::Message and some other goodies. If you didn’t get any of those automatically, make sure you install them before you go on.

Writing your own MQSeries module

Here we’ll write a few methods which should make people’s lives easier when communicating with MQ Series. Now, depending on the way your application works, you will probably have to handle the creating and breaking up of the Data portion of the message, but again, that’s outside the scope of this post.

So let’s get started…

Start your module by giving it a package name. Remember to end the code with a true value. We’ll also import the modules we’ll need

package myMQModule;

use strict;
use MQSeries;
use MQSeries::QueueManager;
use MQSeries::Queue;
use MQSeries::Message;


1; # end with a true value

Now let’s lay out the basics of our module.

package myMQModule;

use strict;
use MQSeries;
use MQSeries::QueueManager;
use MQSeries::Queue;
use MQSeries::Message;

sub new() { # our constructor
}

sub openQueueMgr() { # open the Queue Manager
}

sub openQueue() { # opens the Queue
}

sub putRequest() { # puts the message
}

sub getReply() { # gets the reply using the correlId
} 

1; # end with a true value

You’ll probably notice that we haven’t added methods to close the queue and the queue manager. That’s ok, since the MQSeries* modules do that for us.

Let’s go on with our methods. Starting with new(). We’ll use the basic instantiation code. You will probably need to enhance this to suit your own needs, such as calling openQueueMgr() and openQueue() automatically. For now, we’ll KISS.

package myMQModule;

use strict;
use MQSeries;
use MQSeries::QueueManager;
use MQSeries::Queue;
use MQSeries::Message;
use Carp; # exports carp, confess, etc.

sub new() { # our constructor
    my $invocant = shift;
    my $class = ref($invocant) || $invocant; # Object or class name
    my $self = {};       # initiate our handy hashref
    bless($self,$class); # make it usable
    return $self;
}

sub openQueueMgr() { # open the Queue Manager
}

sub openQueue() { # opens the Queue
}

sub putRequest() { # puts the message
}

sub getReply() { # gets the reply using the correlId
} 

1; # end with a true value

Now we need to get the Queue Manager and the Queue open. In real life, you’ll probably have several queue managers and several queues. In our example, we’ll use 2 of each, one pair for outgoing traffic and another for incoming. Remember that this depends on your MQ Series configuration. These examples should simply allow you to get your toes wet in preparation before diving into your custom settings.

...
sub openQueueMgr() { # open the Queue Manager
    my $self = shift; # take the handy hashref
    my $qm_name    = shift; # the name of the queue manager
    my $qm_ip         = shift; # the IP to the queue manager
    my $qm_port      = shift; # the port it's listening to
    my $qm_channel = shift; # the channel configured in the QM
    my $type           = shift; # request or reply? we will need it later

    # some validation code here making sure that the QM name, ip, 
    # port, and channel have been provided 
    
    # export the MQSERVER environment variable just in case your clients need it
    $ENV{MQSERVER} = "$qm_channel/TCP/$qm_ip($qm_port)";

    # Here we create a MQSeries::QueueManager object to connect to our QM
    # It takes, among other parameters, the QM name, AutoConnect, and 
    # a hashref with the data we gave to $ENV{MQSERVER}
    my $connOpts = {
          ChannelName => $qm_channel,
          TransportType => 'TCP',
          ConnectionName => "$qm_ip($qm_port)",
          MaxMsgLength   => 16 * 1024 * 1024,
    };

    # Create MQSeries::QueueManager object
    my $qm = MQSeries::QueueManager->new(
        QueueManager => $qm_name,
        AutoConnect => 0, # we do not autoconnect now 
                          # because we want to handle it better later
        ClientConn => $connOpts,
    );
  
    #  kick it off and see if it connects
    eval {
        $qm->Connect() 
            || die(sprintf("Connect failed with CompCode: %s", 
                               Reason %sn",$qm->CompCode(),$qm->Reason())); 
    };
    if ($@) { # eval caught the die
       confess($@);
   }

   # got this far, so it obviously connected
   # now save that connection in a safe place
   $self->{uc($type)}->{QM_CONN} = $qm; # MQSeries::QueueManager object.
                                        # We need it to open the queue
   $self->{uc($type)}->{QM_NAME} = $qm_name; # note that we also saved the QM_NAME and it is all under the $type (REQUEST/REPLY)
   
}
 ...

The snippet above showed you how to connect to the queue manager. It’s pretty well commented, so we can go on with our next step, which is opening the queue.

...
sub openQueue() { # opens the Queue
    my $self       = shift;
    my $queue_name = shift; 
    my $type       = shift; # again, we will need it for later. 
                           # (REQUEST/REPLY)

    # set mode according to type
    my $mode = uc($type) eq 'REQUEST' ? 'output' : 'input';

    # open the queue directly this time
    # here we use the MQSeries::Queue module
    eval{ 
        $self->{uc($type)}->{Q_OBJ} = 
           MQSeries::Queue->new(
               QueueManager  => $self->{uc($type)}->{QM_CONN}, # the QueueManager object from before
               Queue         => $queue_name,
               Mode         => $mode,
               AutoOpen         => 1, # open it directly
           )|| die("Could not open Queue $queue_name");
    }; 
    if ($@) {
        confess($@);
    }

    # made it this far, so we obviously got a connection
    # so let's save the queue name for later
    $self->{uc($type)}->{QUENAME} = $queue_name;
}
...

Now we have both our QueueManager and our Queue open. Since we’re dealing with one request pair and one reply pair, you’ll have to call it once for the request and once for the reply. You can also choose to save the connections in a different data structure to handle several queue managers and even more queues per queue manager. The only limit is your creativity. Right now, we’ll stick to the $self->{REQUEST} and $self->{REPLY} structures.

Continuing our layout, we now have to handle putting data into the queue.

...
sub putRequest() { # puts the message
    my $self = shift;
    my $message = shift; # the data we will be putting
    
    # We need a MQSeries::Message object to inject
    my $msg = 
        MQSeries::Message->new(
             MsgDesc => { # this is the MQMD (header) portion
				  Format	=> MQFMT_STRING,
				  Expiry	=> 100, # in tenths of seconds
				  ReplyToQ => $self->{REPLY}->{QUENAME},
				  ReplyToQMgr => $self->{REPLY}->{QM_NAME},
       				},
            Data => $message,
        );

    # now that we have a message object to inject, do the injecting
    my $request = $self->{REQUEST}->{Q_OBJ};
    eval {
       $request->Put(Message=>$msg) || die(sprintf("Put failed with CompCode: %s, Reason: %sn",
                                                            $request->CompCode(),$request->Reason()));
    };
    if($@){
        confess($@);
    }
         
    # got this far, so Put was successful. We'll return the MsgId to be used
    # as CorrelId when getting the reply
    return $msg->MsgDesc('MsgId'); # comes from the injected MQSeries::Message object         
}
...

If you try printing the MsgId, you’ll get some funny characters in the screen, since it’s pure binary data. Look up “perl bin to hex” on Google if you’d like to see some ways of converting and not getting gibberish.

Next comes the last but not least portion of our module – to get the reply message.

...
sub getReply() { # gets the reply using the correlId
    my $self      = shift;
    my $correlId = shift;
   
    # some sanity checking, since we can't do anything without $correlId
    confess("I need a correlId!!!n") unless $correlId;

    # create the Message object with the $correlId
    my $reply_msg = 
            MQSeries::Message->new(
                 MsgDesc =>
				{
				    CorrelId	=> $correlId,
				},
            );
   
    my $reply_q = $self->{REPLY}->{Q_OBJ};
    eval {
       $reply_q->Get(
                     Message=> $reply_msg,
                     Wait => 100, # some interval before timing out
                ) || die(sprintf("Get message failed with CompCode: %s, Reason: %sn",
                                      $reply_q->CompCode(),$reply_q->Reason()));
    };
    if ($@){
        confess($@);
    }

    # got this far, so the Get was successful. 
    # now we return the Data portion of the reply

    return $reply_msg->Data(); # from the Message object, not the Queue object!
} 
...

And that’s all there is to the module! Now you can call it from your script, like so

#!/usr/bin/perl

use strict;
use lib 'directory containing myMQModule';
use myMQModule;

# create object
my $mq = myMQModule->new();

# open queueMgr1
my($qm_name,$qm_ip,$qm_port,$qm_channel,$type) = qw(QMREQ some_ip some_port CHANNEL1 request); 
$mq->openQueueMgr($qm_name,$qm_ip,$qm_port,$qm_channel,$type);

#open queueMgr2
($qm_name,$qm_ip,$qm_port,$qm_channel,$type) = qw(QMREP some_ip2 some_port2 CHANNEL2 reply); 
$mq->openQueueMgr($qm_name,$qm_ip,$qm_port,$qm_channel,$type);

# now for the queues
$mq->openQueue('QUEUE1', 'request');
$mq->openQueue('QUEUE2', 'reply');

# put some message
my $msgId = $mq->putRequest('this is a test message');
my $reply_msg = $mq->getReply($msgId);

print "We're done!";

I really hope this was helpful. Comments are always welcome.

Book suggestions: