Transactional async put support - #32

Author: matrober-uk

asyncput_test.go

@@ -59,7 +59,7 @@ func TestAsyncPutSample(t *testing.T) {
 }
 
 /*
- * Compare the performance benefit of sending messages non-persistent, non-transational
+ * Compare the performance benefit of sending messages non-persistent, non-transactional
  * messages asynchronously - which can give a higher message rate, in exchange for
  * less/no checking for errors.
  *
@@ -298,7 +298,7 @@ func TestAsyncPutCheckCountWithFailure(t *testing.T) {
 		// 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.")
+			fmt.Println("Skipping TestAsyncPutCheckCountWithFailure as queue " + QUEUE_25_NAME + " is not defined.")
 			queueExists = false
 			break // Stop the loop at this point as we know it won't change.
 
@@ -402,3 +402,344 @@ func TestAsyncPutGetterSetter(t *testing.T) {
 	assert.Equal(t, jms20subset.Destination_PUT_ASYNC_ALLOWED_AS_DEST, queue.GetPutAsyncAllowed())
 
 }
+
+/*
+ * Compare the performance benefit of sending messages persistent, transactional
+ * 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 TestAsyncPutPersistentTransactedComparison(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)
+	transactedContext, ctxErr := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, ctxErr)
+	if transactedContext != nil {
+		defer transactedContext.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 := transactedContext.CreateQueue("DEV.QUEUE.1")
+	producer := transactedContext.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_PERSISTENT).SetTimeToLive(60000)
+
+	consumer, errCons := transactedContext.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 (i.e. using normal synchronous put)
+	syncStartTime := currentTimeMillis()
+	for i := 0; i < numberMessages; i++ {
+
+		// Create a TextMessage and send it.
+		msg := transactedContext.CreateTextMessageWithString(syncMsgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(syncQueue, msg)
+		assert.Nil(t, errSend)
+	}
+	transactedContext.Commit()
+	syncEndTime := currentTimeMillis()
+
+	syncSendTime := syncEndTime - syncStartTime
+	//fmt.Println("Took " + strconv.FormatInt(syncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " transacted synchronous messages.")
+
+	// Receive the messages back again to tidy the queue back to a clean state
+	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
+		}
+	}
+	transactedContext.Commit()
+
+	// --------------------------------------------------------
+	// Now repeat the experiment but with ASYNC message put
+	asyncQueue := transactedContext.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 := transactedContext.CreateTextMessageWithString(asyncMsgPrefix + strconv.Itoa(i))
+
+		errSend := producer.Send(asyncQueue, msg)
+		assert.Nil(t, errSend)
+	}
+	transactedContext.Commit()
+	asyncEndTime := currentTimeMillis()
+
+	asyncSendTime := asyncEndTime - asyncStartTime
+	//fmt.Println("Took " + strconv.FormatInt(asyncSendTime, 10) + "ms to send " + strconv.Itoa(numberMessages) + " transacted ASYNC messages.")
+
+	// Receive the messages back again to tidy the queue back to a clean state
+	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
+		}
+	}
+	transactedContext.Commit()
+
+	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)
+
+}
+
+/*
+ * Validate that errors are reported at the correct interval when a problem occurs during a
+ * transactional put of persistent messages (i.e. when the commit 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. In the transacted async put case the Send method always completes
+ * successfully, but the error will be returned on Commit.
+ */
+func TestAsyncPutTransactedCheckCountWithFailure(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)
+
+	// The SendCheckCount is not used for async put under a transacted session.
+	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)
+	transactedContext, ctxErr := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, ctxErr)
+	if transactedContext != nil {
+		defer transactedContext.Close()
+	}
+
+	// Set up the producer and consumer with the async queue.
+	QUEUE_25_NAME := "DEV.MAXDEPTH25"
+	asyncQueue := transactedContext.CreateQueue(QUEUE_25_NAME).SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+	producer := transactedContext.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_PERSISTENT).SetTimeToLive(60000)
+
+	// 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 := transactedContext.CreateTextMessageWithString(msgPrefix + strconv.Itoa(i))
+
+		errSend2 := producer.Send(asyncQueue, msg)
+
+		// Skip the test if the destination does not exist on this queue manager.
+		if i == 0 && errSend2 != nil && errSend2.GetReason() == "MQRC_UNKNOWN_OBJECT_NAME" {
+
+			fmt.Println("Skipping TestAsyncPutTransactedCheckCountWithFailure as queue " + QUEUE_25_NAME + " is not defined.")
+			queueExists = false
+			break // Stop the loop at this point as we know it won't change.
+
+		}
+
+		// In the Transacted case the response from Send is always Nil, because any errors
+		// will be reflected on the Commit call.
+		assert.Nil(t, errSend2)
+
+		if i%10 == 0 {
+			commitErr := transactedContext.Commit()
+
+			if i == 30 || i == 40 {
+
+				assert.NotNil(t, commitErr)
+				assert.Equal(t, "2003", commitErr.GetErrorCode())
+				assert.Equal(t, "MQRC_BACKED_OUT", commitErr.GetReason())
+
+				// Linked error is out normal async put error with message about "N failures"
+				linkedErr1 := commitErr.GetLinkedError()
+				assert.Equal(t, "AsyncPutFailure", linkedErr1.(jms20subset.JMSExceptionImpl).GetErrorCode())
+				assert.True(t, strings.Contains(linkedErr1.(jms20subset.JMSExceptionImpl).GetReason(), "6 failures"))
+				assert.True(t, strings.Contains(linkedErr1.(jms20subset.JMSExceptionImpl).GetReason(), "0 warnings"))
+
+				// Second level linked message should have reason of MQRC_Q_FULL
+				linkedErr2 := linkedErr1.(jms20subset.JMSExceptionImpl).GetLinkedError()
+				assert.NotNil(t, linkedErr2)
+				linkedReason := linkedErr2.(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, commitErr)
+			}
+		}
+	}
+
+	// Clear out the transaction context.
+	transactedContext.Commit()
+
+	// 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 := transactedContext.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
+			}
+		}
+
+		transactedContext.Commit()
+	}
+}
+
+/*
+ * Validate that NO errors are reported when a problem occurs during a
+ * transactional put of a non-persistent message.
+ *
+ * (per https://www.ibm.com/docs/en/ibm-mq/9.1?topic=application-putting-messages-asynchronously-in-mq-classes-jms)
+ *
+ * This test case forces a failure to occur by sending 50 messages to a queue that has had its
+ * maximum depth set to 25.
+ */
+func TestAsyncPutTransactedNonPersistentCheckCountWithFailure(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)
+
+	// The SendCheckCount is not used for async put under a transacted session.
+	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)
+	transactedContext, ctxErr := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, ctxErr)
+	if transactedContext != nil {
+		defer transactedContext.Close()
+	}
+
+	// Set up the producer and consumer with the async queue.
+	QUEUE_25_NAME := "DEV.MAXDEPTH25"
+	asyncQueue := transactedContext.CreateQueue(QUEUE_25_NAME).SetPutAsyncAllowed(jms20subset.Destination_PUT_ASYNC_ALLOWED_ENABLED)
+	producer := transactedContext.CreateProducer().SetDeliveryMode(jms20subset.DeliveryMode_NON_PERSISTENT).SetTimeToLive(60000)
+
+	// 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 := transactedContext.CreateTextMessageWithString(msgPrefix + strconv.Itoa(i))
+
+		errSend2 := producer.Send(asyncQueue, msg)
+
+		// Skip the test if the destination does not exist on this queue manager.
+		if i == 0 && errSend2 != nil && errSend2.GetReason() == "MQRC_UNKNOWN_OBJECT_NAME" {
+
+			fmt.Println("Skipping TestAsyncPutTransactedNonPersistentCheckCountWithFailure as queue " + QUEUE_25_NAME + " is not defined.")
+			queueExists = false
+			break // Stop the loop at this point as we know it won't change.
+
+		}
+
+		// In the Transacted case the response from Send is always Nil, because any errors
+		// will be reflected on the Commit call.
+		assert.Nil(t, errSend2)
+
+		if i%10 == 0 {
+			commitErr := transactedContext.Commit()
+
+			// No errors reported for NonPersistent messages in a transaction.
+			assert.Nil(t, commitErr)
+
+		}
+	}
+
+	// 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 := transactedContext.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
+			}
+		}
+
+		transactedContext.Commit()
+	}
+}