QPID-1135: Fix multi-frame message handling. This fix is suboptimal since it creates an extra copy, as a result it's slower and less memory efficent. But it is correct.

Qpid.Buffer/SlicedByteBuffer.cs: Make sure that we only request the right part of the array when returning a slice

Qpid.Integration.Tests/Qpid.Integration.Tests.csproj: add ByteMessageTest

Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs: Add ConsumeNMessage and ConsumeMessages variants that take byte[] instead of string

Qpid.Integration.Tests/testcases/ByteMessageTest.cs: add tests for various sizes of messages


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1.x@667097 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 176 insertions, 1 deletions

diff --git a/dotnet/Qpid.Buffer/SlicedByteBuffer.cs b/dotnet/Qpid.Buffer/SlicedByteBuffer.cs
index c27b7949b6..aba3390dbd 100644
--- a/dotnet/Qpid.Buffer/SlicedByteBuffer.cs
+++ b/dotnet/Qpid.Buffer/SlicedByteBuffer.cs
@@ -27,6 +27,7 @@ namespace Apache.Qpid.Buffer
       private ByteBuffer _buffer;

       private int _capacity;

       private int _startPos;

+      private byte[] _array = null;

 

       public override int Capacity

       {

@@ -35,7 +36,17 @@ namespace Apache.Qpid.Buffer
 

       public override byte[] Array

       {

-         get { return _buffer.Array; }

+      	get 

+      	{

+      		if (_array == null)

+      		{

+      			// FIXME: this creates an unnecessary copy of the data, 

+      			// which is expensive and gross. A sliced clone() would be better

+      			_array = new byte[Limit];

+      			_buffer.GetBytes(_startPos, _array, 0, Limit);

+      		}

+      		return _array;

+      	}

       }

 

       /// <summary>

diff --git a/dotnet/Qpid.Integration.Tests/Qpid.Integration.Tests.csproj b/dotnet/Qpid.Integration.Tests/Qpid.Integration.Tests.csproj
index 51d91f2bd3..d271b05526 100644
--- a/dotnet/Qpid.Integration.Tests/Qpid.Integration.Tests.csproj
+++ b/dotnet/Qpid.Integration.Tests/Qpid.Integration.Tests.csproj
@@ -40,6 +40,7 @@
     <Compile Include="interop\TestCases\TestCase1DummyRun.cs" />

     <Compile Include="interop\TestCases\TestCase3BasicPubSub.cs" />

     <Compile Include="testcases\BaseMessagingTestFixture.cs" />

+    <Compile Include="testcases\ByteMessageTest.cs" />

     <Compile Include="testcases\ChannelQueueTest.cs" />

     <Compile Include="testcases\CommitRollbackTest.cs" />

     <Compile Include="testcases\ConnectionTest.cs" />

diff --git a/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs b/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
index d4b61a2788..7859848233 100644
--- a/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
+++ b/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
@@ -230,6 +230,46 @@ namespace Apache.Qpid.Integration.Tests.testcases
             }

         }

 

+        /// <summary>

+        /// Consumes n messages, checking that the n+1th is not available within a timeout, and that the consumed messages

+        /// are text messages with contents equal to the specified message body.

+        /// </summary>

+        ///

+        /// <param name="n">The number of messages to consume.</param>

+        /// <param name="body">The body text to match against all messages.</param>

+        /// <param name="consumer">The message consumer to recieve the messages on.</param>

+        public static void ConsumeNMessagesOnly(int n, byte[] body, IMessageConsumer consumer)

+        {

+            ConsumeNMessages(n, body, consumer);

+            

+            // Check that one more than n cannot be received.

+            IMessage msg = consumer.Receive(RECEIVE_WAIT);

+            Assert.IsNull(msg, "Consumer got more messages than the number requested (" + n + ").");

+        }

+

+        /// <summary>

+        /// Consumes n messages, checking that the n+1th is not available within a timeout, and that the consumed messages

+        /// are text messages with contents equal to the specified message body.

+        /// </summary>

+        ///

+        /// <param name="n">The number of messages to consume.</param>

+        /// <param name="body">The body text to match against all messages.</param>

+        /// <param name="consumer">The message consumer to recieve the messages on.</param>

+        public static void ConsumeNMessages(int n, byte[] body, IMessageConsumer consumer)

+        {

+            IMessage msg;

+            

+            // Try to receive n messages.

+            for (int i = 0; i < n; i++)

+            {

+            	msg = consumer.Receive(RECEIVE_WAIT);

+            	byte[] msgbody = new byte[((IBytesMessage)msg).BodyLength];

+            	((IBytesMessage)msg).ReadBytes(msgbody);

+                Assert.IsNotNull(msg, "Consumer did not receive message number: " + i);

+                Assert.AreEqual(body, msgbody, "Incorrect Message received on consumer.");

+            }

+        }

+        

         /// <summary>Creates the requested number of bytes of dummy text. Usually used for filling test messages. </summary>

         ///

         /// <param name="size">The number of bytes of dummy text to generate.</param>

diff --git a/dotnet/Qpid.Integration.Tests/testcases/ByteMessageTest.cs b/dotnet/Qpid.Integration.Tests/testcases/ByteMessageTest.cs
new file mode 100755
index 0000000000..e31ee93f70
--- /dev/null
+++ b/dotnet/Qpid.Integration.Tests/testcases/ByteMessageTest.cs
@@ -0,0 +1,123 @@
+/*

+ *

+ * Licensed to the Apache Software Foundation (ASF) under one

+ * or more contributor license agreements.  See the NOTICE file

+ * distributed with this work for additional information

+ * regarding copyright ownership.  The ASF licenses this file

+ * to you under the Apache License, Version 2.0 (the

+ * "License"); you may not use this file except in compliance

+ * with the License.  You may obtain a copy of the License at

+ * 

+ *   http://www.apache.org/licenses/LICENSE-2.0

+ * 

+ * Unless required by applicable law or agreed to in writing,

+ * software distributed under the License is distributed on an

+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

+ * KIND, either express or implied.  See the License for the

+ * specific language governing permissions and limitations

+ * under the License.

+ *

+ */

+

+using System;

+using System.Threading;

+using log4net;

+using NUnit.Framework;

+using Apache.Qpid.Messaging;

+using Apache.Qpid.Client.Qms;

+using Apache.Qpid.Client;

+

+namespace Apache.Qpid.Integration.Tests.testcases

+{

+	/// <summary>

+	/// Checks that byte messages can be produced and received properly.

+	/// </summary>

+	[TestFixture, Category("Integration")]

+	public class ByteMessageTest : BaseMessagingTestFixture

+		

+	{

+		private static ILog log = LogManager.GetLogger(typeof(ByteMessageTest));

+		private static string TEST_ROUTING_KEY = "BYTE_MESSAGE_TEST_QUEUE";

+		

+		[SetUp]

+        public override void Init()

+        {

+            base.Init();

+

+            // Create one producer and one consumer, p2p, tx, consumer with queue bound to producers routing key.

+            SetUpEndPoint(0, true, false, TEST_ROUTING_KEY + testId, AcknowledgeMode.AutoAcknowledge, false, ExchangeNameDefaults.DIRECT, 

+                          true, false, null);

+            SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId, AcknowledgeMode.AutoAcknowledge, false, ExchangeNameDefaults.DIRECT, 

+                          true, false, null);

+        }

+		

+		[TearDown]

+        public override void Shutdown()

+        {

+            try

+            {

+                // Clean up after the test.

+                CloseEndPoint(0);

+                CloseEndPoint(1);

+            } 

+            finally 

+            {

+                base.Shutdown();

+            }

+        }

+

+        /// <summary> Send the byte message and get it back. </summary>

+        private void TestAnyByteMessage(int size)

+        {

+        	byte[] content = new byte[size];

+        	

+        	for (int i = 0; i < content.Length; i++)

+        	{

+        		content[i] = (byte)i;

+        	}

+            // Send messages.

+            IBytesMessage msg = testChannel[0].CreateBytesMessage();

+            msg.WriteBytes(content);

+            testProducer[0].Send(msg);

+          

+            // Try to receive messages.

+            ConsumeNMessagesOnly(1, content, testConsumer[1]);

+        }

+		

+        [Test]

+        /// <summary> Send a small byte message and get it back. </summary>

+        public void TestSmallByteMessage()

+        {

+        	TestAnyByteMessage(4);

+        }

+        

+        [Test]

+        /// <summary> Send a byte message just under the frame boundry and get it back. </summary>

+        public void TestByteMessageUnderFrameBoundry()

+        {

+        	TestAnyByteMessage((int) Math.Pow(2,16) - 32);

+        }

+        

+        [Test]

+        /// <summary> Send a byte message on the frame boundry and get it back. </summary>

+        public void TestByteMessageOnFrameBoundry()

+        {

+        	TestAnyByteMessage((int) Math.Pow(2,16) - 2);

+        }

+        

+        [Test]

+        /// <summary> Send a byte message on the frame boundry and get it back. </summary>

+        public void TestByteMessageOverFrameBoundry()

+        {

+        	TestAnyByteMessage((int) Math.Pow(2,16) - 1);

+        }

+        

+        [Test]

+        /// <summary> Send a byte message on the frame boundry and get it back. </summary>

+        public void TestByteMessageWellOverFrameBoundry()

+        {

+        	TestAnyByteMessage((int) Math.Pow(2,17));

+        }

+        

+	}

+}