QPID-136 Ported Prefetch with PrefetchHigh and PrefetchLow

QPID-137 Ported AcknowledgeModes


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@481035 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 48 insertions, 32 deletions

diff --git a/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingConsumer.cs b/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingConsumer.cs
index cf26b42c6a..1b27f920b8 100644
--- a/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingConsumer.cs
+++ b/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingConsumer.cs
@@ -58,8 +58,9 @@ namespace Qpid.Client.Tests
             _channel.Bind(queueName, _serviceName, null, CreatePatternAsFieldTable());
 
             IMessageConsumer consumer = _channel.CreateConsumerBuilder(queueName)
-                .withPrefetch(100)
-                .withNoLocal(true)
+                .WithPrefetchLow(100)
+                .WithPrefetchHigh(500)
+                .WithNoLocal(true)
                 .Create();
 
             consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
diff --git a/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingProducer.cs b/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingProducer.cs
index 35f6017f48..c748ef8840 100644
--- a/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingProducer.cs
+++ b/dotnet/Qpid.Client.Tests/HeadersExchange/HeadersMatchingProducer.cs
@@ -43,8 +43,8 @@ namespace Qpid.Client.Tests
             try
             {
                 _publisher = _channel.CreatePublisherBuilder()
-                    .withExchangeName(_commandExchangeName)
-                    .withMandatory(true)
+                    .WithExchangeName(_commandExchangeName)
+                    .WithMandatory(true)
                     .Create();
 
                 // Disabling timestamps - a performance optimisation where timestamps and TTL/expiration 
diff --git a/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs b/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
index 6b6fca20b2..687f08eeef 100644
--- a/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
+++ b/dotnet/Qpid.Client.Tests/MultiConsumer/ProducerMultiConsumer.cs
@@ -70,8 +70,8 @@ namespace Qpid.Client.Tests
         {
             base.Init();
             _publisher = _channel.CreatePublisherBuilder()
-                .withRoutingKey(_commandQueueName)
-                .withExchangeName(ExchangeNameDefaults.TOPIC)
+                .WithRoutingKey(_commandQueueName)
+                .WithExchangeName(ExchangeNameDefaults.TOPIC)
                 .Create();
 
             _publisher.DisableMessageTimestamp = true;
@@ -85,7 +85,7 @@ namespace Qpid.Client.Tests
                 _channel.Bind(queueName, ExchangeNameDefaults.TOPIC, _commandQueueName);
 
                 _consumers[i] = _channel.CreateConsumerBuilder(queueName)
-                    .withPrefetch(100).Create();
+                    .WithPrefetchLow(100).Create();
                 _consumers[i].OnMessage = new MessageReceivedDelegate(OnMessage);
             }
             _connection.Start();
diff --git a/dotnet/Qpid.Client.Tests/failover/FailoverTest.cs b/dotnet/Qpid.Client.Tests/failover/FailoverTest.cs
index 952cc2a58e..52ef76c559 100644
--- a/dotnet/Qpid.Client.Tests/failover/FailoverTest.cs
+++ b/dotnet/Qpid.Client.Tests/failover/FailoverTest.cs
@@ -78,8 +78,8 @@ namespace Qpid.Client.Tests.failover
 
 //            _publisher = _channel.CreatePublisher(exchangeName, exchangeClass, routingKey);
             _publisher = _channel.CreatePublisherBuilder()
-                .withRoutingKey(routingKey)
-                .withExchangeName(exchangeName)
+                .WithRoutingKey(routingKey)
+                .WithExchangeName(exchangeName)
                 .Create();
             _publisher.Send(msg);
         }
@@ -206,8 +206,8 @@ namespace Qpid.Client.Tests.failover
 
                 //return _channel.CreatePublisher(exchangeName, exchangeClass, routingKey);
                 return _session.CreatePublisherBuilder()
-                    .withExchangeName(exchangeName)
-                    .withRoutingKey(routingKey)
+                    .WithExchangeName(exchangeName)
+                    .WithRoutingKey(routingKey)
                     .Create();
             }
         }
diff --git a/dotnet/Qpid.Client.Tests/failover/FailoverTxTest.cs b/dotnet/Qpid.Client.Tests/failover/FailoverTxTest.cs
index 79a04e79eb..4e95c12290 100644
--- a/dotnet/Qpid.Client.Tests/failover/FailoverTxTest.cs
+++ b/dotnet/Qpid.Client.Tests/failover/FailoverTxTest.cs
@@ -36,8 +36,11 @@ namespace Qpid.Client.Tests.failover
         const int NUM_ITERATIONS = 10;
         const int NUM_COMMITED_MESSAGES = 10;
         const int NUM_ROLLEDBACK_MESSAGES = 3;
-        const int SLEEP_MILLIS = 500;
+        const int SLEEP_MILLIS = 50;
 
+        // AutoAcknowledge, ClientAcknowledge, DupsOkAcknowledge, NoAcknowledge, PreAcknowledge
+        AcknowledgeMode _acknowledgeMode = AcknowledgeMode.DupsOkAcknowledge;
+        const bool _noWait = true; // use Receive or ReceiveNoWait
         AMQConnection _connection;
 
         public void OnMessage(IMessage message)
@@ -45,6 +48,11 @@ namespace Qpid.Client.Tests.failover
             try
             {
                 _log.Info("Received: " + ((ITextMessage) message).Text);
+                if (_acknowledgeMode == AcknowledgeMode.ClientAcknowledge)
+                {
+                    _log.Info("client acknowledging");
+                    message.Acknowledge();
+                }
             }
             catch (QpidException e)
             {
@@ -56,11 +64,13 @@ namespace Qpid.Client.Tests.failover
         {
             FailoverTxTest _failoverTxTest;
             IMessageConsumer _consumer;
+            private bool _noWait;
 
-            internal NoWaitConsumer(FailoverTxTest failoverTxTest, IMessageConsumer channel)
+            internal NoWaitConsumer(FailoverTxTest failoverTxTest, IMessageConsumer channel, bool noWait)
             {
                 _failoverTxTest = failoverTxTest;
                 _consumer = channel;
+                _noWait = noWait;
             }
 
             internal void Run()
@@ -68,7 +78,9 @@ namespace Qpid.Client.Tests.failover
                 int messages = 0;
                 while (messages < NUM_COMMITED_MESSAGES)
                 {
-                    IMessage msg = _consumer.ReceiveNoWait();
+                    IMessage msg;
+                    if (_noWait) msg = _consumer.ReceiveNoWait();
+                    else msg = _consumer.Receive();
                     if (msg != null)
                     {
                         _log.Info("NoWait received message");
@@ -93,7 +105,8 @@ namespace Qpid.Client.Tests.failover
             _log.Info("connectionInfo = " + connectionInfo);
             _log.Info("connection.asUrl = " + _connection.toURL());
 
-            IChannel receivingChannel = _connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge);
+            _log.Info("AcknowledgeMode is " + _acknowledgeMode);
+            IChannel receivingChannel = _connection.CreateChannel(false, _acknowledgeMode);
 
             string queueName = receivingChannel.GenerateUniqueName();
 
@@ -103,17 +116,17 @@ namespace Qpid.Client.Tests.failover
             // No need to call Queue.Bind as automatically bound to default direct exchange.
             receivingChannel.Bind(queueName, "amq.direct", queueName);
 
-
-            IMessageConsumer consumer = receivingChannel.CreateConsumerBuilder(queueName).Create();
-            bool useThread = true;
+            IMessageConsumer consumer = receivingChannel.CreateConsumerBuilder(queueName)
+                .WithPrefetchLow(30)
+                .WithPrefetchHigh(60).Create();
+            bool useThread = false;
             if (useThread)
             {
-                NoWaitConsumer noWaitConsumer = new NoWaitConsumer(this, consumer);
+                NoWaitConsumer noWaitConsumer = new NoWaitConsumer(this, consumer, _noWait);
                 new Thread(noWaitConsumer.Run).Start();
             }
             else
             {
-                //receivingChannel.CreateConsumerBuilder(queueName).Create().OnMessage = new MessageReceivedDelegate(onMessage);
                 consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
             }
 
@@ -133,7 +146,7 @@ namespace Qpid.Client.Tests.failover
             bool transacted = true;
             IChannel publishingChannel = _connection.CreateChannel(transacted, AcknowledgeMode.NoAcknowledge);
             IMessagePublisher publisher = publishingChannel.CreatePublisherBuilder()
-                .withRoutingKey(routingKey)
+                .WithRoutingKey(routingKey)
                 .Create();
 
             for (int i = 1; i <= NUM_ITERATIONS; ++i)
diff --git a/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs b/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
index bb5758c18c..a52b4e2c50 100644
--- a/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
+++ b/dotnet/Qpid.Client.Tests/requestreply1/ServiceProvidingClient.cs
@@ -60,8 +60,9 @@ namespace Qpid.Client.Tests
             _channel.DeclareQueue(_serviceName, false, false, false);
 
             IMessageConsumer consumer = _channel.CreateConsumerBuilder(_serviceName)
-                .withPrefetch(100)
-                .withNoLocal(true)
+                .WithPrefetchLow(100)
+                .WithPrefetchHigh(500)
+                .WithNoLocal(true)
                 .Create();
             consumer.OnMessage = new MessageReceivedDelegate(OnMessage);
         }
@@ -100,8 +101,8 @@ namespace Qpid.Client.Tests
 //                Console.WriteLine("ReplyTo.RoutingKey = " + _replyToRoutingKey);
 
                 _destinationPublisher = _channel.CreatePublisherBuilder()
-                    .withExchangeName(_replyToExchangeName)
-                    .withRoutingKey(_replyToRoutingKey)
+                    .WithExchangeName(_replyToExchangeName)
+                    .WithRoutingKey(_replyToRoutingKey)
                     .Create();
                 _destinationPublisher.DisableMessageTimestamp = true;
                 _destinationPublisher.DeliveryMode = DeliveryMode.NonPersistent;
diff --git a/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
index 582f022719..e437b670bf 100644
--- a/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
+++ b/dotnet/Qpid.Client.Tests/requestreply1/ServiceRequestingClient.cs
@@ -53,7 +53,7 @@ namespace Qpid.Client.Tests
             try
             {
                 _publisher = _channel.CreatePublisherBuilder()
-                    .withRoutingKey(_commandQueueName)
+                    .WithRoutingKey(_commandQueueName)
                     .Create();
                 _publisher.DisableMessageTimestamp = true; // XXX: need a "with" for this in builder?
                 _publisher.DeliveryMode = DeliveryMode.NonPersistent;  // XXX: need a "with" for this in builder?
@@ -74,9 +74,10 @@ namespace Qpid.Client.Tests
             _channel.DeclareQueue(replyQueueName, false, true, true);
 
             IMessageConsumer messageConsumer = _channel.CreateConsumerBuilder(replyQueueName)
-                .withPrefetch(100)
-                .withNoLocal(true)
-                .withExclusive(true).Create();
+                .WithPrefetchLow(100)
+                .WithPrefetchHigh(200)
+                .WithNoLocal(true)
+                .WithExclusive(true).Create();
  
             _startTime = DateTime.Now.Ticks;
 
diff --git a/dotnet/Qpid.Client.Tests/undeliverable/UndeliverableTest.cs b/dotnet/Qpid.Client.Tests/undeliverable/UndeliverableTest.cs
index 63c936d667..84ae2c92c1 100644
--- a/dotnet/Qpid.Client.Tests/undeliverable/UndeliverableTest.cs
+++ b/dotnet/Qpid.Client.Tests/undeliverable/UndeliverableTest.cs
@@ -81,11 +81,11 @@ namespace Qpid.Client.Tests
 
             // Send a test message to a non-existant queue on the default exchange. See if message is returned!
             MessagePublisherBuilder builder = _channel.CreatePublisherBuilder()
-                .withRoutingKey("Non-existant route key!")
-                .withMandatory(true);
+                .WithRoutingKey("Non-existant route key!")
+                .WithMandatory(true);
             if (exchangeName != null)
             {
-                builder.withExchangeName(exchangeName);
+                builder.WithExchangeName(exchangeName);
             }
             IMessagePublisher publisher = builder.Create();
             publisher.Send(_channel.CreateTextMessage("Hiya!"));