Merge pull request #26 from ibm-messaging/buffer-length

Make receive buffer length configurable - #9

Author: matrober-uk

README.md

@@ -123,7 +123,7 @@ your own error handling or logging.
 * 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)
 * 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)
 
 As normal with Go, you can run any individual testcase by executing a command such as;
 ```bash

largemessage_test.go

@@ -0,0 +1,314 @@
+/*
+ * Copyright (c) IBM Corporation 2019
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ibm-messaging/mq-golang-jms20/jms20subset"
+	"github.com/ibm-messaging/mq-golang-jms20/mqjms"
+	"github.com/stretchr/testify/assert"
+)
+
+/*
+ * Test the send/receive of a large Text message, over the default 32KB size.
+ */
+func TestLargeTextMessage(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)
+
+	// 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(30000).Send(queue, msg)
+	assert.Nil(t, errSend)
+
+	consumer, errCons := context.CreateConsumer(queue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// This will fail because the default buffer size when receiving a
+	// message is 32kb.
+	_, errRcv := consumer.ReceiveNoWait()
+	assert.NotNil(t, errRcv)
+	assert.Equal(t, "MQRC_TRUNCATED_MSG_FAILED", errRcv.GetReason())
+	assert.Equal(t, "2080", errRcv.GetErrorCode())
+
+	// 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
+
+	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()
+	}
+
+	rcvMsg2, errRcv2 := consumer2.ReceiveNoWait()
+	assert.Nil(t, errRcv2)
+	assert.NotNil(t, rcvMsg2)
+
+	switch msg2 := rcvMsg2.(type) {
+	case jms20subset.TextMessage:
+		assert.Equal(t, &txtOver32kb, msg2.GetText())
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+}
+
+/*
+ * Test the send/receive of a large Text message, over the default 32KB size.
+ */
+func TestLargeReceiveStringBodyTextMessage(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)
+
+	// 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(30000).Send(queue, msg)
+	assert.Nil(t, errSend)
+
+	consumer, errCons := context.CreateConsumer(queue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// This will fail because the default buffer size when receiving a
+	// message is 32kb.
+	_, errRcv := consumer.ReceiveStringBodyNoWait()
+	assert.NotNil(t, errRcv)
+	assert.Equal(t, "MQRC_TRUNCATED_MSG_FAILED", errRcv.GetReason())
+	assert.Equal(t, "2080", errRcv.GetErrorCode())
+
+	// 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
+
+	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()
+	}
+
+	rcvStr, errRcv2 := consumer2.ReceiveStringBodyNoWait()
+	assert.Nil(t, errRcv2)
+	assert.Equal(t, &txtOver32kb, rcvStr)
+
+}
+
+/*
+ * Test the send/receive of a large Bytes message, over the default 32KB size.
+ */
+func TestLargeBytesMessage(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)
+
+	// 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(30000).Send(queue, msg)
+	assert.Nil(t, errSend)
+
+	consumer, errCons := context.CreateConsumer(queue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// This will fail because the default buffer size when receiving a
+	// message is 32kb.
+	_, errRcv := consumer.ReceiveNoWait()
+	assert.NotNil(t, errRcv)
+	assert.Equal(t, "MQRC_TRUNCATED_MSG_FAILED", errRcv.GetReason())
+	assert.Equal(t, "2080", errRcv.GetErrorCode())
+
+	// 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
+
+	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()
+	}
+
+	rcvMsg2, errRcv2 := consumer2.ReceiveNoWait()
+	assert.Nil(t, errRcv2)
+	assert.NotNil(t, rcvMsg2)
+
+	switch msg2 := rcvMsg2.(type) {
+	case jms20subset.BytesMessage:
+		assert.Equal(t, &bytesOver32kb, msg2.ReadBytes())
+	default:
+		assert.Fail(t, "Got something other than a text message")
+	}
+
+}
+
+/*
+ * Test the send/receive of a large Bytes message, over the default 32KB size.
+ */
+func TestLargeReceiveBytesBodyBytesMessage(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)
+
+	// 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(30000).Send(queue, msg)
+	assert.Nil(t, errSend)
+
+	consumer, errCons := context.CreateConsumer(queue)
+	assert.Nil(t, errCons)
+	if consumer != nil {
+		defer consumer.Close()
+	}
+
+	// This will fail because the default buffer size when receiving a
+	// message is 32kb.
+	_, errRcv := consumer.ReceiveBytesBodyNoWait()
+	assert.NotNil(t, errRcv)
+	assert.Equal(t, "MQRC_TRUNCATED_MSG_FAILED", errRcv.GetReason())
+	assert.Equal(t, "2080", errRcv.GetErrorCode())
+
+	// 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
+
+	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()
+	}
+
+	rcvBytes2, errRcv2 := consumer2.ReceiveBytesBodyNoWait()
+	assert.Nil(t, errRcv2)
+	assert.Equal(t, &bytesOver32kb, rcvBytes2)
+
+}
+
+func getStringOver32kb() string {
+
+	// Build a text string which is over 32KB (in a not very efficient way!)
+	// First snippet is 100 chars long.
+	txt100chars := "The quick brown fox jumped over the lazy dog into the flowing river that rushed over the cliff edge."
+	txt1300chars := string(time.Now().UnixNano()) + txt100chars + txt100chars + txt100chars + txt100chars + txt100chars +
+		txt100chars + txt100chars + txt100chars + txt100chars + txt100chars +
+		txt100chars + txt100chars + txt100chars
+
+	txtOver32kb := txt1300chars
+
+	for len(txtOver32kb) < 32*1024 {
+		txtOver32kb += txt1300chars
+	}
+
+	return txtOver32kb
+
+}

mqjms/ConnectionFactoryImpl.go

@@ -43,6 +43,9 @@ type ConnectionFactoryImpl struct {
 
 	// Allthough only available per MQ 9.1.2 it looks like a good idea to have this present in MQ-JMS
 	ApplName string
+
+	// Controls the size of the buffer used when receiving a message (default is 32kb if not set)
+	ReceiveBufferSize int
 }
 
 // CreateContext implements the JMS method to create a connection to an IBM MQ
@@ -131,8 +134,9 @@ func (cf ConnectionFactoryImpl) CreateContextWithSessionMode(sessionMode int) (j
 		// Connection was created successfully, so we wrap the MQI object into
 		// a new ContextImpl and return it to the caller.
 		ctx = ContextImpl{
-			qMgr:        qMgr,
-			sessionMode: sessionMode,
+			qMgr:              qMgr,
+			sessionMode:       sessionMode,
+			receiveBufferSize: cf.ReceiveBufferSize,
 		}
 
 	} else {

mqjms/ConsumerImpl.go

@@ -62,7 +62,14 @@ func (consumer ConsumerImpl) receiveInternal(gmo *ibmmq.MQGMO) (jms20subset.Mess
 	var jmsErr jms20subset.JMSException
 
 	getmqmd := ibmmq.NewMQMD()
-	buffer := make([]byte, 32768)
+
+	myBufferSize := 32768
+
+	if consumer.ctx.receiveBufferSize > 0 {
+		myBufferSize = consumer.ctx.receiveBufferSize
+	}
+
+	buffer := make([]byte, myBufferSize)
 
 	// Calculate the syncpoint value
 	syncpointSetting := ibmmq.MQGMO_NO_SYNCPOINT

mqjms/ContextImpl.go

@@ -19,8 +19,9 @@ import (
 // ContextImpl encapsulates the objects necessary to maintain an active
 // connection to an IBM MQ queue manager.
 type ContextImpl struct {
-	qMgr        ibmmq.MQQueueManager
-	sessionMode int
+	qMgr              ibmmq.MQQueueManager
+	sessionMode       int
+	receiveBufferSize int
 }
 
 // CreateQueue implements the logic necessary to create a provider-specific

next-features.txt

@@ -11,6 +11,7 @@ Not currently implemented:
 - Message Properties etc
 - Temporary destinations
 - Priority
+- Configurable option to auto-set the receive buffer length if the default 32kb is exceeded (less efficient that setting up front)
 
 Client capabilities for participating in Uniform Clusters;
 - CCDT to allow listing queue managers