Support for get and get-put under local transaction

Author: matrober-uk

README.md

@@ -118,6 +118,7 @@ your own error handling or logging.
 * Send a message as Persistent or NonPersistent - [deliverymode_test.go](deliverymode_test.go)
 * Get by CorrelationID - [getbycorrelid_test.go](getbycorrelid_test.go)
 * 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)
 * Handle error codes returned by the queue manager - [sample_errorhandling_test.go](sample_errorhandling_test.go)
 

local_transaction_test.go

@@ -125,16 +125,253 @@ func TestPutTransaction(t *testing.T) {
 
 }
 
-// get-transaction
-//   - get without transaction, immediately disappears
-//   - get multiple with transaction, immedatiately disappears
-//   - rollback, multiple reappear
-//   - get multiple with transaction, commit, not available
-
-// get-put transaction
-//   - place initial message on queue
-//   - get message under transaction, put reply message to different queue
-//   - neither request nor reply message is available
-//   - rollback; request message is available, reply message is not
-//   - (again) get message under transaction, put reply message to different queue
-//   - commit; request message is gone, and reply message is available
+/**
+ * Test the behaviour of receiving a message under a transaction.
+ *
+ * - get without transaction, immediately disappears
+ * - get multiple with transaction, immediately disappears
+ * - rollback, multiple reappear
+ * - get with transaction, commit, not available
+ * - receive under transaction, close connection - should be available again (rollback)
+ */
+func TestGetTransaction(t *testing.T) {
+
+	// Create a ConnectionFactory using some property files
+	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	assert.Nil(t, cfErr)
+
+	// Creates a connection to the queue manager.
+	untransactedContext, errCtx := cf.CreateContext()
+	assert.Nil(t, errCtx)
+	if untransactedContext != nil {
+		defer untransactedContext.Close()
+	}
+
+	transactedContext, errCtx := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, errCtx)
+	if transactedContext != nil {
+		defer transactedContext.Close()
+	}
+
+	// Create queue objects that points at an IBM MQ queue
+	queueName := "DEV.QUEUE.1"
+	unQueue := untransactedContext.CreateQueue(queueName)
+	trQueue := transactedContext.CreateQueue(queueName)
+
+	// Create an transacted consumer
+	transactedConsumer, errCons := transactedContext.CreateConsumer(trQueue)
+	assert.Nil(t, errCons)
+	if transactedConsumer != nil {
+		defer transactedConsumer.Close()
+	}
+
+	// Create an untransacted consumer
+	untransactedConsumer, errCons := untransactedContext.CreateConsumer(unQueue)
+	assert.Nil(t, errCons)
+	if untransactedConsumer != nil {
+		defer untransactedConsumer.Close()
+	}
+
+	// Send an untransacted message and check it is immediately available for untransacted receive
+	untransactedProducer := untransactedContext.CreateProducer().SetTimeToLive(20000)
+	bodyTxt := "untransacted-get"
+	errSend := untransactedProducer.SendString(unQueue, bodyTxt)
+	assert.Nil(t, errSend)
+	rcvBody, errRcv := untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt, *rcvBody)
+	rcvBody, errRcv = transactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody) // Has been consumed by the untransacted consumer
+
+	// get multiple with transaction, immediately disappears
+	bodyTxt1 := "transacted-get-1"
+	bodyTxt2 := "transacted-get-2"
+	errSend = untransactedProducer.SendString(unQueue, bodyTxt1)
+	assert.Nil(t, errSend)
+	errSend = untransactedProducer.SendString(unQueue, bodyTxt2)
+	assert.Nil(t, errSend)
+
+	rcvBody, errRcv = transactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt1, *rcvBody)
+	rcvBody, errRcv = transactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt2, *rcvBody)
+
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody) // Message is not available (consumed pending transaction)
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+
+	// rollback, messages reappear
+	transactedContext.Rollback() // puts the two messages back on the queue
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt1, *rcvBody)
+
+	// get the second reappeared message with transaction, commit, not available
+	rcvBody, errRcv = transactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt2, *rcvBody)
+	transactedContext.Commit() // commit the consumption of the one message.
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody) // No message should be available
+
+	// receive under transaction, close connection - should be available again (rollback)
+	bodyTxt3 := "transacted-get-3"
+	errSend = untransactedProducer.SendString(unQueue, bodyTxt3)
+	assert.Nil(t, errSend)
+	rcvBody, errRcv = transactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt3, *rcvBody)
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)    // message not available
+	transactedContext.Close() // causes rollback
+	rcvBody, errRcv = untransactedConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, bodyTxt3, *rcvBody) // message now available
+
+}
+
+/**
+ * Test the behaviour of receiving and sending a message under the same local transaction.
+ *
+ * - place initial message on queue
+ * - get message under transaction, put reply message to different queue
+ * - neither request nor reply message is available
+ * - rollback; request message is available, reply message is not
+ * - (again) get message under transaction, put reply message to different queue
+ * - commit; request message is gone, and reply message is available
+ */
+func TestPutGetTransaction(t *testing.T) {
+
+	// Create a ConnectionFactory using some property files
+	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	assert.Nil(t, cfErr)
+
+	// Creates a connection to the queue manager.
+	untransactedContext, errCtx := cf.CreateContext()
+	assert.Nil(t, errCtx)
+	if untransactedContext != nil {
+		defer untransactedContext.Close()
+	}
+
+	transactedContext, errCtx := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, errCtx)
+	if transactedContext != nil {
+		defer transactedContext.Close()
+	}
+
+	senderTransactedContext, errCtx := cf.CreateContextWithSessionMode(jms20subset.JMSContextSESSIONTRANSACTED)
+	assert.Nil(t, errCtx)
+	if senderTransactedContext != nil {
+		defer senderTransactedContext.Close()
+	}
+
+	// Create Request + Reply queue objects that points at IBM MQ queues
+	reqQueueName := "DEV.QUEUE.1"
+	unReqQueue := untransactedContext.CreateQueue(reqQueueName)
+	trReqQueue := transactedContext.CreateQueue(reqQueueName)
+
+	replyQueueName := "DEV.QUEUE.2"
+	unReplyQueue := untransactedContext.CreateQueue(replyQueueName)
+	trReplyQueue := transactedContext.CreateQueue(replyQueueName)
+
+	// Create an unrelated transacted producer (different connection)
+	transactedReqProducer := senderTransactedContext.CreateProducer().SetTimeToLive(20000)
+
+	// Create an transacted consumer for the request queue
+	transactedReqConsumer, errCons := transactedContext.CreateConsumer(trReqQueue)
+	assert.Nil(t, errCons)
+	if transactedReqConsumer != nil {
+		defer transactedReqConsumer.Close()
+	}
+
+	// Create an untransacted consumer for the request queue
+	untransactedReqConsumer, errCons := untransactedContext.CreateConsumer(unReqQueue)
+	assert.Nil(t, errCons)
+	if untransactedReqConsumer != nil {
+		defer untransactedReqConsumer.Close()
+	}
+
+	// Create a transacted producer for the reply
+	transactedReplyProducer := transactedContext.CreateProducer().SetTimeToLive(20000)
+
+	// Create an untransacted consumer for the reply queue
+	untransactedReplyConsumer, errCons := untransactedContext.CreateConsumer(unReplyQueue)
+	assert.Nil(t, errCons)
+	if untransactedReplyConsumer != nil {
+		defer untransactedReplyConsumer.Close()
+	}
+
+	// First check that both queues are empty
+	rcvBody, errRcv := untransactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+	rcvBody, errRcv = untransactedReplyConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+
+	// Use the transacted sender context to send a request message (under a transaction)
+	msgBody := "putget-transaction"
+	errSend := transactedReqProducer.SendString(trReqQueue, msgBody)
+	assert.Nil(t, errSend)
+	rcvBody, errRcv = untransactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)           // Not yet visible for consumption
+	senderTransactedContext.Commit() // Make the message visible
+
+	// get message under transaction, put reply message to different queue
+	rcvBody, errRcv = transactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, msgBody, *rcvBody)
+	replyMsgBody := "putget-transaction-reply"
+	errSend = transactedReplyProducer.SendString(trReplyQueue, replyMsgBody)
+	assert.Nil(t, errSend)
+
+	// neither request nor reply message is available
+	rcvBody, errRcv = untransactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+	rcvBody, errRcv = untransactedReplyConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+
+	// rollback; request message is available, reply message is not
+	transactedContext.Rollback()
+	rcvBody, errRcv = untransactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, msgBody, *rcvBody)
+	rcvBody, errRcv = untransactedReplyConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+
+	// Put a new request message
+	msgBody2 := "putget-transaction-2"
+	errSend = transactedReqProducer.SendString(trReqQueue, msgBody2)
+	assert.Nil(t, errSend)
+	senderTransactedContext.Commit()
+
+	// (again) get message under transaction, put reply message to the other queue
+	rcvBody, errRcv = transactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, msgBody2, *rcvBody)
+	replyMsgBody2 := "putget-transaction-reply-2"
+	errSend = transactedReplyProducer.SendString(trReplyQueue, replyMsgBody2)
+	assert.Nil(t, errSend)
+
+	// commit; request message is gone, and reply message is available
+	transactedContext.Commit()
+	rcvBody, errRcv = untransactedReqConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Nil(t, rcvBody)
+	rcvBody, errRcv = untransactedReplyConsumer.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv)
+	assert.Equal(t, replyMsgBody2, *rcvBody)
+
+}

mqjms/ConsumerImpl.go

@@ -21,6 +21,7 @@ import (
 // ConsumerImpl defines a struct that contains the necessary objects for
 // receiving messages from a queue on an IBM MQ queue manager.
 type ConsumerImpl struct {
+	ctx      ContextImpl
 	qObject  ibmmq.MQObject
 	selector string
 }
@@ -63,8 +64,14 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 	getmqmd := ibmmq.NewMQMD()
 	buffer := make([]byte, 32768)
 
+	// Calculate the syncpoint value
+	syncpointSetting := ibmmq.MQGMO_NO_SYNCPOINT
+	if consumer.ctx.sessionMode == jms20subset.JMSContextSESSIONTRANSACTED {
+		syncpointSetting = ibmmq.MQGMO_SYNCPOINT
+	}
+
 	// Set the GMO (get message options)
-	gmo.Options |= ibmmq.MQGMO_NO_SYNCPOINT
+	gmo.Options |= syncpointSetting
 	gmo.Options |= ibmmq.MQGMO_FAIL_IF_QUIESCING
 
 	// Apply the selector if one has been specified in the Consumer

mqjms/ContextImpl.go

@@ -90,6 +90,7 @@ func (ctx ContextImpl) CreateConsumerWithSelector(dest jms20subset.Destination,
 		// Success - store the necessary objects away for later use to receive
 		// messages.
 		consumer = ConsumerImpl{
+			ctx:      ctx,
 			qObject:  qObject,
 			selector: selector,
 		}