QPID-2327 - Enhance qpid-config to deal with xml and headers brokers

Committed patch from John Dunning


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

4 files changed, 229 insertions, 6 deletions

diff --git a/cpp/src/tests/cli_tests.py b/cpp/src/tests/cli_tests.py
index a65097d431..4df9169cc4 100755
--- a/cpp/src/tests/cli_tests.py
+++ b/cpp/src/tests/cli_tests.py
@@ -99,6 +99,116 @@ class CliTests(TestBase010):
                 found = True
         self.assertEqual(found, False)
 
+        # helpers for some of the test methods
+    def helper_find_exchange(self, xchgname, typ, expected=True):
+        xchgs = self.qmf.getObjects(_class = "exchange")
+        found = False
+        for xchg in xchgs:
+            if xchg.name == xchgname:
+                if typ:
+                    self.assertEqual(xchg.type, typ)
+                found = True
+        self.assertEqual(found, expected)
+
+    def helper_create_exchange(self, xchgname, typ="direct", opts=""):
+        foo = self.command(opts + " add exchange " + typ + " " + xchgname)
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+        self.helper_find_exchange(xchgname, typ, True)
+
+    def helper_destroy_exchange(self, xchgname):
+        foo = self.command(" del exchange " + xchgname)
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+        self.helper_find_exchange(xchgname, False, expected=False)
+
+    def helper_find_queue(self, qname, expected=True):
+        queues = self.qmf.getObjects(_class="queue")
+        found = False
+        for queue in queues:
+            if queue.name == qname:
+                self.assertEqual(queue.durable, False)
+                found = True
+        self.assertEqual(found, expected)
+
+    def helper_create_queue(self, qname):
+        foo = self.command(" add queue " + qname)
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+        self.helper_find_queue(qname, True)
+
+    def helper_destroy_queue(self, qname):
+        foo = self.command(" del queue " + qname)
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+        self.helper_find_queue(qname, False)
+
+
+        # test the bind-queue-to-header-exchange functionality
+    def test_qpid_config_headers(self):
+        self.startQmf();
+        qmf = self.qmf
+        qname = "test_qpid_config"
+        xchgname = "test_xchg"
+
+        # first create a header xchg
+        self.helper_create_exchange(xchgname, typ="headers")
+
+        # create the queue
+        self.helper_create_queue(qname)
+
+        # now bind the queue to the xchg
+        foo = self.command(" bind " + xchgname + " " + qname + 
+                                     " key all foo=bar baz=quux")
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+
+        # he likes it, mikey.  Ok, now tear it all down.  first the binding
+        ret = os.system(self.command(" unbind " + xchgname + " " + qname +
+                                     " key"))
+        self.assertEqual(ret, 0)
+
+        # then the queue 
+        self.helper_destroy_queue(qname)
+
+        # then the exchange
+        self.helper_destroy_exchange(xchgname)
+
+        # test the bind-queue-xml-filter functionality
+    def test_qpid_config_xml(self):
+        self.startQmf();
+        qmf = self.qmf
+        qname = "test_qpid_config"
+        xchgname = "test_xchg"
+
+        # first create a header xchg
+        self.helper_create_exchange(xchgname, typ="xml")
+
+        # create the queue
+        self.helper_create_queue(qname)
+
+        # now bind the queue to the xchg
+        foo = self.command("-f test.xquery bind " + xchgname + " " + qname)
+        # print foo
+        ret = os.system(foo)
+        self.assertEqual(ret, 0)
+
+        # he likes it, mikey.  Ok, now tear it all down.  first the binding
+        ret = os.system(self.command(" unbind " + xchgname + " " + qname +
+                                     " key"))
+        self.assertEqual(ret, 0)
+
+        # then the queue 
+        self.helper_destroy_queue(qname)
+
+        # then the exchange
+        self.helper_destroy_exchange(xchgname)
+
     def test_qpid_config_durable(self):
         self.startQmf();
         qmf = self.qmf
diff --git a/cpp/src/tests/run_cli_tests b/cpp/src/tests/run_cli_tests
index 34ce3c4f1a..f3cdea0d4f 100755
--- a/cpp/src/tests/run_cli_tests
+++ b/cpp/src/tests/run_cli_tests
@@ -26,10 +26,39 @@ CLI_DIR=$PYTHON_COMMANDS
 
 trap stop_brokers INT TERM QUIT
 
+# helper function to create test.xquery in the current directory, so
+# that the python test program can find it.  yes, it leaves a turd.
+create_test_xquery() {
+     cat <<EOF > ./test.xquery
+    let \$w := ./weather
+    return \$w/station = 'Raleigh-Durham International Airport (KRDU)'
+       and \$w/temperature_f > 50
+       and \$w/temperature_f - \$w/dewpoint > 5
+       and \$w/wind_speed_mph > 7
+       and \$w/wind_speed_mph < 20
+EOF
+}
+
 start_brokers() {
-    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no > qpidd.port
+    # if the xml lib is present, use it.  if not, disable any tests which
+    # look like they're xml related.
+    # if we start supporting xml on windows, it will need something similar 
+    # here
+    if [ -f ../.libs/xml.so ] ; then
+	xargs="--load-module ../.libs/xml.so"
+	if [ ! -f test.xquery ] ; then 
+	    create_test_xquery
+	fi
+	targs=""
+    else
+	echo "Ignoring XML tests"
+	xargs=""
+	targs="--ignore=*xml*"
+    fi
+
+    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no $xargs > qpidd.port
     LOCAL_PORT=`cat qpidd.port`
-    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no > qpidd.port
+    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no $xargs > qpidd.port
     REMOTE_PORT=`cat qpidd.port`
 }
 
@@ -41,7 +70,7 @@ stop_brokers() {
 if test -d ${PYTHON_DIR} ;  then
     start_brokers
     echo "Running CLI tests using brokers on ports $LOCAL_PORT $REMOTE_PORT"
-    $PYTHON_COMMANDS/qpid-python-test -m cli_tests -b localhost:$LOCAL_PORT -Dremote-port=$REMOTE_PORT -Dcli-dir=$CLI_DIR $@
+    $PYTHON_COMMANDS/qpid-python-test -m cli_tests -b localhost:$LOCAL_PORT -Dremote-port=$REMOTE_PORT -Dcli-dir=$CLI_DIR $targs $@
     RETCODE=$?
     stop_brokers
     if test x$RETCODE != x0; then 
diff --git a/cpp/src/tests/test.xquery b/cpp/src/tests/test.xquery
new file mode 100644
index 0000000000..4cfe3af02d
--- /dev/null
+++ b/cpp/src/tests/test.xquery
@@ -0,0 +1,6 @@
+   let $w := ./weather
+   return $w/station = 'Raleigh-Durham International Airport (KRDU)'
+      and $w/temperature_f > 50
+      and $w/temperature_f - $w/dewpoint > 5
+      and $w/wind_speed_mph > 7
+      and $w/wind_speed_mph < 20
diff --git a/python/commands/qpid-config b/python/commands/qpid-config
index 39af67f39c..5b7dd68f90 100755
--- a/python/commands/qpid-config
+++ b/python/commands/qpid-config
@@ -24,6 +24,7 @@ import getopt
 import sys
 import locale
 from qmf.console import Session
+import io
 
 _recursive         = False
 _host              = "localhost"
@@ -43,6 +44,7 @@ _order             = None
 _msgSequence       = False
 _ive               = False
 _eventGeneration   = None
+_file              = None
 
 FILECOUNT = "qpid.file_count"
 FILESIZE  = "qpid.file_size"
@@ -65,6 +67,8 @@ def Usage ():
     print "        qpid-config [OPTIONS] add queue <name> [AddQueueOptions]"
     print "        qpid-config [OPTIONS] del queue <name> [DelQueueOptions]"
     print "        qpid-config [OPTIONS] bind   <exchange-name> <queue-name> [binding-key]"
+    print "                  <for type xml>     [-f -|filename]"
+    print "                  <for type header>  [all|any] k1=v1 [, k2=v2...]"
     print "        qpid-config [OPTIONS] unbind <exchange-name> <queue-name> [binding-key]"
     print
     print "Options:"
@@ -134,6 +138,46 @@ def Usage ():
     print
     sys.exit (1)
 
+
+#
+# helpers for the arg parsing in bind().  return multiple values; "ok"
+# followed by the resultant args
+
+#
+# accept -f followed by either
+# a filename or "-", for stdin.  pull the bits into a string, to be 
+# passed to the xml binding.
+#
+def snarf_xquery_args():
+    if not _file:
+        print "Invalid args to bind xml:  need an input file or stdin"
+        return [False]
+    if _file == "-":
+        res = sys.stdin.read()
+    else:
+        f = io.open(_file)   # let this signal if it can't find it
+        res = f.read()
+        f.close()
+    return [True, res]
+    
+#
+# look for "any"/"all" and grok the rest of argv into a map
+#
+def snarf_header_args(cargs):
+    if len(cargs) < 2:
+        print "Invalid args to bind headers:  need 'any'/'all' plus conditions"
+        return [False]
+    op = cargs[0]
+    if op == "all" or op == "any":
+        kv = {}
+        for thing in cargs[1:]:
+            k_and_v = thing.split("=")
+            kv[k_and_v[0]] = k_and_v[1]
+        return [True, op, kv]
+    else:
+        print "Invalid condition arg to bind headers, need 'any' or 'all', not '" + op + "'"
+        return [False]
+
 class BrokerManager:
     def __init__ (self):
         self.brokerName = None
@@ -344,7 +388,36 @@ class BrokerManager:
         key   = ""
         if len (args) > 2:
             key = args[2]
-        self.broker.getAmqpSession().exchange_bind (queue=qname, exchange=ename, binding_key=key)
+
+        # query the exchange to determine its type.
+        res = self.broker.getAmqpSession().exchange_query(ename)
+
+        # type of the xchg determines the processing of the rest of
+        # argv.  if it's an xml xchg, we want to find a file
+        # containing an x-query, and pass that.  if it's a headers
+        # exchange, we need to pass either "any" or all, followed by a
+        # map containing key/value pairs.  if neither of those, extra
+        # args are ignored.
+        ok = True
+        args = None
+        if res.type == "xml":
+            # this checks/imports the -f arg
+            [ok, xquery] = snarf_xquery_args()
+            args = { "xquery" : xquery }
+            # print args
+        else:
+            if res.type == "headers":
+                [ok, op, kv] = snarf_header_args(cargs[4:])
+                args = kv
+                args["x-match"] = op
+
+        if not ok:
+            sys.exit(1)
+
+        self.broker.getAmqpSession().exchange_bind (queue=qname,
+                                                    exchange=ename,
+                                                    binding_key=key,
+                                                    arguments=args)
 
     def Unbind (self, args):
         if len (args) < 2:
@@ -383,8 +456,8 @@ try:
     longOpts = ("durable", "cluster-durable", "bindings", "broker-addr=", "file-count=",
                 "file-size=", "max-queue-size=", "max-queue-count=", "limit-policy=",
                 "order=", "sequence", "ive", "generate-queue-events=", "force", "force-if-not-empty",
-                "force_if_used", "alternate-exchange=", "passive", "timeout=")
-    (optlist, encArgs) = getopt.gnu_getopt (sys.argv[1:], "a:b", longOpts)
+                "force_if_used", "alternate-exchange=", "passive", "timeout=", "file=")
+    (optlist, encArgs) = getopt.gnu_getopt (sys.argv[1:], "a:bf:", longOpts)
 except:
     Usage ()
 
@@ -399,6 +472,8 @@ for opt in optlist:
         _recursive = True
     if opt[0] == "-a" or opt[0] == "--broker-addr":
         _host = opt[1]
+    if opt[0] == "-f" or opt[0] == "--file":
+        _file = opt[1]
     if opt[0] == "--timeout":
         _connTimeout = int(opt[1])
         if _connTimeout == 0:
@@ -488,6 +563,9 @@ try:
             Usage ()
 except KeyboardInterrupt:
     print
+except IOError, e:
+    print e
+    sys.exit(1)
 except Exception,e:
     print "Failed: %s: %s" % (e.__class__.__name__, e)
     sys.exit(1)