Hi Rajesh,
I'm not sure that this is possible programatically in the way that you
are suggesting - well short of receiving and discarding which I'm
guessing you'd already thought of :-)
It is definitely possible to do it programatically though using QMF
(assuming that you are talking about using the C++ broker?). In QMF
there is a purge method on the queue Management Object that allows you
to remove a specified number of messages (zero for all messages).
The down side is that it takes something of an up front effort to do
things using QMF from a standing start, with C# it's probably even
harder (I've done things with Java, Python, C++ and JavaScript).
You can actually do this using the REST API that is the back end of my
GUI (it's a POST method taking a JSON String with the QMF arguments) but
it's some effort to do from a standing start.
Have you thought about addressing your problem from a different
direction. For example if the first thing that you want to do when you
connect your receiver is to clear the queues I'm wondering why you need
the queues in place at all when the receiver is not connected? Couldn't
you set up your receiver's Address to declare a queue (create receive)
and make it an exclusive auto delete queue. In other words create the
receiver's queues when it connects and delete them when it disconnects.
That'll have the same net effect as having clear queues when the
receiver connects.
I'm assuming of course that the queues in question are exclusive to the
receiver in question, if you want to clear a queue shared with another
receiver when a particular receiver connects then I think that QMF purge
or receive/discard are your only two options.
Unless anyone else has a brighter idea??
I couldn't tell you how to do this in C# but I've attached an example in
Java that uses my QMF2 API. It's called QueueFuse it was a little
experiment whereby instead of using circular queues I intercepted the
queueThresholdExceeded QMF Event and invoked purge to delete ~20% of the
messages on the queue that triggered the Event.
This application is bundled in the archive with the GUI and QMF2 API.
Unfortunately as I say it's Java/REST/JavaScript centric but hopefully
it might give you some ideas. The Console class has the code to show how
to construct the Map message necessary to invoke a QMF method so if
you're keen you could use that as a basis for constructing the same
message in C#
HTH
Frase
On 14/02/13 15:26, Rajesh Khan wrote:
I am currently designing a receiver in C# and I wanted to know if there was
a way for me to clear the queues (all its items) as soon as I connect to it
. Any suggestions in this regard would be appreciated.
/*
*
* 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.
*
*/
package org.apache.qpid.qmf2.tools;
// JMS Imports
import javax.jms.Connection;
// Misc Imports
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
// QMF2 Imports
import org.apache.qpid.qmf2.common.ObjectId;
import org.apache.qpid.qmf2.common.QmfData;
import org.apache.qpid.qmf2.common.QmfEvent;
import org.apache.qpid.qmf2.common.QmfEventListener;
import org.apache.qpid.qmf2.common.QmfException;
import org.apache.qpid.qmf2.common.SchemaClassId;
import org.apache.qpid.qmf2.common.WorkItem;
import org.apache.qpid.qmf2.console.Agent;
import org.apache.qpid.qmf2.console.Console;
import org.apache.qpid.qmf2.console.EventReceivedWorkItem;
import org.apache.qpid.qmf2.console.QmfConsoleData;
import org.apache.qpid.qmf2.util.ConnectionHelper;
import org.apache.qpid.qmf2.util.GetOpt;
/**
* QueueFuse provides protection to message producers from consumers who can't consume messages fast enough.
* <p>
* With the default "reject" limit policy when a queue exceeds its capacity an exception is thrown to the
* producer. This behaviour is unfortunate, because if there happen to be multiple consumers consuming
* messages from a given producer it is possible for a single slow consumer to cause message flow to be
* stopped to <u>all</u> consumers, in other words a de-facto denial of service may take place.
* <p>
* In an Enterprise environment it is likely that this sort of behaviour is unwelcome, so QueueFuse makes it
* possible for queueThresholdExceeded Events to be detected and for the offending queues to have messages
* purged, thus protecting the other consumers by preventing an exception being thrown to the message producer.
* <p>
* The original intention with this class was to unbind bindings to queues that exceed the threshold. This method
* works, but it has a number of disadvantages. In particular there is no way to unbind from (and thus protect)
* queues bound to the default direct exchange, in addition in order to unbind it is necessary to retrieve
* binding and exchange information, both of which require further exchanges with the broker (which is not
* desirable as when the queueThresholdExceeded occurs we need to act pretty quickly). Finally as it happens
* it is also necessary to purge some messages after unbinding anyway as if this is not done the queue remains
* in the flowStopped state and producers will eventually time out and throw an exception if this is not cleared.
* So all in all simply purging each time we cross the threshold is simpler and has the additional advantage that
* if and when the consumer speeds up message delivery will eventually return to normal.
*
* <pre>
* Usage: QueueFuse [options] [broker-addr]...
*
* Monitors one or more Qpid message brokers for queueThresholdExceeded Events.
*
* If a queueThresholdExceeded Event occurs messages are purged from the queue,
* in other words this class behaves rather like a fuse 'blowing' if the
* threshold gets exceeded.
*
* If no broker-addr is supplied, QueueFuse connects to 'localhost:5672'.
*
* [broker-addr] syntax:
*
* [username/password@] hostname
* ip-address [:<port>]
*
* Examples:
*
* $ QueueFuse localhost:5672
* $ QueueFuse 10.1.1.7:10000
* $ QueueFuse guest/guest@broker-host:10000
*
* Options:
* -h, --help show this help message and exit
* -f <filter>, --filter=<filter>
* a list of comma separated queue names (regex are
* accepted) to protect (default is to protect all).
* -p <PERCENT>, --purge=<PERCENT>\n" +
* The percentage of messages to purge when the queue\n" +
* threshold gets exceeded (default = 20%).\n" +
* N.B. if this gets set too low the fuse may not blow.\n" +
* --sasl-mechanism=<mech>
* SASL mechanism for authentication (e.g. EXTERNAL,
* ANONYMOUS, PLAIN, CRAM-MD, DIGEST-MD5, GSSAPI). SASL
* automatically picks the most secure available
* mechanism - use this option to override.
* </pre>
* @author Fraser Adams
*/
public final class QueueFuse implements QmfEventListener
{
private static final String _usage =
"Usage: QueueFuse [options] [broker-addr]...\n";
private static final String _description =
"Monitors one or more Qpid message brokers for queueThresholdExceeded Events.\n" +
"\n" +
"If a queueThresholdExceeded Event occurs messages are purged from the queue,\n" +
"in other words this class behaves rather like a fuse 'blowing' if the\n" +
"threshold gets exceeded.\n" +
"\n" +
"If no broker-addr is supplied, QueueFuse connects to 'localhost:5672'.\n" +
"\n" +
"[broker-addr] syntax:\n" +
"\n" +
"[username/password@] hostname\n" +
"ip-address [:<port>]\n" +
"\n" +
"Examples:\n" +
"\n" +
"$ QueueFuse localhost:5672\n" +
"$ QueueFuse 10.1.1.7:10000\n" +
"$ QueueFuse guest/guest@broker-host:10000\n";
private static final String _options =
"Options:\n" +
" -h, --help show this help message and exit\n" +
" -f <filter>, --filter=<filter>\n" +
" a list of comma separated queue names (regex are\n" +
" accepted) to protect (default is to protect all).\n" +
" -p <PERCENT>, --purge=<PERCENT>\n" +
" The percentage of messages to purge when the queue\n" +
" threshold gets exceeded (default = 20%).\n" +
" N.B. if this gets set too low the fuse may not blow.\n" +
" --sasl-mechanism=<mech>\n" +
" SASL mechanism for authentication (e.g. EXTERNAL,\n" +
" ANONYMOUS, PLAIN, CRAM-MD, DIGEST-MD5, GSSAPI). SASL\n" +
" automatically picks the most secure available\n" +
" mechanism - use this option to override.\n";
private final String _url;
private final List<Pattern> _filter;
private final float _purge;
private Map<String, QmfConsoleData> _queueCache = new HashMap<String, QmfConsoleData>(50);
private Console _console;
/**
* Basic constructor. Creates JMS Session, Initialises Destinations, Producers & Consumers and starts connection.
* @param url the connection URL.
* @param connectionOptions the options String to pass to ConnectionHelper.
* @param filter a list of regex Patterns used to choose the queues we wish to protect.
* @param purge the ratio of messages that we wish to purge if the threshold gets exceeded.
*/
public QueueFuse(final String url, final String connectionOptions, final List<Pattern> filter, final float purge)
{
System.out.println("QueueFuse Connecting to " + url);
if (filter.size() > 0)
{
System.out.println("Filter = " + filter);
}
_url = url;
_filter = filter;
_purge = purge;
try
{
Connection connection = ConnectionHelper.createConnection(url, connectionOptions);
_console = new Console(this);
_console.addConnection(connection);
updateQueueCache();
}
catch (QmfException qmfe)
{
System.err.println("QmfException " + qmfe.getMessage() + " caught in QueueFuse constructor");
}
}
/**
* Looks up queue objects and stores them in _queueCache keyed by the queue name
*/
private void updateQueueCache()
{
_queueCache.clear();
List<QmfConsoleData> queues = _console.getObjects("org.apache.qpid.broker", "queue");
for (QmfConsoleData queue : queues)
{
String queueName = queue.getStringValue("name");
_queueCache.put(queueName, queue);
}
}
/**
* Look up a queue object with the given name and if it's not a ring queue invoke the queue's purge method.
* @param queueName the name of the queue to purge
* @param msgDepth the number of messages on the queue, used to determine how many messages to purge.
*/
private void purgeQueue(final String queueName, long msgDepth)
{
QmfConsoleData queue = _queueCache.get(queueName);
if (queue == null)
{
System.out.printf("%s ERROR QueueFuse.disconnectQueue() %s reference couldn't be found\n",
new Date().toString(), queueName);
}
else
{ // If we've found a queue called queueName we then find the bindings that reference it.
Map args = (Map)queue.getValue("arguments");
String policyType = (String)args.get("qpid.policy_type");
if (policyType != null && policyType.equals("ring"))
{ // If qpid.policy_type=ring we return.
return;
}
try
{
QmfData arguments = new QmfData();
arguments.setValue("request", (long)(_purge*msgDepth));
queue.invokeMethod("purge", arguments);
}
catch (QmfException e)
{
System.out.println(e.getMessage());
}
}
}
/**
* Main Event handler.
* @param wi a QMF2 WorkItem object
*/
public void onEvent(final WorkItem wi)
{
if (wi instanceof EventReceivedWorkItem)
{
EventReceivedWorkItem item = (EventReceivedWorkItem)wi;
Agent agent = item.getAgent();
QmfEvent event = item.getEvent();
String className = event.getSchemaClassId().getClassName();
if (className.equals("queueDeclare"))
{
updateQueueCache();
}
else if (className.equals("queueThresholdExceeded"))
{
String queueName = event.getStringValue("qName");
boolean matches = false;
for (Pattern x : _filter)
{ // Check the queue name against the regexes in the filter List (if any)
Matcher m = x.matcher(queueName);
if (m.find())
{
matches = true;
break;
}
}
if (_filter.isEmpty() || matches)
{ // If there's no filter enabled or the filter matches the queue name we call purgeQueue().
long msgDepth = event.getLongValue("msgDepth");
purgeQueue(queueName, msgDepth);
}
}
}
}
/**
* Runs QueueFuse.
* @param args the command line arguments.
*/
public static void main(final String[] args)
{
String logLevel = System.getProperty("amqj.logging.level");
logLevel = (logLevel == null) ? "FATAL" : logLevel; // Set default log level to FATAL rather than DEBUG.
System.setProperty("amqj.logging.level", logLevel);
String[] longOpts = {"help", "filter=", "purge=", "sasl-mechanism="};
try
{
boolean includeRingQueues = false;
String connectionOptions = "{reconnect: true}";
List<Pattern> filter = new ArrayList<Pattern>();
float purge = 0.2f;
GetOpt getopt = new GetOpt(args, "hf:p:", longOpts);
List<String[]> optList = getopt.getOptList();
String[] cargs = {};
cargs = getopt.getEncArgs().toArray(cargs);
for (String[] opt : optList)
{
if (opt[0].equals("-h") || opt[0].equals("--help"))
{
System.out.println(_usage);
System.out.println(_description);
System.out.println(_options);
System.exit(1);
}
else if (opt[0].equals("-f") || opt[0].equals("--filter"))
{
String[] split = opt[1].split(",");
for (String s : split)
{
Pattern p = Pattern.compile(s);
filter.add(p);
}
}
else if (opt[0].equals("-p") || opt[0].equals("--purge"))
{
int percent = Integer.parseInt(opt[1]);
if (percent < 0 || percent > 100)
{
System.out.println(_usage);
System.exit(1);
}
purge = percent/100.0f;
}
else if (opt[0].equals("--sasl-mechanism"))
{
connectionOptions = "{reconnect: true, sasl_mechs: " + opt[1] + "}";
}
}
int nargs = cargs.length;
if (nargs == 0)
{
cargs = new String[] {"localhost"};
}
for (String url : cargs)
{
QueueFuse queueFuse = new QueueFuse(url, connectionOptions, filter, purge);
}
}
catch (IllegalArgumentException e)
{
System.out.println(_usage);
System.out.println(e.getMessage());
System.exit(1);
}
try
{ // Block here
Thread.currentThread().join();
}
catch (InterruptedException ie)
{
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
