Support for message Priority - #51

Author: matrober-uk

README.md

@@ -124,6 +124,7 @@ your own error handling or logging.
 * Request/reply messaging pattern - [requestreply_test.go](requestreply_test.go)
 * Send and receive under a local transaction - [local_transaction_test.go](local_transaction_test.go)
 * Sending a message that expires after a period of time - [timetolive_test.go](timetolive_test.go)
+* Sending a message with a specified priority - [priority_test.go](priority_test.go)
 * Handle error codes returned by the queue manager - [sample_errorhandling_test.go](sample_errorhandling_test.go)
 * Set the application name (ApplName) on connections - [applname_test.go](applname_test.go)
 * Receive messages over 32kb in size by setting the receive buffer size - [largemessage_test.go](largemessage_test.go)

jms20subset/JMSProducer.go

@@ -59,4 +59,11 @@ type JMSProducer interface {
 	// GetTimeToLive returns the time to live (in milliseconds) that will be
 	// applied to messages that are sent using this JMSProducer.
 	GetTimeToLive() int
+
+	// SetPriority sets the message priority for all messages sent by this producer.
+	SetPriority(priority int) JMSProducer
+
+	// GetPriority returns the priority for all messages sent by this producer.
+	// Default priority is 4.
+	GetPriority() int
 }

jms20subset/Message.go

@@ -52,6 +52,9 @@ type Message interface {
 	// jms20subset.DeliveryMode_PERSISTENT and jms20subset.DeliveryMode_NON_PERSISTENT
 	GetJMSDeliveryMode() int
 
+	// GetJMSPriority returns the priority that is specified for this message.
+	GetJMSPriority() int
+
 	// SetStringProperty enables an application to set a string-type message property.
 	//
 	// value is *string which allows a nil value to be specified, to unset an individual

jms20subset/Priority.go

@@ -0,0 +1,18 @@
+// Derived from the Eclipse Project for JMS, available at;
+//     https://github.com/eclipse-ee4j/jms-api
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0, which is available at
+// http://www.eclipse.org/legal/epl-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0
+
+// Package jms20subset provides interfaces for messaging applications in the style of the Java Message Service (JMS) API.
+package jms20subset
+
+// Go doesn't allow constants in structs so the naming of this file is only for
+// logical grouping purposes. The constants are package scoped, but we use a
+// prefix to the naming in order to maintain similarity with Java JMS.
+
+// Priority_DEFAULT is the default priority for messages sent using a Producer.
+const Priority_DEFAULT int = 4

mqjms/ContextImpl.go

@@ -49,6 +49,7 @@ func (ctx ContextImpl) CreateProducer() jms20subset.JMSProducer {
 	producer := ProducerImpl{
 		ctx:          ctx,
 		deliveryMode: jms20subset.DeliveryMode_PERSISTENT,
+		priority:     jms20subset.Priority_DEFAULT,
 	}
 
 	return &producer

mqjms/MessageImpl.go

@@ -54,6 +54,20 @@ func (msg *MessageImpl) GetJMSDeliveryMode() int {
 	return jmsPersistence
 }
 
+// GetJMSPriority extracts the message priority from the native MQ message descriptor.
+func (msg *MessageImpl) GetJMSPriority() int {
+
+	pri := 4
+
+	// Extract the MsgId field from the MQ message descriptor if one exists.
+	// Note that if there is no MQMD then there is no messageID to return.
+	if msg.mqmd != nil {
+		pri = int(msg.mqmd.Priority)
+	}
+
+	return pri
+}
+
 // GetJMSMessageID extracts the message ID from the native MQ message descriptor.
 func (msg *MessageImpl) GetJMSMessageID() string {
 	msgIDStr := ""

mqjms/ProducerImpl.go

@@ -25,6 +25,7 @@ type ProducerImpl struct {
 	ctx          ContextImpl
 	deliveryMode int
 	timeToLive   int
+	priority     int
 }
 
 // SendString sends a TextMessage with the specified body to the specified Destination
@@ -152,6 +153,10 @@ func (producer ProducerImpl) Send(dest jms20subset.Destination, msg jms20subset.
 		putmqmd.Expiry = (int32(producer.timeToLive) / 100)
 	}
 
+	// Convert the JMS priority into the equivalent MQ message descriptor
+	// attribute.
+	putmqmd.Priority = int32(producer.priority)
+
 	// Invoke the MQ command to put the message using MQPUT1 to avoid MQOPEN and MQCLOSE.
 	// Any Err that occurs will be handled below.
 	err := producer.ctx.qMgr.Put1(mqod, putmqmd, pmo, buffer)
@@ -310,3 +315,28 @@ func (producer *ProducerImpl) SetTimeToLive(timeToLive int) jms20subset.JMSProdu
 func (producer *ProducerImpl) GetTimeToLive() int {
 	return producer.timeToLive
 }
+
+// SetPriority contains the MQ logic necessary to store the specified
+// priority parameter inside the Producer object so that it can be
+// applied when sending messages using this Producer.
+func (producer *ProducerImpl) SetPriority(priority int) jms20subset.JMSProducer {
+
+	// Only accept a non-negative value for priority.
+	if priority >= 0 {
+		producer.priority = priority
+
+	} else {
+		// Normally we would throw an error here to indicate that an invalid value
+		// was specified, however we have decided that it is more useful to support
+		// method chaining, which prevents us from returning an error object.
+		// Instead we settle for printing an error message to the console.
+		fmt.Println("Invalid Priority specified: " + strconv.FormatInt(int64(priority), 10))
+	}
+
+	return producer
+}
+
+// GetPriority returns the priority for all messages sent by this producer.
+func (producer *ProducerImpl) GetPriority() int {
+	return producer.priority
+}

next-features.txt

@@ -9,7 +9,6 @@ Not currently implemented:
 - SendToQmgr, ReplyToQmgr
 - Topics (pub/sub)
 - Temporary destinations
-- Priority
 - Configurable option to auto-set the receive buffer length if the default 32kb is exceeded (less efficient that setting up front)
 
 Client capabilities for participating in Uniform Clusters;

priority_test.go

@@ -0,0 +1,213 @@
+/*
+ * Copyright (c) IBM Corporation 2022
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package main
+
+import (
+	"testing"
+
+	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
+	"github.com/ibm-messaging/mq-golang-jms20/mqjms"
+	"github.com/stretchr/testify/assert"
+)
+
+/*
+ * Test the retrieval of special header properties
+ */
+func TestPrioritySetGet(t *testing.T) {
+
+	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
+	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	assert.Nil(t, cfErr)
+
+	// Creates a connection to the queue manager, using defer to close it automatically
+	// at the end of the function (if it was created successfully)
+	context, ctxErr := cf.CreateContext()
+	assert.Nil(t, ctxErr)
+	if context != nil {
+		defer context.Close()
+	}
+
+	// Create a TextMessage and check that we can populate it
+	msgBody := "PriorityMsg"
+	txtMsg := context.CreateTextMessage()
+	txtMsg.SetText(msgBody)
+
+	// Set up objects for send/receive
+	queue := context.CreateQueue("DEV.QUEUE.1")
+	consumer, errCons := context.CreateConsumer(queue)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+	assert.Nil(t, errCons)
+
+	// Now send the message and get it back again, to check that it roundtripped.
+	ttlMillis := 20000
+	producer := context.CreateProducer().SetTimeToLive(ttlMillis)
+
+	// Check the default priority.
+	assert.Equal(t, jms20subset.Priority_DEFAULT, producer.GetPriority())
+
+	errSend := producer.Send(queue, txtMsg)
+	assert.Nil(t, errSend)
+
+	rcvMsg, errRvc := consumer.ReceiveNoWait()
+	assert.Nil(t, errRvc)
+	assert.NotNil(t, rcvMsg)
+
+	switch msg := rcvMsg.(type) {
+	case jms20subset.TextMessage:
+		assert.Equal(t, msgBody, *msg.GetText())
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+	// Check the Priority
+	gotPropValue := rcvMsg.GetJMSPriority()
+	assert.Equal(t, jms20subset.Priority_DEFAULT, gotPropValue)
+
+	// -------
+	// Go again with a different priority.
+	newMsgPriority := 2
+	producer = producer.SetPriority(newMsgPriority)
+	assert.Equal(t, newMsgPriority, producer.GetPriority())
+
+	errSend = producer.Send(queue, txtMsg)
+	assert.Nil(t, errSend)
+
+	rcvMsg, errRvc = consumer.ReceiveNoWait()
+	assert.Nil(t, errRvc)
+	assert.NotNil(t, rcvMsg)
+
+	switch msg := rcvMsg.(type) {
+	case jms20subset.TextMessage:
+		assert.Equal(t, msgBody, *msg.GetText())
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+	// Check the Priority
+	gotPropValue = rcvMsg.GetJMSPriority()
+	assert.Equal(t, newMsgPriority, gotPropValue)
+
+	// -------
+	// Go again with a different priority.
+	newMsgPriority2 := 7
+	producer = producer.SetPriority(newMsgPriority2)
+	assert.Equal(t, newMsgPriority2, producer.GetPriority())
+
+	errSend = producer.Send(queue, txtMsg)
+	assert.Nil(t, errSend)
+
+	rcvMsg, errRvc = consumer.ReceiveNoWait()
+	assert.Nil(t, errRvc)
+	assert.NotNil(t, rcvMsg)
+
+	switch msg := rcvMsg.(type) {
+	case jms20subset.TextMessage:
+		assert.Equal(t, msgBody, *msg.GetText())
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+	// Check the Priority
+	gotPropValue = rcvMsg.GetJMSPriority()
+	assert.Equal(t, newMsgPriority2, gotPropValue)
+
+}
+
+/*
+ * Test the functional behaviour of messages sent with different priorities,
+ * where the expectation is that they should be returned from the queue in priority
+ * order (with FIFO within a priority group), and not FIFO across the entire queue.
+ */
+func TestPriorityOrdering(t *testing.T) {
+
+	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
+	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	assert.Nil(t, cfErr)
+
+	// Creates a connection to the queue manager, using defer to close it automatically
+	// at the end of the function (if it was created successfully)
+	context, ctxErr := cf.CreateContext()
+	assert.Nil(t, ctxErr)
+	if context != nil {
+		defer context.Close()
+	}
+
+	// Set up objects for send/receive
+	queue := context.CreateQueue("DEV.QUEUE.1")
+	consumer, errCons := context.CreateConsumer(queue)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+	assert.Nil(t, errCons)
+
+	// Send a sequence of messages with different priorities.
+	id_msg1_pri5 := sendMessageWithPriority(t, context, queue, 5)
+	id_msg2_pri2 := sendMessageWithPriority(t, context, queue, 2)
+	id_msg3_pri8 := sendMessageWithPriority(t, context, queue, 8)
+	id_msg4_pri2 := sendMessageWithPriority(t, context, queue, 2)
+	id_msg5_pri4 := sendMessageWithPriority(t, context, queue, 4)
+	id_msg6_pri5 := sendMessageWithPriority(t, context, queue, 5)
+	id_msg7_pri2 := sendMessageWithPriority(t, context, queue, 2)
+	id_msg8_pri2 := sendMessageWithPriority(t, context, queue, 2)
+	id_msg9_pri4 := sendMessageWithPriority(t, context, queue, 4)
+	id_msg10_pri1 := sendMessageWithPriority(t, context, queue, 1)
+
+	// We expect them to be returned in priority group order (highest first),
+	// with FIFO within the priority group.
+	//
+	// This behaviour relies on MsgDeliverySequence for the queue
+	// being set to MQMDS_PRIORITY (which is the default).
+	receiveMessageAndCheck(t, consumer, id_msg3_pri8, 8)
+	receiveMessageAndCheck(t, consumer, id_msg1_pri5, 5)
+	receiveMessageAndCheck(t, consumer, id_msg6_pri5, 5)
+	receiveMessageAndCheck(t, consumer, id_msg5_pri4, 4)
+	receiveMessageAndCheck(t, consumer, id_msg9_pri4, 4)
+	receiveMessageAndCheck(t, consumer, id_msg2_pri2, 2)
+	receiveMessageAndCheck(t, consumer, id_msg4_pri2, 2)
+	receiveMessageAndCheck(t, consumer, id_msg7_pri2, 2)
+	receiveMessageAndCheck(t, consumer, id_msg8_pri2, 2)
+	receiveMessageAndCheck(t, consumer, id_msg10_pri1, 1)
+
+}
+
+// Send a message with the specified priority.
+// Return the generated MessageID.
+func sendMessageWithPriority(t *testing.T, context jms20subset.JMSContext, queue jms20subset.Queue, priority int) string {
+
+	ttlMillis := 20000
+	producer := context.CreateProducer().SetTimeToLive(ttlMillis)
+	producer = producer.SetPriority(priority)
+
+	// Create a TextMessage and check that we can populate it
+	msgBody := "PriorityOrderingMsg"
+	txtMsg := context.CreateTextMessage()
+	txtMsg.SetText(msgBody)
+
+	errSend := producer.Send(queue, txtMsg)
+	assert.Nil(t, errSend)
+
+	return txtMsg.GetJMSMessageID()
+
+}
+
+// Try to receive a message from the queue and check that it matches
+// the expected attributes
+func receiveMessageAndCheck(t *testing.T, consumer jms20subset.JMSConsumer, expectedMsgID string, expectedPriority int) {
+
+	rcvMsg, errRvc := consumer.ReceiveNoWait()
+	assert.Nil(t, errRvc)
+	assert.NotNil(t, rcvMsg)
+
+	assert.Equal(t, expectedPriority, rcvMsg.GetJMSPriority())
+	assert.Equal(t, expectedMsgID, rcvMsg.GetJMSMessageID())
+
+}