Merge pull request #78 from ibm-messaging/accept_truncated_msg

Add support for handling truncated messages

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)
 

jms20subset/JMSException.go

@@ -24,9 +24,10 @@ type JMSException interface {
 
 // JMSExceptionImpl is a struct that implements the JMSException interface
 type JMSExceptionImpl struct {
-	reason    string
-	errorCode string
-	linkedErr error
+	reason        string
+	errorCode     string
+	linkedErr     error
+	messageLength int
 }
 
 // GetReason returns the provider-specific reason string describing the error.
@@ -51,6 +52,15 @@ func (ex JMSExceptionImpl) GetLinkedError() error {
 
 }
 
+// GetMessageLength gives the data length that was returned by a Receive (GET) call,
+// for example if the message was truncated, or could not be read because the receive
+// buffer was too small.
+func (ex JMSExceptionImpl) GetMessageLength() int {
+
+	return ex.messageLength
+
+}
+
 // Error allows the JMSExceptionImpl struct to be treated as a Golang error,
 // while also returning a human readable string representation of the error.
 func (ex JMSExceptionImpl) Error() string {
@@ -66,11 +76,17 @@ func (ex JMSExceptionImpl) Error() string {
 
 // CreateJMSException is a helper function for creating a JMSException
 func CreateJMSException(reason string, errorCode string, linkedErr error) JMSException {
+	return CreateJMSExceptionWithExtraParams(reason, errorCode, linkedErr, 0)
+}
+
+// CreateJMSException is a helper function for creating a JMSException
+func CreateJMSExceptionWithExtraParams(reason string, errorCode string, linkedErr error, messageLength int) JMSException {
 
 	ex := JMSExceptionImpl{
-		reason:    reason,
-		errorCode: errorCode,
-		linkedErr: linkedErr,
+		reason:        reason,
+		errorCode:     errorCode,
+		linkedErr:     linkedErr,
+		messageLength: messageLength,
 	}
 
 	return ex

largemessage_test.go

@@ -62,9 +62,20 @@ func TestLargeTextMessage(t *testing.T) {
 
 	// The message is still left on the queue since it failed to be received successfully.
 
-	// Since the buffer size is configured using the ConnectionFactoy we will
-	// create a second connection in order to successfully retrieve the message.
-	cf.ReceiveBufferSize = len(txtOver32kb) + 50
+	// Use a special attribute of the returned exception to look up what the actual length of the
+	// message is, so that we can correctly increase the buffer size in order to receive the message.
+	switch jmsExc := errRcv.(type) {
+	case jms20subset.JMSExceptionImpl:
+		realMessageLength := jmsExc.GetMessageLength()
+		assert.Equal(t, len(txtOver32kb), realMessageLength) // check it matches the original (long) length
+
+		// Since the buffer size is configured using the ConnectionFactory we will
+		// create a second connection in order to successfully retrieve the message.
+		cf.ReceiveBufferSize = realMessageLength
+
+	default:
+		assert.Fail(t, "Got something other than a JMSExceptionImpl")
+	}
 
 	context2, ctxErr2 := cf.CreateContext()
 	assert.Nil(t, ctxErr2)
@@ -78,7 +89,7 @@ func TestLargeTextMessage(t *testing.T) {
 		defer consumer2.Close()
 	}
 
-	rcvMsg2, errRcv2 := consumer2.ReceiveNoWait()
+	rcvMsg2, errRcv2 := consumer2.ReceiveNoWait() // receive the message using the correct (larger) buffer size
 	assert.Nil(t, errRcv2)
 	assert.NotNil(t, rcvMsg2)
 
@@ -134,9 +145,20 @@ func TestLargeReceiveStringBodyTextMessage(t *testing.T) {
 
 	// The message is still left on the queue since it failed to be received successfully.
 
-	// Since the buffer size is configured using the ConnectionFactoy we will
-	// create a second connection in order to successfully retrieve the message.
-	cf.ReceiveBufferSize = len(txtOver32kb) + 50
+	// Use a special attribute of the returned exception to look up what the actual length of the
+	// message is, so that we can correctly increase the buffer size in order to receive the message.
+	switch jmsExc := errRcv.(type) {
+	case jms20subset.JMSExceptionImpl:
+		realMessageLength := jmsExc.GetMessageLength()
+		assert.Equal(t, len(txtOver32kb), realMessageLength) // check it matches the original (long) length
+
+		// Since the buffer size is configured using the ConnectionFactory we will
+		// create a second connection in order to successfully retrieve the message.
+		cf.ReceiveBufferSize = realMessageLength
+
+	default:
+		assert.Fail(t, "Got something other than a JMSExceptionImpl")
+	}
 
 	context2, ctxErr2 := cf.CreateContext()
 	assert.Nil(t, ctxErr2)
@@ -200,9 +222,20 @@ func TestLargeBytesMessage(t *testing.T) {
 
 	// The message is still left on the queue since it failed to be received successfully.
 
-	// Since the buffer size is configured using the ConnectionFactoy we will
-	// create a second connection in order to successfully retrieve the message.
-	cf.ReceiveBufferSize = len(bytesOver32kb) + 50
+	// Use a special attribute of the returned exception to look up what the actual length of the
+	// message is, so that we can correctly increase the buffer size in order to receive the message.
+	switch jmsExc := errRcv.(type) {
+	case jms20subset.JMSExceptionImpl:
+		realMessageLength := jmsExc.GetMessageLength()
+		assert.Equal(t, len(txtOver32kb), realMessageLength) // check it matches the original (long) length
+
+		// Since the buffer size is configured using the ConnectionFactory we will
+		// create a second connection in order to successfully retrieve the message.
+		cf.ReceiveBufferSize = realMessageLength
+
+	default:
+		assert.Fail(t, "Got something other than a JMSExceptionImpl")
+	}
 
 	context2, ctxErr2 := cf.CreateContext()
 	assert.Nil(t, ctxErr2)
@@ -273,9 +306,20 @@ func TestLargeReceiveBytesBodyBytesMessage(t *testing.T) {
 
 	// The message is still left on the queue since it failed to be received successfully.
 
-	// Since the buffer size is configured using the ConnectionFactoy we will
-	// create a second connection in order to successfully retrieve the message.
-	cf.ReceiveBufferSize = len(bytesOver32kb) + 50
+	// Use a special attribute of the returned exception to look up what the actual length of the
+	// message is, so that we can correctly increase the buffer size in order to receive the message.
+	switch jmsExc := errRcv.(type) {
+	case jms20subset.JMSExceptionImpl:
+		realMessageLength := jmsExc.GetMessageLength()
+		assert.Equal(t, len(txtOver32kb), realMessageLength) // check it matches the original (long) length
+
+		// Since the buffer size is configured using the ConnectionFactory we will
+		// create a second connection in order to successfully retrieve the message.
+		cf.ReceiveBufferSize = realMessageLength
+
+	default:
+		assert.Fail(t, "Got something other than a JMSExceptionImpl")
+	}
 
 	context2, ctxErr2 := cf.CreateContext()
 	assert.Nil(t, ctxErr2)
@@ -295,6 +339,168 @@ func TestLargeReceiveBytesBodyBytesMessage(t *testing.T) {
 
 }
 
+/*
+ * 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
+	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()
+
+	// Create a TextMessage with it
+	msg := context.CreateTextMessageWithString(txtOver32kb)
+
+	// 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 txtTruncMsg := truncMsg.(type) {
+	case jms20subset.TextMessage:
+		receivedTxt := txtTruncMsg.GetText()
+		assert.Equal(t, cf.ReceiveBufferSize, len(*receivedTxt))
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+	// Use a special attribute of the returned exception to look up what the actual length of the
+	// message is, so that we can tidy up the message (read successfully) in the event the previous
+	// step failed.
+	switch jmsExc := errRcv.(type) {
+	case jms20subset.JMSExceptionImpl:
+		realMessageLength := jmsExc.GetMessageLength()
+		assert.Equal(t, len(txtOver32kb), realMessageLength) // check it matches the original (long) length
+
+		// Since the buffer size is configured using the ConnectionFactory we will
+		// create a second connection in order to successfully retrieve the message.
+		cf.ReceiveBufferSize = realMessageLength
+
+	default:
+		assert.Fail(t, "Got something other than a JMSExceptionImpl")
+	}
+
+	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()
+	}
+
+	// If the first part of this text was successful then there should be no message to receive.
+	tidyMsg, tidyErr := consumer2.ReceiveNoWait()
+	assert.Nil(t, tidyErr)
+	assert.Nil(t, tidyMsg)
+
+}
+
+/*
+ * 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,
 		}
 
 	}