You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2010/01/24 00:17:59 UTC
svn commit: r902505 - in /qpid/trunk/qpid/dotnet: Qpid.Client/Client/
Qpid.Integration.Tests/testcases/ Qpid.Messaging/
Author: rgodfrey
Date: Sat Jan 23 23:17:58 2010
New Revision: 902505
URL: http://svn.apache.org/viewvc?rev=902505&view=rev
Log:
QPID-2321 : Add queue browsing capability to 0-8 .net client to allow for use of LVQ
Added:
qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/QueueBrowsingTest.cs
Modified:
qpid/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs
qpid/trunk/qpid/dotnet/Qpid.Client/Client/BasicMessageConsumer.cs
qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/Qpid.Integration.Tests.csproj
qpid/trunk/qpid/dotnet/Qpid.Messaging/IChannel.cs
qpid/trunk/qpid/dotnet/Qpid.Messaging/MessageConsumerBuilder.cs
Modified: qpid/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Client/Client/AmqChannel.cs Sat Jan 23 23:17:58 2010
@@ -251,7 +251,20 @@
/// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
public void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete)
{
- DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete);
+ DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete, null);
+ }
+
+ /// <summary>
+ /// Declare a new queue with the specified set of arguments.
+ /// </summary>
+ /// <param name="queueName">Name of the queue</param>
+ /// <param name="isDurable">True if the queue should be durable</param>
+ /// <param name="isExclusive">True if the queue should be exclusive to this channel</param>
+ /// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
+ /// <param name="args">Optional arguments to Queue.Declare</param>
+ public void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete, IFieldTable args)
+ {
+ DoQueueDeclare(queueName, isDurable, isExclusive, isAutoDelete, args);
}
/// <summary>
@@ -386,9 +399,33 @@
_logger.Debug(String.Format("CreateConsumer queueName={0} prefetchLow={1} prefetchHigh={2} noLocal={3} exclusive={4} ",
queueName, prefetchLow, prefetchHigh, noLocal, exclusive));
- return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive);
+ return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive, false);
+ }
+
+ /// <summary>
+ /// Creates a new consumer.
+ /// </summary>
+ /// <param name="queueName">Name of queue to receive messages from</param>
+ /// <param name="prefetchLow">Low prefetch value</param>
+ /// <param name="prefetchHigh">High prefetch value</param>
+ /// <param name="noLocal">If true, messages sent on this channel will not be received by this consumer</param>
+ /// <param name="exclusive">If true, the consumer opens the queue in exclusive mode</param>
+ /// <param name="browse">If true, the consumer only browses and does not consume messages</param>
+ /// <returns>The new consumer</returns>
+ public IMessageConsumer CreateConsumer(string queueName,
+ int prefetchLow,
+ int prefetchHigh,
+ bool noLocal,
+ bool exclusive,
+ bool browse)
+ {
+ _logger.Debug(String.Format("CreateConsumer queueName={0} prefetchLow={1} prefetchHigh={2} noLocal={3} exclusive={4} browse={5}",
+ queueName, prefetchLow, prefetchHigh, noLocal, exclusive, browse));
+
+ return CreateConsumerImpl(queueName, prefetchLow, prefetchHigh, noLocal, exclusive, browse);
}
+
/// <summary>
/// Unsubscribe from a queue.
/// </summary>
@@ -712,7 +749,8 @@
int prefetchLow,
int prefetchHigh,
bool noLocal,
- bool exclusive)
+ bool exclusive,
+ bool browse)
{
lock (_closingLock)
{
@@ -720,7 +758,8 @@
BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, queueName, noLocal,
_messageFactoryRegistry, this,
- prefetchHigh, prefetchLow, exclusive);
+ prefetchHigh, prefetchLow, exclusive,
+ browse);
try
{
RegisterConsumer(consumer);
@@ -894,7 +933,7 @@
_consumers.Add(tag, consumer);
String consumerTag = ConsumeFromQueue(consumer.QueueName, consumer.NoLocal,
- consumer.Exclusive, consumer.AcknowledgeMode, tag);
+ consumer.Exclusive, consumer.AcknowledgeMode, tag, consumer.Browse);
}
@@ -919,13 +958,17 @@
routingKey, true, args));
}
- private String ConsumeFromQueue(String queueName, bool noLocal, bool exclusive, AcknowledgeMode acknowledgeMode, String tag)
+ private String ConsumeFromQueue(String queueName, bool noLocal, bool exclusive, AcknowledgeMode acknowledgeMode, String tag, bool browse)
{
-
+ FieldTable args = new FieldTable();
+ if(browse)
+ {
+ args["x-filter-no-consume"] = true;
+ }
AMQFrame basicConsume = BasicConsumeBody.CreateAMQFrame(_channelId, 0,
queueName, tag, noLocal,
acknowledgeMode == AcknowledgeMode.NoAcknowledge,
- exclusive, true, new FieldTable());
+ exclusive, true, args);
_replayFrames.Add(basicConsume);
@@ -958,13 +1001,13 @@
}
}
- private void DoQueueDeclare(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete)
+ private void DoQueueDeclare(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete, IFieldTable args)
{
_logger.Debug(string.Format("DeclareQueue name={0} durable={1} exclusive={2}, auto-delete={3}",
queueName, isDurable, isExclusive, isAutoDelete));
AMQFrame queueDeclare = QueueDeclareBody.CreateAMQFrame(_channelId, 0, queueName, false, isDurable, isExclusive,
- isAutoDelete, false, null);
+ isAutoDelete, false, (FieldTable) args);
lock (_connection.FailoverMutex)
Modified: qpid/trunk/qpid/dotnet/Qpid.Client/Client/BasicMessageConsumer.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Client/Client/BasicMessageConsumer.cs?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Client/Client/BasicMessageConsumer.cs (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Client/Client/BasicMessageConsumer.cs Sat Jan 23 23:17:58 2010
@@ -44,6 +44,13 @@
get { return _exclusive; }
}
+ private bool _browse;
+
+ public bool Browse
+ {
+ get { return _browse; }
+ }
+
public bool NoLocal
{
get { return _noLocal; }
@@ -131,7 +138,7 @@
internal BasicMessageConsumer(ushort channelId, string queueName, bool noLocal,
MessageFactoryRegistry messageFactory, AmqChannel channel,
- int prefetchHigh, int prefetchLow, bool exclusive)
+ int prefetchHigh, int prefetchLow, bool exclusive, bool browse)
{
_channelId = channelId;
_queueName = queueName;
@@ -142,6 +149,7 @@
_prefetchHigh = prefetchHigh;
_prefetchLow = prefetchLow;
_exclusive = exclusive;
+ _browse = browse;
if (_acknowledgeMode == AcknowledgeMode.SessionTransacted)
{
Modified: qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/BaseMessagingTestFixture.cs Sat Jan 23 23:17:58 2010
@@ -89,7 +89,7 @@
{
log.Debug("public virtual void Shutdown(): called");
}
-
+
/// <summary> Sets up the nth test end-point. </summary>
///
/// <param name="n">The index of the test end-point to set up.</param>
@@ -105,6 +105,25 @@
public void SetUpEndPoint(int n, bool producer, bool consumer, string routingKey, AcknowledgeMode ackMode, bool transacted,
string exchangeName, bool declareBind, bool durable, string subscriptionName)
{
+ SetUpEndPoint(n, producer, consumer, routingKey, ackMode, transacted, exchangeName, declareBind, durable, subscriptionName, true, false);
+ }
+ /// <summary> Sets up the nth test end-point. </summary>
+ ///
+ /// <param name="n">The index of the test end-point to set up.</param>
+ /// <param name="producer"><tt>true</tt> to set up a producer on the end-point.</param>
+ /// <param name="consumer"><tt>true</tt> to set up a consumer on the end-point.</param>
+ /// <param name="routingKey">The routing key for the producer to send on.</param>
+ /// <param name="ackMode">The ack mode for the end-points channel.</param>
+ /// <param name="transacted"><tt>true</tt> to use transactions on the end-points channel.</param>
+ /// <param name="exchangeName">The exchange to produce or consume on.</param>
+ /// <param name="declareBind"><tt>true</tt> if the consumers queue should be declared and bound, <tt>false</tt> if it has already been.</param>
+ /// <param name="durable"><tt>true</tt> to declare the consumers queue as durable.</param>
+ /// <param name="subscriptionName">If durable is true, the fixed unique queue name to use.</param>
+ /// <param name="exclusive"><tt>true</tt> declare queue as exclusive.</param>
+ /// <param name="browse"><tt>true</tt> only browse, don''t consume.</param>
+ public void SetUpEndPoint(int n, bool producer, bool consumer, string routingKey, AcknowledgeMode ackMode, bool transacted,
+ string exchangeName, bool declareBind, bool durable, string subscriptionName, bool exclusive, bool browse)
+ {
// Allow client id to be fixed, or undefined.
{
// Use unique id for end point.
@@ -137,7 +156,7 @@
if (declareBind)
{
- testChannel[n].DeclareQueue(queueName, durable, true, false);
+ testChannel[n].DeclareQueue(queueName, durable, exclusive, false);
testChannel[n].Bind(queueName, exchangeName, routingKey);
}
}
@@ -156,7 +175,7 @@
}
}
- testConsumer[n] = testChannel[n].CreateConsumerBuilder(queueName).Create();
+ testConsumer[n] = testChannel[n].CreateConsumerBuilder(queueName).WithBrowse(browse).Create();
}
}
Modified: qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/Qpid.Integration.Tests.csproj
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/Qpid.Integration.Tests.csproj?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/Qpid.Integration.Tests.csproj (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/Qpid.Integration.Tests.csproj Sat Jan 23 23:17:58 2010
@@ -59,5 +59,6 @@
<Compile Include="MandatoryMessageTest.cs" />
<Compile Include="ProducerMultiConsumerTest.cs" />
<Compile Include="SslConnectionTest.cs" />
+ <Compile Include="QueueBrowsingTest.cs" />
</ItemGroup>
</Project>
Added: qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/QueueBrowsingTest.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/QueueBrowsingTest.cs?rev=902505&view=auto
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/QueueBrowsingTest.cs (added)
+++ qpid/trunk/qpid/dotnet/Qpid.Integration.Tests/testcases/QueueBrowsingTest.cs Sat Jan 23 23:17:58 2010
@@ -0,0 +1,121 @@
+/*
+ *
+ * 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.Framing;
+
+namespace Apache.Qpid.Integration.Tests.testcases
+{
+ [TestFixture, Category("Integration")]
+ public class QueueBrowsingTest : BaseMessagingTestFixture
+ {
+ /// <summary>Used for debugging purposes.</summary>
+ private static ILog log = LogManager.GetLogger(typeof(QueueBrowsingTest));
+
+ public const string TEST_ROUTING_KEY = "queuebrowsingkey";
+ public const string TEST_ROUTING_KEY2 = "lvqbrowsingkey";
+
+
+ [SetUp]
+ public override void Init()
+ {
+ base.Init();
+ }
+
+ [TearDown]
+ public override void Shutdown()
+ {
+ base.Shutdown();
+ }
+
+ [Test]
+ public void TestQueueBrowsing()
+ {
+ // Create a topic with one producer and two consumers.
+ SetUpEndPoint(0, true, false, TEST_ROUTING_KEY + testId, AcknowledgeMode.AutoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, null, false, false);
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY + testId, AcknowledgeMode.NoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, TEST_ROUTING_KEY + testId, false, true);
+ SetUpEndPoint(2, false, true, TEST_ROUTING_KEY + testId, AcknowledgeMode.NoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, TEST_ROUTING_KEY + testId, false, true);
+
+ Thread.Sleep(500);
+
+ // Send messages and receive on both consumers.
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+ testProducer[0].Send(testChannel[0].CreateTextMessage("msg"));
+
+ Thread.Sleep(2000);
+
+
+ ConsumeNMessagesOnly(6, "msg", testConsumer[1]);
+ ConsumeNMessagesOnly(6, "msg", testConsumer[2]);
+
+ // Clean up any open consumers at the end of the test.
+ CloseEndPoint(2);
+ CloseEndPoint(1);
+ CloseEndPoint(0);
+ }
+
+ [Test]
+ public void TestQueueBrowsingLVQ()
+ {
+ // Create a topic with one producer and two consumers.
+ SetUpEndPoint(0, true, false, TEST_ROUTING_KEY2 + testId, AcknowledgeMode.AutoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, TEST_ROUTING_KEY2 + testId, false, false);
+ FieldTable args = new FieldTable();
+ args.SetBoolean("qpid.last_value_queue", true);
+ args.SetString("qpid.last_value_queue_key", "key");
+ testChannel[0].DeclareQueue(TEST_ROUTING_KEY2 + testId, true, false, false, args);
+ testChannel[0].Bind(TEST_ROUTING_KEY2 + testId, ExchangeNameDefaults.DIRECT, TEST_ROUTING_KEY2 + testId);
+ Thread.Sleep(500);
+
+
+ for (int i = 0; i < 12; i++)
+ {
+ ITextMessage msg = testChannel[0].CreateTextMessage("msg");
+ msg.Headers.SetInt("key", i%6);
+ testProducer[0].Send(msg);
+ }
+
+ Thread.Sleep(2000);
+
+ SetUpEndPoint(1, false, true, TEST_ROUTING_KEY2 + testId, AcknowledgeMode.NoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, TEST_ROUTING_KEY2 + testId, false, true);
+ SetUpEndPoint(2, false, true, TEST_ROUTING_KEY2 + testId, AcknowledgeMode.NoAcknowledge, false, ExchangeNameDefaults.DIRECT, true, true, TEST_ROUTING_KEY2 + testId, false, true);
+
+ Thread.Sleep(500);
+
+
+ ConsumeNMessagesOnly(6, "msg", testConsumer[1]);
+ ConsumeNMessagesOnly(6, "msg", testConsumer[2]);
+
+ // Clean up any open consumers at the end of the test.
+ CloseEndPoint(2);
+ CloseEndPoint(1);
+ CloseEndPoint(0);
+ }
+
+ }
+}
Modified: qpid/trunk/qpid/dotnet/Qpid.Messaging/IChannel.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Messaging/IChannel.cs?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Messaging/IChannel.cs (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Messaging/IChannel.cs Sat Jan 23 23:17:58 2010
@@ -97,6 +97,17 @@
/// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete);
+
+ /// <summary>
+ /// Declare a new queue with the specified set of arguments.
+ /// </summary>
+ /// <param name="queueName">Name of the queue</param>
+ /// <param name="isDurable">True if the queue should be durable</param>
+ /// <param name="isExclusive">True if the queue should be exclusive to this channel</param>
+ /// <param name="isAutoDelete">True if the queue should be deleted when the channel closes</param>
+ /// <param name="args">Optional arguments to Queue.Declare</param>
+ void DeclareQueue(string queueName, bool isDurable, bool isExclusive, bool isAutoDelete, IFieldTable args);
+
/// <summary>
/// Delete a queue with the specifies arguments.
/// </summary>
@@ -191,6 +202,23 @@
int prefetchHigh,
bool noLocal,
bool exclusive);
+
+ /// <summary>
+ /// Creates a new consumer.
+ /// </summary>
+ /// <param name="queueName">Name of queue to receive messages from</param>
+ /// <param name="prefetchLow">Low prefetch value</param>
+ /// <param name="prefetchHigh">High prefetch value</param>
+ /// <param name="noLocal">If true, messages sent on this channel will not be received by this consumer</param>
+ /// <param name="exclusive">If true, the consumer opens the queue in exclusive mode</param>
+ /// <param name="browse">If true, the consumer only browses and does not consume</param>
+ /// <returns>The new consumer</returns>
+ IMessageConsumer CreateConsumer(string queueName,
+ int prefetchLow,
+ int prefetchHigh,
+ bool noLocal,
+ bool exclusive,
+ bool browse);
/// <summary>
/// Unsubscribe from a queue.
Modified: qpid/trunk/qpid/dotnet/Qpid.Messaging/MessageConsumerBuilder.cs
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/dotnet/Qpid.Messaging/MessageConsumerBuilder.cs?rev=902505&r1=902504&r2=902505&view=diff
==============================================================================
--- qpid/trunk/qpid/dotnet/Qpid.Messaging/MessageConsumerBuilder.cs (original)
+++ qpid/trunk/qpid/dotnet/Qpid.Messaging/MessageConsumerBuilder.cs Sat Jan 23 23:17:58 2010
@@ -38,6 +38,8 @@
private bool _exclusive = false;
+ private bool _browse = false;
+
//private bool _durable = false;
//private string _subscriptionName = null;
@@ -81,6 +83,12 @@
return this;
}
+ public MessageConsumerBuilder WithBrowse(bool browse)
+ {
+ _browse = browse;
+ return this;
+ }
+
/*
public MessageConsumerBuilder WithDurable(bool durable)
{
@@ -99,7 +107,7 @@
public IMessageConsumer Create()
{
- return _channel.CreateConsumer(_queueName, _prefetchLow, _prefetchHigh, _noLocal, _exclusive);
+ return _channel.CreateConsumer(_queueName, _prefetchLow, _prefetchHigh, _noLocal, _exclusive, _browse);
}
}
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org