Check on a defined interval for errors from async put - #32

Author: matrober-uk

README.md

@@ -124,6 +124,7 @@ your own error handling or logging.
 * 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)
+* Asynchronous put - [asyncput_test.go](asyncput_test.go)
 
 As normal with Go, you can run any individual testcase by executing a command such as;
 ```bash

asyncput_test.go

@@ -10,7 +10,9 @@
 package main
 
 import (
+	"fmt"
 	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
@@ -19,10 +21,10 @@ import (
 )
 
 /*
- * Test the ability to send a message asynchronously, which can give a higher
+ * Minimal example showing how 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) {
+func TestAsyncPutSample(t *testing.T) {
 
 	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
 	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
@@ -36,6 +38,55 @@ func TestAsyncPut(t *testing.T) {
 		defer context.Close()
 	}
 
+	// Set up a Producer for NonPersistent messages and Destination the PutAsyncAllowed=true
+	producer := context.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_NON_PERSISTENT)
+	asyncQueue := context.CreateQueue("DEV.QUEUE.1").SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+
+	// Send a message (asynchronously)
+	msg := context.CreateTextMessageWithString("some text")
+	errSend := producer.Send(asyncQueue, msg)
+	assert.Nil(t, errSend)
+
+	// Tidy up the message to leave the test clean.
+	consumer, errCons := context.CreateConsumer(asyncQueue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+	_, errRvc := consumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRvc)
+
+}
+
+/*
+ * Compare the performance benefit of sending messages non-persistent, non-transational
+ * messages asynchronously - which can give a higher message rate, in exchange for
+ * less/no checking for errors.
+ *
+ * The test checks that async put is at least 10% faster than synchronous put.
+ * (in testing against a remote queue manager it was actually 30% faster)
+ */
+func TestAsyncPutComparison(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)
+
+	// Check the default value for SendCheckCount, which means never check for errors.
+	assert.Equal(t, 0, cf.SendCheckCount)
+
+	// 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()
+	}
+
+	// --------------------------------------------------------
+	// Start by sending a set of messages using the normal synchronous approach, in
+	// order that we can get a baseline timing.
+
 	// 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)
@@ -53,7 +104,7 @@ func TestAsyncPut(t *testing.T) {
 	numberMessages := 50
 
 	// First get a baseline for how long it takes us to send the batch of messages
-	// WITHOUT async put.
+	// WITHOUT async put (i.e. using normal synchronous put)
 	syncStartTime := currentTimeMillis()
 	for i := 0; i < numberMessages; i++ {
 
@@ -68,7 +119,7 @@ func TestAsyncPut(t *testing.T) {
 	syncSendTime := syncEndTime - syncStartTime
 	//fmt.Println("Took " + strconv.FormatInt(syncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " synchronous messages.")
 
-	// Receive the messages back again
+	// Receive the messages back again to tidy the queue back to a clean state
 	finishedReceiving := false
 	rcvCount := 0
 
@@ -103,7 +154,7 @@ func TestAsyncPut(t *testing.T) {
 	asyncSendTime := asyncEndTime - asyncStartTime
 	//fmt.Println("Took " + strconv.FormatInt(asyncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " ASYNC messages.")
 
-	// Receive the messages back again
+	// Receive the messages back again to tidy the queue back to a clean state
 	finishedReceiving = false
 	rcvCount = 0
 
@@ -128,6 +179,193 @@ func TestAsyncPut(t *testing.T) {
 
 }
 
+/*
+ * Test the ability to successfully send async messages with checking enabled.
+ *
+ * This test is checking that no failures are reported when the interval checking
+ * is enabled.
+ */
+func TestAsyncPutCheckCount(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)
+
+	// Set the CF flag to enable checking for errors after a certain number of messages
+	cf.SendCheckCount = 10
+
+	// Check the default value for SendCheckCount
+	assert.Equal(t, 10, cf.SendCheckCount)
+
+	// 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 async queue.
+	asyncQueue := context.CreateQueue("DEV.QUEUE.1").SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+	producer := context.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_NON_PERSISTENT)
+
+	// Create a unique message prefix representing this execution of the test case.
+	testcasePrefix := strconv.FormatInt(currentTimeMillis(), 10)
+	msgPrefix := "checkCount_" + testcasePrefix + "_"
+	numberMessages := 50
+
+	// --------------------------------------------------------
+	// Do ASYNC message put
+	for i := 0; i < numberMessages; i++ {
+
+		// Create a TextMessage and send it.
+		msg := context.CreateTextMessageWithString(msgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(asyncQueue, msg)
+		assert.Nil(t, errSend)
+	}
+
+	// ----------------------------------
+	// Receive the messages back again to tidy the queue back to a clean state
+	consumer, errCons := context.CreateConsumer(asyncQueue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	finishedReceiving := false
+
+	for !finishedReceiving {
+		rcvMsg, errRvc := consumer.ReceiveNoWait()
+		assert.Nil(t, errRvc)
+
+		if rcvMsg == nil {
+			finishedReceiving = true
+		}
+	}
+}
+
+/*
+ * Validate that errors are reported at the correct interval when a problem occurs.
+ *
+ * This test case forces a failure to occur by sending 50 messages to a queue that has had its
+ * maximum depth set to 25. With SendCheckCount of 10 we will not receive an error until message 30,
+ * which is the first time the error check is made after the point at which the queue has filled up.
+ */
+func TestAsyncPutCheckCountWithFailure(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)
+
+	// Set the CF flag to enable checking for errors after a certain number of messages
+	cf.SendCheckCount = 10
+
+	// Check the value for SendCheckCount was stored correctly.
+	assert.Equal(t, 10, cf.SendCheckCount)
+
+	// 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 async queue.
+	QUEUE_25_NAME := "DEV.MAXDEPTH25"
+	asyncQueue := context.CreateQueue(QUEUE_25_NAME).SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+	producer := context.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_NON_PERSISTENT)
+
+	// Create a unique message prefix representing this execution of the test case.
+	testcasePrefix := strconv.FormatInt(currentTimeMillis(), 10)
+	msgPrefix := "checkCount_" + testcasePrefix + "_"
+	numberMessages := 50
+
+	// Variable to track whether the queue exists or not.
+	queueExists := true
+
+	// --------------------------------------------------------
+	// Send ASYNC message put
+	for i := 0; i < numberMessages; i++ {
+
+		// Create a TextMessage and send it.
+		msg := context.CreateTextMessageWithString(msgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(asyncQueue, msg)
+
+		// Messages will start to fail at number 25 but we don't get an error until
+		// the next check which takes place at 30.
+		if i == 0 && errSend != nil && errSend.GetReason() == "MQRC_UNKNOWN_OBJECT_NAME" {
+
+			fmt.Println("Skipping TestAsyncPutCheckCountWithFailure as " + QUEUE_25_NAME + " is not defined.")
+			queueExists = false
+			break // Stop the loop at this point as we know it won't change.
+
+		} else if i < 30 {
+			assert.Nil(t, errSend)
+		} else if i == 30 {
+
+			assert.NotNil(t, errSend)
+			assert.Equal(t, "AsyncPutFailure", errSend.GetErrorCode())
+
+			// Message should be "N failures"
+			assert.True(t, strings.Contains(errSend.GetReason(), "6 failures"))
+			assert.True(t, strings.Contains(errSend.GetReason(), "0 warnings"))
+
+			// Linked message should have reason of MQRC_Q_FULL
+			linkedErr := errSend.GetLinkedError()
+			assert.NotNil(t, linkedErr)
+			linkedReason := linkedErr.(jms20subset.JMSExceptionImpl).GetReason()
+			assert.Equal(t, "MQRC_Q_FULL", linkedReason)
+
+		} else if i == 40 {
+
+			assert.NotNil(t, errSend)
+			assert.Equal(t, "AsyncPutFailure", errSend.GetErrorCode())
+
+			// Message should be "N failures"
+			assert.True(t, strings.Contains(errSend.GetReason(), "10 failures")) // all of these failed
+			assert.True(t, strings.Contains(errSend.GetReason(), "0 warnings"))
+
+			// Linked message should have reason of MQRC_Q_FULL
+			linkedErr := errSend.GetLinkedError()
+			assert.NotNil(t, linkedErr)
+			linkedReason := linkedErr.(jms20subset.JMSExceptionImpl).GetReason()
+			assert.Equal(t, "MQRC_Q_FULL", linkedReason)
+
+		} else {
+			// Messages 31, 32, ... 39, 41, 42, ...
+			// do not give an error because we don't make an error check.
+			assert.Nil(t, errSend)
+		}
+	}
+
+	// If the queue exists then tidy up the messages we sent.
+	if queueExists {
+
+		// ----------------------------------
+		// Receive the messages back again to tidy the queue back to a clean state
+		consumer, errCons := context.CreateConsumer(asyncQueue)
+		assert.Nil(t, errCons)
+		if consumer != nil {
+			defer consumer.Close()
+		}
+
+		// Receive the messages back again
+		finishedReceiving := false
+
+		for !finishedReceiving {
+			rcvMsg, errRvc := consumer.ReceiveNoWait()
+			assert.Nil(t, errRvc)
+
+			if rcvMsg == nil {
+				finishedReceiving = true
+			}
+		}
+	}
+}
+
 /*
  * Test the getter/setter functions for controlling async put.
  */

go.mod

@@ -3,6 +3,7 @@ module github.com/ibm-messaging/mq-golang-jms20
 go 1.13
 
 require (
+	github.com/ibm-messaging/mq-golang v3.0.0+incompatible
 	github.com/ibm-messaging/mq-golang/v5 v5.1.3
 	github.com/stretchr/testify v1.4.0
 )

go.sum

@@ -3,6 +3,7 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/ibm-messaging/mq-golang v1.0.1-0.20190820103725-19b946c185a8 h1:kUwSXeftVen12FRnShG+Ykhb2Kd6Cd/DbpWwbYal7j0=
 github.com/ibm-messaging/mq-golang v1.0.1-0.20190820103725-19b946c185a8/go.mod h1:qjsZDb7m1oKnbPeDma2JVJTKgyCA91I4bcJ1qHY+gcA=
 github.com/ibm-messaging/mq-golang v3.0.0+incompatible h1:Yc3c8emAyveT54uNDRMkgvS+EBAHeLNWHkc3hk5x+IY=
+github.com/ibm-messaging/mq-golang v3.0.0+incompatible/go.mod h1:qjsZDb7m1oKnbPeDma2JVJTKgyCA91I4bcJ1qHY+gcA=
 github.com/ibm-messaging/mq-golang/v5 v5.0.0 h1:9J8bsDoCo60rbSgB7ZAURPG3L5Kpr+F8dYNOwQ7Qnnk=
 github.com/ibm-messaging/mq-golang/v5 v5.0.0/go.mod h1:ywCwmYbJOU/E0rl+z4GiNoxVMty68O+LVO39a1VMXrE=
 github.com/ibm-messaging/mq-golang/v5 v5.1.2 h1:u0e1Vce2TNqJpH088vF77rDMsnMRWnGaOIlxZo4DMZc=

jms20subset/Destination.go

@@ -26,6 +26,10 @@ type Destination interface {
 	// SetPutAsyncAllowed controls whether asynchronous put is allowed for this
 	// destination.
 	//
+	// See also ConnectionFactoryImpl.SendCheckCount to control the frequency with
+	// which checks will be made for errors. Default of 0 (zero) means no error checks
+	// will be made for errors during async put.
+	//
 	// Permitted values are:
 	//  * Destination_PUT_ASYNC_ALLOWED_ENABLED - enables async put
 	//  * Destination_PUT_ASYNC_ALLOWED_DISABLED - disables async put

jms20subset/JMSException.go

@@ -19,6 +19,7 @@ type JMSException interface {
 	GetReason() string
 	GetErrorCode() string
 	GetLinkedError() error
+	Error() string
 }
 
 // JMSExceptionImpl is a struct that implements the JMSException interface

mqjms/ConnectionFactoryImpl.go

@@ -46,6 +46,15 @@ type ConnectionFactoryImpl struct {
 
 	// Controls the size of the buffer used when receiving a message (default is 32kb if not set)
 	ReceiveBufferSize int
+
+	// SetCheckCount defines the number of messages that will be asynchronously put using
+	// this Context between checks for errors. For example a value of 10 will cause an error
+	// check to be triggered once for every 10 messages.
+	//
+	// See also Destination#SetPutAsyncAllowed(int)
+	//
+	// Default of 0 (zero) means that no checks are made for asynchronous put calls.
+	SendCheckCount int
 }
 
 // CreateContext implements the JMS method to create a connection to an IBM MQ
@@ -131,12 +140,20 @@ func (cf ConnectionFactoryImpl) CreateContextWithSessionMode(sessionMode int) (j
 
 	if err == nil {
 
+		// Initialize the countInc value to 1 so that if CheckCount is enabled (>0)
+		// then an error check will be made after the first message - to catch any
+		// failures quickly.
+		countInc := new(int)
+		*countInc = 1
+
 		// Connection was created successfully, so we wrap the MQI object into
 		// a new ContextImpl and return it to the caller.
 		ctx = ContextImpl{
 			qMgr:              qMgr,
 			sessionMode:       sessionMode,
 			receiveBufferSize: cf.ReceiveBufferSize,
+			sendCheckCount:    cf.SendCheckCount,
+			sendCheckCountInc: countInc,
 		}
 
 	} else {

mqjms/ContextImpl.go

@@ -22,6 +22,8 @@ type ContextImpl struct {
 	qMgr              ibmmq.MQQueueManager
 	sessionMode       int
 	receiveBufferSize int
+	sendCheckCount    int
+	sendCheckCountInc *int // Internal counter to keep track of async-put messages sent
 }
 
 // CreateQueue implements the logic necessary to create a provider-specific