Testing the MQ REST API

I have been playing around with the MQ REST API. It works very well. Also certificate-based authentication work out of the box.

Of course, you are doing something that MQ-fanatics might find horrific: reliable messaging over an unreliable protocol. They are somewhat right. By no means can MQ provide assured message delivery over an unreliable HTTP protocol. When using this in application, make sure you handle all error situations. For example, when you do not get an http response, you don’t know whether the message was successfully delivered or not. You application has to cater for such situations. Some call this idempotence.

Here is my small Python program that illustrates how you can use the MQ REST API.

import requests
import json
import sys

class MQWebManager:
    baseapiurl = "/ibmmq/rest/v1/messaging"
    
    def __init__(self, ep, ak, cert_file_path, key_file_path):
        self.endpoint = ep
        self.apikey = ak
        self.cert = (cert_file_path, key_file_path) 

    def apideleterequest(self, qmgr, queue, msgid):
        # operation = POST or DELETE
        resourceurl = self.endpoint + "/ibmmq/rest/v1/messaging/qmgr/" + qmgr + "/queue/" + queue + "/message"
        request_headers = {
            'messageId': "'" + msgid + "'",
            'Content-Type' : 'text/plain;charset=utf-8' ,
            'ibm-mq-rest-csrf-token' : 'somevalue',
            'correlationId' : ''
            }
        data = {}
        response = requests.delete(resourceurl, data=data, cert=self.cert, verify=False, headers=request_headers)
        return response

    def apipostrequest(self, qmgr, queue):
        # operation = POST or DELETE
        resourceurl = self.endpoint + "/ibmmq/rest/v1/messaging/qmgr/" + qmgr + "/queue/" + queue + "/message"
        request_headers = {
            'Content-Type' : 'text/plain;charset=utf-8' ,
            'ibm-mq-rest-csrf-token' : 'somevalue'
            }
        data = 'hello from apipostrequest'
        print('resource url: ', resourceurl)
        response = requests.post(resourceurl, data=data, cert=self.cert, verify=False, headers=request_headers)
        return response



print('---------')

#cert_file_path = "/yourpath/yourcert.crt"   
#key_file_path = "/yourpath/yourcert.privkey"

cert_file_path = sys.argv[1]
key_file_path = sys.argv[2]

m1 = MQWebManager("https://mqweb.yourzos.com:12345","", cert_file_path, key_file_path)

#put a message on the queue
response = m1.apipostrequest("QMGR","YOUR.Q.NAME") 
print(">>>", response.status_code, response.json)

print(response.headers) 

print(response)
#retrieve msgid from the message we just put there
msgid = response.headers['ibm-mq-md-messageId']
print(response.headers['ibm-mq-md-messageId'])

#delete that message we just put there
response = m1.apideleterequest("QMGR","YOUR.Q.NAME", msgid) 
print(">>>", response.status_code, response.json)



print('---------')

Leave a Reply