Author: rupertlssmith
Date: Mon Jan 21 07:40:48 2008
New Revision: 613917
URL: http://svn.apache.org/viewvc?rev=613917&view=rev
Log:
Qpid-727. Further test rationalization; pushed cut and paste code up into base
case.
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/CommitRollbackTest.cs
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/DurableSubscriptionTest.cs
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/ProducerMultiConsumerTest.cs
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SslConnectionTest.cs
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SyncConsumerTest.cs
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
Mon Jan 21 07:40:48 2008
@@ -19,6 +19,7 @@
*
*/
using System;
+using System.Text;
using log4net;
using NUnit.Framework;
using Apache.Qpid.Messaging;
@@ -36,6 +37,9 @@
{
private static ILog log =
LogManager.GetLogger(typeof(BaseMessagingTestFixture));
+ /// <summary> Used to build dummy data to fill test messages with.
</summary>
+ private const string MESSAGE_DATA_BYTES = "-- Test Message -- Test
Message -- Test Message -- Test Message -- Test Message ";
+
/// <summary> The default AMQ connection URL to use for tests.
</summary>
const string connectionUri = "amqp://guest:[EMAIL
PROTECTED]/test?brokerlist='tcp://localhost:5672'";
@@ -49,16 +53,22 @@
protected IChannel _channel;
/// <summary> Holds an array of connections for building mutiple test
end-points. </summary>
- protected IConnection[] testConnection = new IConnection[2];
+ protected IConnection[] testConnection = new IConnection[10];
/// <summary> Holds an array of channels for building mutiple test
end-points. </summary>
- protected IChannel[] testChannel = new IChannel[2];
+ protected IChannel[] testChannel = new IChannel[10];
/// <summary> Holds an array of producers for building mutiple test
end-points. </summary>
- protected IMessagePublisher[] testProducer = new IMessagePublisher[2];
+ protected IMessagePublisher[] testProducer = new IMessagePublisher[10];
/// <summary> Holds an array of consumers for building mutiple test
end-points. </summary>
- protected IMessageConsumer[] testConsumer = new IMessageConsumer[2];
+ protected IMessageConsumer[] testConsumer = new IMessageConsumer[10];
+
+ /// <summary> A counter used to supply unique ids. </summary>
+ private int uniqueId = 0;
+
+ /// <summary> Used to hold unique ids per test. </summary>
+ protected int testId;
/// <summary>
/// Creates the test connection and channel.
@@ -68,6 +78,9 @@
{
log.Debug("public virtual void Init(): called");
+ // Set up a unique id for this test.
+ testId = uniqueId++;
+
_connection = new AMQConnection(connectionInfo);
_channel = _connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 500, 300);
}
@@ -84,23 +97,24 @@
if (_connection != null)
{
log.Debug("Disposing connection.");
- //_connection.Close();
+ _connection.Close();
_connection.Dispose();
log.Debug("Connection disposed.");
}
}
/// <summary> Sets up the nth test end-point. </summary>
- public void SetUpEndPoint(int n, bool producer, bool consumer, string
routingKey)
+ public void SetUpEndPoint(int n, bool producer, bool consumer, string
routingKey, AcknowledgeMode ackMode, bool transacted,
+ string exchangeName, bool durable, string
subscriptionName)
{
testConnection[n] = new AMQConnection(connectionInfo);
testConnection[n].Start();
- testChannel[n] = testConnection[n].CreateChannel(true,
AcknowledgeMode.AutoAcknowledge, 1);
+ testChannel[n] = testConnection[n].CreateChannel(transacted,
ackMode, 1);
if (producer)
{
testProducer[n] = testChannel[n].CreatePublisherBuilder()
- .WithExchangeName(ExchangeNameDefaults.DIRECT)
+ .WithExchangeName(exchangeName)
.WithRoutingKey(routingKey)
.Create();
}
@@ -110,8 +124,16 @@
string queueName = testChannel[n].GenerateUniqueName();
testChannel[n].DeclareQueue(queueName, false, true, true);
testChannel[n].Bind(queueName, ExchangeNameDefaults.DIRECT,
routingKey);
- testConsumer[n] =
testChannel[n].CreateConsumerBuilder(queueName)
- .Create();
+ MessageConsumerBuilder consumerBuilder =
testChannel[n].CreateConsumerBuilder(queueName);
+
+ if (durable)
+ {
+ consumerBuilder
+ .WithSubscriptionName(subscriptionName)
+ .WithDurable(true);
+ }
+
+ testConsumer[n] = consumerBuilder.Create();
}
}
@@ -177,5 +199,33 @@
Assert.AreEqual(body, ((ITextMessage)msg).Text, "Incorrect
Message recevied on consumer1.");
}
}
+
+ /// <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>
+ ///
+ /// <return>The requested number of bytes of dummy text.</return>
+ public static String GetData(int size)
+ {
+ StringBuilder buf = new StringBuilder(size);
+
+ if (size > 0)
+ {
+ int div = MESSAGE_DATA_BYTES.Length / size;
+ int mod = MESSAGE_DATA_BYTES.Length % size;
+
+ for (int i = 0; i < div; i++)
+ {
+ buf.Append(MESSAGE_DATA_BYTES);
+ }
+
+ if (mod != 0)
+ {
+ buf.Append(MESSAGE_DATA_BYTES, 0, mod);
+ }
+ }
+
+ return buf.ToString();
+ }
}
}
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/CommitRollbackTest.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/CommitRollbackTest.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/CommitRollbackTest.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/CommitRollbackTest.cs
Mon Jan 21 07:40:48 2008
@@ -50,23 +50,14 @@
/// <summary>Defines the name of the test topic to use with the
tests.</summary>
public const string TEST_ROUTING_KEY = "commitrollbacktestkey";
- /// <summary>A counter used to supply unique ids. </summary>
- int uniqueId = 0;
-
- /// <summary>Used to hold unique ids per test. </summary>
- int testId;
-
[SetUp]
public override void Init()
{
base.Init();
- // Set up a unqiue id for this test.
- testId = uniqueId++;
-
// Create one producer and one consumer, p2p, tx, consumer with
queue bound to producers routing key.
- SetUpEndPoint(0, true, false, TEST_ROUTING_KEY + testId);
- SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId);
+ SetUpEndPoint(0, true, false, TEST_ROUTING_KEY + testId,
AcknowledgeMode.AutoAcknowledge, true, ExchangeNameDefaults.DIRECT, false,
null);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId,
AcknowledgeMode.AutoAcknowledge, true, ExchangeNameDefaults.DIRECT, false,
null);
}
[TearDown]
@@ -135,7 +126,7 @@
// Close end-point 1 without committing the message, then re-open
to consume again.
CloseEndPoint(1);
- SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId,
AcknowledgeMode.AutoAcknowledge, true, ExchangeNameDefaults.DIRECT, false,
null);
// Try to receive messages.
ConsumeNMessagesOnly(1, "A", testConsumer[1]);
@@ -155,7 +146,7 @@
// Close end-point 1 without committing the message, then re-open
to consume again.
CloseEndPoint(1);
- SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId,
AcknowledgeMode.AutoAcknowledge, true, ExchangeNameDefaults.DIRECT, false,
null);
// Try to receive messages.
ConsumeNMessagesOnly(0, "A", testConsumer[1]);
@@ -175,7 +166,7 @@
// Close end-point 1 without committing the message, then re-open
to consume again.
CloseEndPoint(1);
- SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId,
AcknowledgeMode.AutoAcknowledge, true, ExchangeNameDefaults.DIRECT, false,
null);
// Try to receive messages.
ConsumeNMessagesOnly(1, "A", testConsumer[1]);
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/DurableSubscriptionTest.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/DurableSubscriptionTest.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/DurableSubscriptionTest.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/DurableSubscriptionTest.cs
Mon Jan 21 07:40:48 2008
@@ -82,54 +82,35 @@
private void TestDurableSubscription(AcknowledgeMode ackMode)
{
// Create a topic with one producer and two consumers.
- IChannel channel0 = _connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 1);
- IMessagePublisher publisher = channel0.CreatePublisherBuilder()
- .WithExchangeName(ExchangeNameDefaults.TOPIC)
- .WithRoutingKey(TEST_ROUTING_KEY)
- .Create();
-
- IChannel channel1 = _connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 1);
- string topicQueueName1 = channel1.GenerateUniqueName();
- channel1.DeclareQueue(topicQueueName1, false, true, true);
- channel1.Bind(topicQueueName1, ExchangeNameDefaults.TOPIC,
TEST_ROUTING_KEY);
- IMessageConsumer consumer1 =
channel1.CreateConsumerBuilder(topicQueueName1)
- .Create();
-
- IChannel channel2 = _connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 1);
- string topicQueueName2 = channel2.GenerateUniqueName();
- channel2.DeclareQueue(topicQueueName2, false, true, true);
- channel2.Bind(topicQueueName2, ExchangeNameDefaults.TOPIC,
TEST_ROUTING_KEY);
- IMessageConsumer consumer2 =
channel2.CreateConsumerBuilder(topicQueueName2)
- .WithSubscriptionName("TestSubscription")
- .WithDurable(true)
- .Create();
+ SetUpEndPoint(0, true, false, TEST_ROUTING_KEY + testId, ackMode,
false, ExchangeNameDefaults.TOPIC, false, null);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId, ackMode,
false, ExchangeNameDefaults.TOPIC, false, null);
+ SetUpEndPoint(2, false, true, TEST_ROUTING_KEY + testId, ackMode,
false, ExchangeNameDefaults.TOPIC,
+ true, "TestSubscription" + testId);
// Send messages and receive on both consumers.
- publisher.Send(channel0.CreateTextMessage("A"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("A"));
- ConsumeNMessagesOnly(1, "A", consumer1);
- ConsumeNMessagesOnly(1, "A", consumer2);
+ ConsumeNMessagesOnly(1, "A", testConsumer[1]);
+ ConsumeNMessagesOnly(1, "A", testConsumer[2]);
// Detach one consumer.
- consumer2.Dispose();
+ CloseEndPoint(2);
// Send message and receive on one consumer.
- publisher.Send(channel0.CreateTextMessage("B"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("B"));
- ConsumeNMessagesOnly(1, "B", consumer1);
+ ConsumeNMessagesOnly(1, "B", testConsumer[1]);
// Re-attach consumer, check that it gets the messages that it
missed.
- IChannel channel3 = _connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 1);
- IMessageConsumer consumer3 =
channel3.CreateConsumerBuilder(topicQueueName2)
- .WithSubscriptionName("TestSubscription")
- .WithDurable(true)
- .Create();
+ SetUpEndPoint(3, false, true, TEST_ROUTING_KEY + testId, ackMode,
false, ExchangeNameDefaults.TOPIC,
+ true, "TestSubscription" + testId);
- ConsumeNMessagesOnly(1, "B", consumer3);
+ ConsumeNMessagesOnly(1, "B", testConsumer[3]);
// Clean up any open consumers at the end of the test.
- consumer1.Dispose();
- consumer3.Dispose();
+ CloseEndPoint(0);
+ CloseEndPoint(1);
+ CloseEndPoint(3);
}
}
}
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/ProducerMultiConsumerTest.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/ProducerMultiConsumerTest.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/ProducerMultiConsumerTest.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/ProducerMultiConsumerTest.cs
Mon Jan 21 07:40:48 2008
@@ -38,35 +38,15 @@
private const int MESSAGE_COUNT = 1000;
- private const string MESSAGE_DATA_BYTES = "****jfd ghljgl hjvhlj
cvhvjf ldhfsj lhfdsjf hldsjfk hdslkfj hsdflk ";
-
AutoResetEvent _finishedEvent = new AutoResetEvent(false);
- private static String GetData(int size)
- {
- StringBuilder buf = new StringBuilder(size);
- int count = 0;
- while (count < size + MESSAGE_DATA_BYTES.Length)
- {
- buf.Append(MESSAGE_DATA_BYTES);
- count += MESSAGE_DATA_BYTES.Length;
- }
- if (count < size)
- {
- buf.Append(MESSAGE_DATA_BYTES, 0, size - count);
- }
-
- return buf.ToString();
- }
-
private IMessagePublisher _publisher;
private IMessageConsumer[] _consumers = new
IMessageConsumer[CONSUMER_COUNT];
private int _messageReceivedCount = 0;
- /*
- [SetUp]
+ //[SetUp]
public override void Init()
{
base.Init();
@@ -91,32 +71,15 @@
}
_connection.Start();
}
- */
- /*
- [TearDown]
+ //[TearDown]
public override void Shutdown()
{
_connection.Stop();
base.Shutdown();
}
- */
- public void OnMessage(IMessage m)
- {
- int newCount = Interlocked.Increment(ref _messageReceivedCount);
- if (newCount % 1000 == 0) _logger.Info("Received count=" +
newCount);
- if (newCount == (MESSAGE_COUNT * CONSUMER_COUNT))
- {
- _logger.Info("All messages received");
- _finishedEvent.Set();
- }
- if ( newCount % 100 == 0 )
- System.Diagnostics.Debug.WriteLine(((ITextMessage)m).Text);
- }
-
- /*
- [Test]
+ //[Test]
public void RunTest()
{
for (int i = 0; i < MESSAGE_COUNT; i++)
@@ -124,7 +87,7 @@
ITextMessage msg;
try
{
- msg = _channel.CreateTextMessage(GetData(512 + 8*i));
+ msg = _channel.CreateTextMessage(GetData(512 + 8*i));
}
catch (Exception e)
{
@@ -135,6 +98,18 @@
}
_finishedEvent.WaitOne();
}
- */
+
+ public void OnMessage(IMessage m)
+ {
+ int newCount = Interlocked.Increment(ref _messageReceivedCount);
+ if (newCount % 1000 == 0) _logger.Info("Received count=" +
newCount);
+ if (newCount == (MESSAGE_COUNT * CONSUMER_COUNT))
+ {
+ _logger.Info("All messages received");
+ _finishedEvent.Set();
+ }
+ if ( newCount % 100 == 0 )
+ System.Diagnostics.Debug.WriteLine(((ITextMessage)m).Text);
+ }
}
}
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SslConnectionTest.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SslConnectionTest.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SslConnectionTest.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SslConnectionTest.cs
Mon Jan 21 07:40:48 2008
@@ -29,36 +29,36 @@
namespace Apache.Qpid.Integration.Tests.testcases
{
- /// <summary>
- /// Test SSL/TLS connections to the broker
- /// </summary>
- [TestFixture, Category("Integration")]
- public class SslConnectionTest
- {
- /// <summary>
- /// Make a test TLS connection to the broker
- /// without using client-certificates
- /// </summary>
- [Test]
- public void DoSslConnection()
- {
- // because for tests we don't usually trust the server certificate
- // we need here to tell the client to ignore certificate validation
errors
- SslOptions sslConfig = new SslOptions(null, true);
-
- MakeBrokerConnection(sslConfig);
- }
-
- private static void MakeBrokerConnection(SslOptions options)
- {
- IConnectionInfo connectionInfo = new QpidConnectionInfo();
- connectionInfo.VirtualHost = "test";
- connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp", "localhost",
8672, options));
-
- using ( IConnection connection = new AMQConnection(connectionInfo) )
- {
- Console.WriteLine("connection = " + connection);
- }
- }
- }
+ /// <summary>
+ /// Test SSL/TLS connections to the broker
+ /// </summary>
+ [TestFixture, Category("Integration")]
+ public class SslConnectionTest
+ {
+ /// <summary>
+ /// Make a test TLS connection to the broker
+ /// without using client-certificates
+ /// </summary>
+ //[Test]
+ public void DoSslConnection()
+ {
+ // because for tests we don't usually trust the server certificate
+ // we need here to tell the client to ignore certificate
validation errors
+ SslOptions sslConfig = new SslOptions(null, true);
+
+ MakeBrokerConnection(sslConfig);
+ }
+
+ private static void MakeBrokerConnection(SslOptions options)
+ {
+ IConnectionInfo connectionInfo = new QpidConnectionInfo();
+ connectionInfo.VirtualHost = "test";
+ connectionInfo.AddBrokerInfo(new AmqBrokerInfo("amqp",
"localhost", 8672, options));
+
+ using ( IConnection connection = new AMQConnection(connectionInfo)
)
+ {
+ Console.WriteLine("connection = " + connection);
+ }
+ }
+ }
}
Modified:
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SyncConsumerTest.cs
URL:
http://svn.apache.org/viewvc/incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SyncConsumerTest.cs?rev=613917&r1=613916&r2=613917&view=diff
==============================================================================
---
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SyncConsumerTest.cs
(original)
+++
incubator/qpid/branches/M2.1/dotnet/Qpid.Integration.Tests/testcases/SyncConsumerTest.cs
Mon Jan 21 07:40:48 2008
@@ -27,105 +27,83 @@
namespace Apache.Qpid.Integration.Tests.testcases
{
- [TestFixture, Category("Integration")]
- public class SyncConsumerTest : BaseMessagingTestFixture
- {
- private static readonly ILog _logger =
LogManager.GetLogger(typeof(SyncConsumerTest));
-
- private string _commandQueueName = "ServiceQ1";
- private const int MESSAGE_COUNT = 1000;
- private const string MESSAGE_DATA_BYTES = "jfd ghljgl hjvhlj cvhvjf
ldhfsj lhfdsjf hldsjfk hdslkfj hsdflk ";
-
- private static String GetData(int size)
- {
- StringBuilder buf = new StringBuilder(size);
- int count = 0;
- while ( count < size + MESSAGE_DATA_BYTES.Length )
- {
- buf.Append(MESSAGE_DATA_BYTES);
- count += MESSAGE_DATA_BYTES.Length;
- }
- if ( count < size )
- {
- buf.Append(MESSAGE_DATA_BYTES, 0, size - count);
- }
-
- return buf.ToString();
- }
-
- private IMessageConsumer _consumer;
- private IMessagePublisher _publisher;
-
- /*
- [SetUp]
- public override void Init()
- {
- base.Init();
- _publisher = _channel.CreatePublisherBuilder()
- .WithRoutingKey(_commandQueueName)
- .WithExchangeName(ExchangeNameDefaults.TOPIC)
- .Create();
-
- _publisher.DisableMessageTimestamp = true;
- _publisher.DeliveryMode = DeliveryMode.NonPersistent;
-
- string queueName = _channel.GenerateUniqueName();
- _channel.DeclareQueue(queueName, false, true, true);
-
- _channel.Bind(queueName, ExchangeNameDefaults.TOPIC,
_commandQueueName);
-
- _consumer = _channel.CreateConsumerBuilder(queueName)
- .WithPrefetchLow(100).Create();
- _connection.Start();
- }
- */
-
- /*
- [Test]
- public void ReceiveWithInfiniteWait()
- {
- // send all messages
- for ( int i = 0; i < MESSAGE_COUNT; i++ )
- {
- ITextMessage msg;
- try
- {
- msg = _channel.CreateTextMessage(GetData(512 + 8 * i));
- } catch ( Exception e )
+ [TestFixture, Category("Integration")]
+ public class SyncConsumerTest : BaseMessagingTestFixture
+ {
+ private static readonly ILog _logger =
LogManager.GetLogger(typeof(SyncConsumerTest));
+
+ private string _commandQueueName = "ServiceQ1";
+ private const int MESSAGE_COUNT = 1000;
+
+ private IMessageConsumer _consumer;
+ private IMessagePublisher _publisher;
+
+ //[SetUp]
+ public override void Init()
+ {
+ base.Init();
+ _publisher = _channel.CreatePublisherBuilder()
+ .WithRoutingKey(_commandQueueName)
+ .WithExchangeName(ExchangeNameDefaults.TOPIC)
+ .Create();
+
+ _publisher.DisableMessageTimestamp = true;
+ _publisher.DeliveryMode = DeliveryMode.NonPersistent;
+
+ string queueName = _channel.GenerateUniqueName();
+ _channel.DeclareQueue(queueName, false, true, true);
+
+ _channel.Bind(queueName, ExchangeNameDefaults.TOPIC,
_commandQueueName);
+
+ _consumer = _channel.CreateConsumerBuilder(queueName)
+ .WithPrefetchLow(100).Create();
+ _connection.Start();
+ }
+
+ //[Test]
+ public void ReceiveWithInfiniteWait()
+ {
+ // send all messages
+ for ( int i = 0; i < MESSAGE_COUNT; i++ )
{
- _logger.Error("Error creating message: " + e, e);
- break;
+ ITextMessage msg;
+ try
+ {
+ msg = _channel.CreateTextMessage(GetData(512 + 8 * i));
+ } catch ( Exception e )
+ {
+ _logger.Error("Error creating message: " + e, e);
+ break;
+ }
+ _publisher.Send(msg);
}
- _publisher.Send(msg);
- }
-
- _logger.Debug("All messages sent");
- // receive all messages
- for ( int i = 0; i < MESSAGE_COUNT; i++ )
- {
- try
+
+ _logger.Debug("All messages sent");
+ // receive all messages
+ for ( int i = 0; i < MESSAGE_COUNT; i++ )
{
- IMessage msg = _consumer.Receive();
- Assert.IsNotNull(msg);
- } catch ( Exception e )
- {
- _logger.Error("Error receiving message: " + e, e);
- Assert.Fail(e.ToString());
+ try
+ {
+ IMessage msg = _consumer.Receive();
+ Assert.IsNotNull(msg);
+ } catch ( Exception e )
+ {
+ _logger.Error("Error receiving message: " + e, e);
+ Assert.Fail(e.ToString());
+ }
}
- }
- }
-
- [Test]
- public void ReceiveWithTimeout()
- {
- ITextMessage msg = _channel.CreateTextMessage(GetData(512 + 8));
- _publisher.Send(msg);
-
- IMessage recvMsg = _consumer.Receive();
- Assert.IsNotNull(recvMsg);
- // empty queue, should timeout
- Assert.IsNull(_consumer.Receive(1000));
- }
- */
- }
+ }
+
+ //[Test]
+ public void ReceiveWithTimeout()
+ {
+ ITextMessage msg = _channel.CreateTextMessage(GetData(512 + 8));
+ _publisher.Send(msg);
+
+ IMessage recvMsg = _consumer.Receive();
+ Assert.IsNotNull(recvMsg);
+ // empty queue, should timeout
+ Assert.IsNull(_consumer.Receive(1000));
+ }
+ }
}
