Ability to read a truncated text message - #77

Author: matrober-uk

largemessage_test.go

@@ -331,8 +331,8 @@ func TestTruncatedTextMessage(t *testing.T) {
 		defer consumer.Close()
 	}
 
-	// This will fail because the default buffer size when receiving a
-	// message is 32kb.
+	// 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())
@@ -363,9 +363,8 @@ func TestTruncatedTextMessage(t *testing.T) {
 		defer consumer2.Close()
 	}
 
-	rcvMsg2, errRcv2 := consumer2.ReceiveNoWait()
-	assert.Nil(t, errRcv2)
-	assert.NotNil(t, rcvMsg2)
+	// Attempt to receive a message, and don't worry whether it does or not.
+	consumer2.ReceiveNoWait()
 
 }
 

mqjms/ConsumerImpl.go

@@ -71,13 +71,13 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 
 	getmqmd := ibmmq.NewMQMD()
 
-	myBufferSize := 32768
+	bufferSize := 32768
 
 	if consumer.ctx.receiveBufferSize > 0 {
-		myBufferSize = consumer.ctx.receiveBufferSize
+		bufferSize = consumer.ctx.receiveBufferSize
 	}
 
-	buffer := make([]byte, myBufferSize)
+	buffer := make([]byte, bufferSize)
 
 	// Calculate the syncpoint value
 	syncpointSetting := ibmmq.MQGMO_NO_SYNCPOINT
@@ -88,6 +88,7 @@ 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
 
 	// Include the message properties in the msgHandle
 	gmo.Options |= ibmmq.MQGMO_PROPERTIES_IN_HANDLE
@@ -105,7 +106,24 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 	// Use the prepared objects to ask for a message from the queue.
 	datalen, err := consumer.qObject.Get(getmqmd, gmo, buffer)
 
-	if err == nil {
+	// Establish the read length so that it does not exceed the size of the buffer.
+	readLength := datalen
+	if datalen > bufferSize {
+		readLength = bufferSize
+	}
+
+	// Check whether this is a truncated message.
+	isTruncatedMessage := false
+	if err != nil && ((err.(*ibmmq.MQReturn)).MQRC == ibmmq.MQRC_TRUNCATED_MSG_ACCEPTED) {
+		isTruncatedMessage = true
+
+		// In the truncated message case we want to return the warning object as well as the message
+		// so that the application can tell that some of the data is missing.
+		jmsErr = CreateJMSExceptionFromMQReturn(err)
+	}
+
+	// Golden path - typically a message was received without error.
+	if err == nil || isTruncatedMessage {
 
 		// Set a finalizer on the message handle to allow it to be deleted
 		// when it is no longer referenced by an active object, to reduce/prevent
@@ -119,8 +137,8 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 
 			var msgBodyStr *string
 
-			if datalen > 0 {
-				strContent := string(buffer[:datalen])
+			if readLength > 0 {
+				strContent := string(buffer[:readLength])
 				msgBodyStr = &strContent
 			}
 
@@ -135,11 +153,11 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 
 		} else {
 
-			if datalen == 0 {
+			if readLength == 0 {
 				buffer = []byte{}
 			}
 
-			trimmedBuffer := buffer[0:datalen]
+			trimmedBuffer := buffer[0:readLength]
 
 			// Not a string, so fall back to BytesMessage
 			msg = &BytesMessageImpl{
@@ -172,11 +190,7 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 
 			// Parse the details of the error and return it to the caller as
 			// a JMSException
-			rcInt := int(mqret.MQRC)
-			errCode := strconv.Itoa(rcInt)
-			reason := ibmmq.MQItoString("RC", rcInt)
-
-			jmsErr = jms20subset.CreateJMSException(reason, errCode, err)
+			jmsErr = CreateJMSExceptionFromMQReturn(err)
 		}
 
 	}
@@ -417,3 +431,20 @@ func (consumer ConsumerImpl) Close() {
 
 	return
 }
+
+func CreateJMSExceptionFromMQReturn(mqretError error) jms20subset.JMSException {
+
+	// Assumes this error code was returned from MQ call.
+	mqret := mqretError.(*ibmmq.MQReturn)
+
+	// Parse the details of the error and return it to the caller as
+	// a JMSException
+	rcInt := int(mqret.MQRC)
+	errCode := strconv.Itoa(rcInt)
+	reason := ibmmq.MQItoString("RC", rcInt)
+
+	jmsErr := jms20subset.CreateJMSException(reason, errCode, mqretError)
+
+	return jmsErr
+
+}