Async Put, with no checks for errors - #32

Author: matrober-uk

asyncput_test.go

@@ -0,0 +1,166 @@
+/*
+ * Copyright (c) IBM Corporation 2021
+ *
+ * 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 (
+	"strconv"
+	"testing"
+
+	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
+	"github.com/ibm-messaging/mq-golang-jms20/mqjms"
+	"github.com/stretchr/testify/assert"
+)
+
+/*
+ * Test the ability to send a message asynchronously, which can give a higher
+ * rate of sending non-persistent messages, in exchange for less/no checking for errors.
+ */
+func TestAsyncPut(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 the producer and consumer with the SYNCHRONOUS (not async yet) queue
+	syncQueue := context.CreateQueue("DEV.QUEUE.1")
+	producer := context.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_NON_PERSISTENT).SetTimeToLive(60000)
+
+	consumer, errCons := context.CreateConsumer(syncQueue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// Create a unique message prefix representing this execution of the test case.
+	testcasePrefix := strconv.FormatInt(currentTimeMillis(), 10)
+	syncMsgPrefix := "syncput_" + testcasePrefix + "_"
+	asyncMsgPrefix := "asyncput_" + testcasePrefix + "_"
+	numberMessages := 50
+
+	// First get a baseline for how long it takes us to send the batch of messages
+	// WITHOUT async put.
+	syncStartTime := currentTimeMillis()
+	for i := 0; i < numberMessages; i++ {
+
+		// Create a TextMessage and send it.
+		msg := context.CreateTextMessageWithString(syncMsgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(syncQueue, msg)
+		assert.Nil(t, errSend)
+	}
+	syncEndTime := currentTimeMillis()
+
+	syncSendTime := syncEndTime - syncStartTime
+	//fmt.Println("Took " + strconv.FormatInt(syncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " synchronous messages.")
+
+	// Receive the messages back again
+	finishedReceiving := false
+	rcvCount := 0
+
+	for !finishedReceiving {
+		rcvTxt, errRvc := consumer.ReceiveStringBodyNoWait()
+		assert.Nil(t, errRvc)
+
+		if rcvTxt != nil {
+			// Check the message bod matches what we expect
+			assert.Equal(t, syncMsgPrefix+strconv.Itoa(rcvCount), *rcvTxt)
+			rcvCount++
+		} else {
+			finishedReceiving = true
+		}
+	}
+
+	// --------------------------------------------------------
+	// Now repeat the experiment but with ASYNC message put
+	asyncQueue := context.CreateQueue("DEV.QUEUE.1").SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+
+	asyncStartTime := currentTimeMillis()
+	for i := 0; i < numberMessages; i++ {
+
+		// Create a TextMessage and send it.
+		msg := context.CreateTextMessageWithString(asyncMsgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(asyncQueue, msg)
+		assert.Nil(t, errSend)
+	}
+	asyncEndTime := currentTimeMillis()
+
+	asyncSendTime := asyncEndTime - asyncStartTime
+	//fmt.Println("Took " + strconv.FormatInt(asyncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " ASYNC messages.")
+
+	// Receive the messages back again
+	finishedReceiving = false
+	rcvCount = 0
+
+	for !finishedReceiving {
+		rcvTxt, errRvc := consumer.ReceiveStringBodyNoWait()
+		assert.Nil(t, errRvc)
+
+		if rcvTxt != nil {
+			// Check the message bod matches what we expect
+			assert.Equal(t, asyncMsgPrefix+strconv.Itoa(rcvCount), *rcvTxt)
+			rcvCount++
+		} else {
+			finishedReceiving = true
+		}
+	}
+
+	assert.Equal(t, numberMessages, rcvCount)
+
+	// Expect that async put is at least 10% faster than sync put for non-persistent messages
+	// (in testing against a remote queue manager it was actually 30% faster)
+	assert.True(t, 100*asyncSendTime < 90*syncSendTime)
+
+}
+
+/*
+ * Test the getter/setter functions for controlling async put.
+ */
+func TestAsyncPutGetterSetter(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 the producer and consumer
+	queue := context.CreateQueue("DEV.QUEUE.1")
+
+	// Check the default
+	assert.Equal(t, jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST, queue.GetPutAsyncAllowed())
+
+	// Check enabled
+	queue = queue.SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+	assert.Equal(t, jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED, queue.GetPutAsyncAllowed())
+
+	// Check disabled
+	queue = queue.SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_DISABLED)
+	assert.Equal(t, jms20subset.Destination_PUT_ASYNC_ALLOWED_DISABLED, queue.GetPutAsyncAllowed())
+
+	// Check as-dest
+	queue = queue.SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST)
+	assert.Equal(t, jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST, queue.GetPutAsyncAllowed())
+
+}

jms20subset/Destination.go

@@ -22,4 +22,32 @@ type Destination interface {
 	// is automatically implemented by every object, so we need to define at least
 	// one method here in order to make it meet the JMS style semantics.
 	GetDestinationName() string
+
+	// SetPutAsyncAllowed controls whether asynchronous put is allowed for this
+	// destination.
+	//
+	// Permitted values are:
+	//  * Destination_PUT_ASYNC_ALLOWED_ENABLED - enables async put
+	//  * Destination_PUT_ASYNC_ALLOWED_DISABLED - disables async put
+	//  * Destination_PUT_ASYNC_ALLOWED_AS_DEST - delegate to queue configuration (default)
+	SetPutAsyncAllowed(paa int) Queue
+
+	// GetPutAsyncAllowed returns whether asynchronous put is configured for this
+	// destination.
+	//
+	// Returned value is one of:
+	//  * Destination_PUT_ASYNC_ALLOWED_ENABLED - async put is enabled
+	//  * Destination_PUT_ASYNC_ALLOWED_DISABLED - async put is disabled
+	//  * Destination_PUT_ASYNC_ALLOWED_AS_DEST - delegated to queue configuration (default)
+	GetPutAsyncAllowed() int
 }
+
+// Destination_PUT_ASYNC_ALLOWED_ENABLED is used to enable messages being sent asynchronously.
+const Destination_PUT_ASYNC_ALLOWED_ENABLED int = 1
+
+// Destination_PUT_ASYNC_ALLOWED_DISABLED is used to disable messages being sent asynchronously.
+const Destination_PUT_ASYNC_ALLOWED_DISABLED int = 0
+
+// Destination_PUT_ASYNC_ALLOWED_AS_DEST allows the async message behaviour to be controlled by
+// the queue on the queue manager.
+const Destination_PUT_ASYNC_ALLOWED_AS_DEST int = -1

jms20subset/Queue.go

@@ -15,14 +15,11 @@ package jms20subset
 // specifies the identity of a queue to the JMS API functions.
 type Queue interface {
 
+	// Encapsulate the root Destination type so that this interface "inherits" the
+	// accessors for standard attributes that apply to all destination types
+	Destination
+
 	// GetQueueName returns the provider-specific name of the queue that is
 	// represented by this object.
 	GetQueueName() string
-
-	// GetDestinationName returns the provider-specific name of the queue that is
-	// represented by this object.
-	//
-	// This method is implemented to allow us to consider the Queue interface
-	// as a specialization of the Destination interface.
-	GetDestinationName() string
 }

mqjms/ContextImpl.go

@@ -30,7 +30,8 @@ func (ctx ContextImpl) CreateQueue(queueName string) jms20subset.Queue {
 
 	// Store the name of the queue
 	queue := QueueImpl{
-		queueName: queueName,
+		queueName:       queueName,
+		putAsyncAllowed: jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST,
 	}
 
 	return queue

mqjms/ProducerImpl.go

@@ -91,6 +91,11 @@ func (producer ProducerImpl) Send(dest jms20subset.Destination, msg jms20subset.
 		// unique message ID
 		pmo.Options = syncpointSetting | ibmmq.MQPMO_NEW_MSG_ID
 
+		// Is async put has been requested then apply the appropriate PMO option
+		if dest.GetPutAsyncAllowed() == jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED {
+			pmo.Options |= ibmmq.MQPMO_ASYNC_RESPONSE
+		}
+
 		// Convert the JMS persistence into the equivalent MQ message descriptor
 		// attribute.
 		if producer.deliveryMode == jms20subset.DeliveryMode_NON_PERSISTENT {

mqjms/QueueImpl.go

@@ -9,10 +9,18 @@
 // Package mqjms provides the implementation of the JMS style Golang interfaces to communicate with IBM MQ.
 package mqjms
 
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
+)
+
 // QueueImpl encapsulates the provider-specific attributes necessary to
 // communicate with an IBM MQ queue.
 type QueueImpl struct {
-	queueName string
+	queueName       string
+	putAsyncAllowed int
 }
 
 // GetQueueName returns the provider-specific name of the queue that is
@@ -30,3 +38,30 @@ func (queue QueueImpl) GetDestinationName() string {
 	return queue.queueName
 
 }
+
+// SetPutAsyncAllowed allows the async allowed setting to be updated.
+func (queue QueueImpl) SetPutAsyncAllowed(paa int) jms20subset.Queue {
+
+	// Check that the specified paa parameter is one of the values that we permit,
+	// and if so store that value inside queue.
+	if paa == jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED ||
+		paa == jms20subset.Destination_PUT_ASYNC_ALLOWED_DISABLED ||
+		paa == jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST {
+
+		queue.putAsyncAllowed = paa
+
+	} 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 PutAsyncAllowed value specified: " + strconv.Itoa(paa))
+	}
+
+	return queue
+}
+
+// GetPutAsyncAllowed returns the current setting for async put.
+func (queue QueueImpl) GetPutAsyncAllowed() int {
+	return queue.putAsyncAllowed
+}