Add ConnectionFactory option for AcceptTruncatedMessage - #77

Author: matrober-uk

README.md

@@ -128,6 +128,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)
+* Receive a truncated message body for the case where the body is larger than the allowed buffer size - [largemessage_test.go#301](largemessage_test.go#301)
 * Asynchronous put - [asyncput_test.go](asyncput_test.go)
 * Special header properties such as JMS_IBM_Format - [specialproperties_test.go](specialproperties_test.go)
 

largemessage_test.go

@@ -296,14 +296,14 @@ func TestLargeReceiveBytesBodyBytesMessage(t *testing.T) {
 }
 
 /*
- * Test receiving a truncated message, where the body of the message is larger than the receive buffer.
+ * Test receiving a truncated text message, where the body of the message is larger than the receive buffer.
  */
 func TestTruncatedTextMessage(t *testing.T) {
 
 	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
 	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
 	cf.ReceiveBufferSize = 1024
-	// TODO - set parameter to allow receive of truncated message
+	cf.AcceptTruncatedMessage = true
 	assert.Nil(t, cfErr)
 
 	// Creates a connection to the queue manager, using defer to close it automatically
@@ -368,6 +368,80 @@ func TestTruncatedTextMessage(t *testing.T) {
 
 }
 
+/*
+ * Test receiving a truncated bytes message, where the body of the message is larger than the receive buffer.
+ */
+func TestTruncatedBytesMessage(t *testing.T) {
+
+	// Loads CF parameters from connection_info.json and applicationApiKey.json in the Downloads directory
+	cf, cfErr := mqjms.CreateConnectionFactoryFromDefaultJSONFiles()
+	cf.ReceiveBufferSize = 1024
+	cf.AcceptTruncatedMessage = true
+	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()
+	}
+
+	// Get a long text string over 32kb in length
+	txtOver32kb := getStringOver32kb()
+	bytesOver32kb := []byte(txtOver32kb)
+
+	// Create a TextMessage with it
+	msg := context.CreateBytesMessageWithBytes(bytesOver32kb)
+
+	// Now send the message and get it back again.
+	queue := context.CreateQueue("DEV.QUEUE.1")
+	errSend := context.CreateProducer().SetTimeToLive(5000).Send(queue, msg)
+	assert.Nil(t, errSend)
+
+	consumer, errCons := context.CreateConsumer(queue) // with 1kb buffer size as applied at the beginning of this test
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// Since we have set the flag to allow a truncated message to be returned, this will pass
+	// but will only return the first 1024 bytes (cf.ReceiveBufferSize) of the message.
+	truncMsg, errRcv := consumer.ReceiveNoWait()
+	assert.NotNil(t, errRcv)
+	assert.Equal(t, "MQRC_TRUNCATED_MSG_ACCEPTED", errRcv.GetReason())
+	assert.Equal(t, "2079", errRcv.GetErrorCode())
+
+	assert.NotNil(t, truncMsg) // We have received the message, but it is truncated.
+
+	switch bytesTruncMsg := truncMsg.(type) {
+	case jms20subset.BytesMessage:
+		receivedBytes := bytesTruncMsg.ReadBytes()
+		assert.Equal(t, cf.ReceiveBufferSize, len(*receivedBytes))
+	default:
+		assert.Fail(t, "Got something other than a bytes message")
+	}
+
+	// Make sure we tidy up in case the previous part of the test failed.
+	cf.ReceiveBufferSize = len(txtOver32kb) + 50
+
+	context2, ctxErr2 := cf.CreateContext()
+	assert.Nil(t, ctxErr2)
+	if context2 != nil {
+		defer context2.Close()
+	}
+
+	consumer2, errCons2 := context2.CreateConsumer(queue)
+	assert.Nil(t, errCons2)
+	if consumer2 != nil {
+		defer consumer2.Close()
+	}
+
+	// Attempt to receive a message, and don't worry whether it does or not.
+	consumer2.ReceiveNoWait()
+
+}
+
 func getStringOver32kb() string {
 
 	// Build a text string which is over 32KB (in a not very efficient way!)

mqjms/ConnectionFactoryImpl.go

@@ -48,6 +48,9 @@ type ConnectionFactoryImpl struct {
 	// Controls the size of the buffer used when receiving a message (default is 32kb if not set)
 	ReceiveBufferSize int
 
+	// Controls whether a message should be returned if the body will be truncated (because the body is larger than the ReceiveBufferSize)
+	AcceptTruncatedMessage bool
+
 	// 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.
@@ -155,12 +158,13 @@ func (cf ConnectionFactoryImpl) CreateContextWithSessionMode(sessionMode int, mq
 		// Connection was created successfully, so we wrap the MQI object into
 		// a new ContextImpl and return it to the caller.
 		ctx = ContextImpl{
-			qMgr:              qMgr,
-			ctxLock:           &sync.Mutex{},
-			sessionMode:       sessionMode,
-			receiveBufferSize: cf.ReceiveBufferSize,
-			sendCheckCount:    cf.SendCheckCount,
-			sendCheckCountInc: countInc,
+			qMgr:                   qMgr,
+			ctxLock:                &sync.Mutex{},
+			sessionMode:            sessionMode,
+			receiveBufferSize:      cf.ReceiveBufferSize,
+			acceptTruncatedMessage: cf.AcceptTruncatedMessage,
+			sendCheckCount:         cf.SendCheckCount,
+			sendCheckCountInc:      countInc,
 		}
 
 	}

mqjms/ConsumerImpl.go

@@ -88,7 +88,11 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 	// Set the GMO (get message options)
 	gmo.Options |= syncpointSetting
 	gmo.Options |= ibmmq.MQGMO_FAIL_IF_QUIESCING
-	gmo.Options |= ibmmq.MQGMO_ACCEPT_TRUNCATED_MSG
+
+	// Allow a truncated message to be returned if the user has requested it (default is to return an error on get)
+	if consumer.ctx.acceptTruncatedMessage {
+		gmo.Options |= ibmmq.MQGMO_ACCEPT_TRUNCATED_MSG
+	}
 
 	// Include the message properties in the msgHandle
 	gmo.Options |= ibmmq.MQGMO_PROPERTIES_IN_HANDLE

mqjms/ContextImpl.go

@@ -21,12 +21,13 @@ import (
 // ContextImpl encapsulates the objects necessary to maintain an active
 // connection to an IBM MQ queue manager.
 type ContextImpl struct {
-	qMgr              ibmmq.MQQueueManager
-	ctxLock           *sync.Mutex // Mutex to synchronize MQRC calls to the queue manager
-	sessionMode       int
-	receiveBufferSize int
-	sendCheckCount    int
-	sendCheckCountInc *int // Internal counter to keep track of async-put messages sent
+	qMgr                   ibmmq.MQQueueManager
+	ctxLock                *sync.Mutex // Mutex to synchronize MQRC calls to the queue manager
+	sessionMode            int
+	acceptTruncatedMessage bool
+	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