You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2006/09/20 00:07:25 UTC
svn commit: r447994 [43/46] - in /incubator/qpid/trunk/qpid: ./ cpp/
cpp/bin/ cpp/broker/ cpp/broker/inc/ cpp/broker/src/ cpp/broker/test/
cpp/client/ cpp/client/inc/ cpp/client/src/ cpp/client/test/ cpp/common/
cpp/common/concurrent/ cpp/common/concur...
Added: incubator/qpid/trunk/qpid/python/qpid/spec.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/spec.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/spec.py (added)
+++ incubator/qpid/trunk/qpid/python/qpid/spec.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,349 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+"""
+This module loads protocol metadata into python objects. It provides
+access to spec metadata via a python object model, and can also
+dynamically creating python methods, classes, and modules based on the
+spec metadata. All the generated methods have proper signatures and
+doc strings based on the spec metadata so the python help system can
+be used to browse the spec documentation. The generated methods all
+dispatch to the self.invoke(meth, args) callback of the containing
+class so that the generated code can be reused in a variety of
+situations.
+"""
+
+import re, textwrap, new, xmlutil
+
+class SpecContainer:
+
+ def __init__(self):
+ self.items = []
+ self.byname = {}
+ self.byid = {}
+ self.indexes = {}
+ self.bypyname = {}
+
+ def add(self, item):
+ if self.byname.has_key(item.name):
+ raise ValueError("duplicate name: %s" % item)
+ if self.byid.has_key(item.id):
+ raise ValueError("duplicate id: %s" % item)
+ pyname = pythonize(item.name)
+ if self.bypyname.has_key(pyname):
+ raise ValueError("duplicate pyname: %s" % item)
+ self.indexes[item] = len(self.items)
+ self.items.append(item)
+ self.byname[item.name] = item
+ self.byid[item.id] = item
+ self.bypyname[pyname] = item
+
+ def index(self, item):
+ try:
+ return self.indexes[item]
+ except KeyError:
+ raise ValueError(item)
+
+ def __iter__(self):
+ return iter(self.items)
+
+ def __len__(self):
+ return len(self.items)
+
+class Metadata:
+
+ PRINT = []
+
+ def __init__(self):
+ pass
+
+ def __str__(self):
+ args = map(lambda f: "%s=%s" % (f, getattr(self, f)), self.PRINT)
+ return "%s(%s)" % (self.__class__.__name__, ", ".join(args))
+
+ def __repr__(self):
+ return str(self)
+
+class Spec(Metadata):
+
+ PRINT=["major", "minor", "file"]
+
+ def __init__(self, major, minor, file):
+ Metadata.__init__(self)
+ self.major = major
+ self.minor = minor
+ self.file = file
+ self.constants = SpecContainer()
+ self.classes = SpecContainer()
+
+ def post_load(self):
+ self.module = self.define_module("amqp%s%s" % (self.major, self.minor))
+ self.klass = self.define_class("Amqp%s%s" % (self.major, self.minor))
+
+ def parse_method(self, name):
+ parts = re.split(r"\s*\.\s*", name)
+ if len(parts) != 2:
+ raise ValueError(name)
+ klass, meth = parts
+ return self.classes.byname[klass].methods.byname[meth]
+
+ def define_module(self, name, doc = None):
+ module = new.module(name, doc)
+ module.__file__ = self.file
+ for c in self.classes:
+ classname = pythonize(c.name)
+ cls = c.define_class(classname)
+ cls.__module__ = module.__name__
+ setattr(module, classname, cls)
+ return module
+
+ def define_class(self, name):
+ methods = {}
+ for c in self.classes:
+ for m in c.methods:
+ meth = pythonize(m.klass.name + "_" + m.name)
+ methods[meth] = m.define_method(meth)
+ return type(name, (), methods)
+
+class Constant(Metadata):
+
+ PRINT=["name", "id"]
+
+ def __init__(self, spec, name, id, klass, docs):
+ Metadata.__init__(self)
+ self.spec = spec
+ self.name = name
+ self.id = id
+ self.klass = klass
+ self.docs = docs
+
+class Class(Metadata):
+
+ PRINT=["name", "id"]
+
+ def __init__(self, spec, name, id, handler, docs):
+ Metadata.__init__(self)
+ self.spec = spec
+ self.name = name
+ self.id = id
+ self.handler = handler
+ self.fields = SpecContainer()
+ self.methods = SpecContainer()
+ self.docs = docs
+
+ def define_class(self, name):
+ methods = {}
+ for m in self.methods:
+ meth = pythonize(m.name)
+ methods[meth] = m.define_method(meth)
+ return type(name, (), methods)
+
+class Method(Metadata):
+
+ PRINT=["name", "id"]
+
+ def __init__(self, klass, name, id, content, responses, synchronous,
+ description, docs):
+ Metadata.__init__(self)
+ self.klass = klass
+ self.name = name
+ self.id = id
+ self.content = content
+ self.responses = responses
+ self.synchronous = synchronous
+ self.fields = SpecContainer()
+ self.description = description
+ self.docs = docs
+ self.response = False
+
+ def docstring(self):
+ s = "\n\n".join([fill(d, 2) for d in [self.description] + self.docs])
+ for f in self.fields:
+ if f.docs:
+ s += "\n\n" + "\n\n".join([fill(f.docs[0], 4, pythonize(f.name))] +
+ [fill(d, 4) for d in f.docs[1:]])
+ return s
+
+ METHOD = "__method__"
+ DEFAULTS = {"bit": False,
+ "shortstr": "",
+ "longstr": "",
+ "table": {},
+ "octet": 0,
+ "short": 0,
+ "long": 0,
+ "longlong": 0}
+
+ def define_method(self, name):
+ g = {Method.METHOD: self}
+ l = {}
+ args = [(pythonize(f.name), Method.DEFAULTS[f.type]) for f in self.fields]
+ if self.content:
+ args += [("content", None)]
+ code = "def %s(self, %s):\n" % \
+ (name, ", ".join(["%s = %r" % a for a in args]))
+ code += " %r\n" % self.docstring()
+ if self.content:
+ methargs = args[:-1]
+ else:
+ methargs = args
+ argnames = ", ".join([a[0] for a in methargs])
+ code += " return self.invoke(%s" % Method.METHOD
+ if argnames:
+ code += ", (%s,)" % argnames
+ if self.content:
+ code += ", content"
+ code += ")"
+ exec code in g, l
+ return l[name]
+
+class Field(Metadata):
+
+ PRINT=["name", "id", "type"]
+
+ def __init__(self, name, id, type, docs):
+ Metadata.__init__(self)
+ self.name = name
+ self.id = id
+ self.type = type
+ self.docs = docs
+
+def get_docs(nd):
+ return [n.text for n in nd["doc"]]
+
+def load_fields(nd, l, domains):
+ for f_nd in nd["field"]:
+ try:
+ type = f_nd["@type"]
+ except KeyError:
+ type = domains[f_nd["@domain"]]
+ l.add(Field(f_nd["@name"], f_nd.index(), type, get_docs(f_nd)))
+
+def load(specfile):
+ doc = xmlutil.parse(specfile)
+ root = doc["amqp"][0]
+ spec = Spec(int(root["@major"]), int(root["@minor"]), specfile)
+
+ # constants
+ for nd in root["constant"]:
+ const = Constant(spec, nd["@name"], int(nd["@value"]), nd.get("@class"),
+ get_docs(nd))
+ spec.constants.add(const)
+
+ # domains are typedefs
+ domains = {}
+ for nd in root["domain"]:
+ domains[nd["@name"]] = nd["@type"]
+
+ # classes
+ for c_nd in root["class"]:
+ klass = Class(spec, c_nd["@name"], int(c_nd["@index"]), c_nd["@handler"],
+ get_docs(c_nd))
+ load_fields(c_nd, klass.fields, domains)
+ for m_nd in c_nd["method"]:
+ meth = Method(klass, m_nd["@name"],
+ int(m_nd["@index"]),
+ m_nd.get_bool("@content", False),
+ [nd["@name"] for nd in m_nd["response"]],
+ m_nd.get_bool("@synchronous", False),
+ m_nd.text,
+ get_docs(m_nd))
+ load_fields(m_nd, meth.fields, domains)
+ klass.methods.add(meth)
+ # resolve the responses
+ for m in klass.methods:
+ m.responses = [klass.methods.byname[r] for r in m.responses]
+ for resp in m.responses:
+ resp.response = True
+ spec.classes.add(klass)
+ spec.post_load()
+ return spec
+
+REPLACE = {" ": "_", "-": "_"}
+KEYWORDS = {"global": "global_",
+ "return": "return_"}
+
+def pythonize(name):
+ name = str(name)
+ for key, val in REPLACE.items():
+ name = name.replace(key, val)
+ try:
+ name = KEYWORDS[name]
+ except KeyError:
+ pass
+ return name
+
+def fill(text, indent, heading = None):
+ sub = indent * " "
+ if heading:
+ init = (indent - 2) * " " + heading + " -- "
+ else:
+ init = sub
+ w = textwrap.TextWrapper(initial_indent = init, subsequent_indent = sub)
+ return w.fill(" ".join(text.split()))
+
+class Rule(Metadata):
+
+ PRINT = ["text", "implement", "tests"]
+
+ def __init__(self, text, implement, tests, path):
+ self.text = text
+ self.implement = implement
+ self.tests = tests
+ self.path = path
+
+def find_rules(node, rules):
+ if node.name == "rule":
+ rules.append(Rule(node.text, node.get("@implement"),
+ [ch.text for ch in node if ch.name == "test"],
+ node.path()))
+ if node.name == "doc" and node.get("@name") == "rule":
+ tests = []
+ if node.has("@test"):
+ tests.append(node["@test"])
+ rules.append(Rule(node.text, None, tests, node.path()))
+ for child in node:
+ find_rules(child, rules)
+
+def load_rules(specfile):
+ rules = []
+ find_rules(xmlutil.parse(specfile), rules)
+ return rules
+
+def test_summary():
+ template = """
+ <html><head><title>AMQP Tests</title></head>
+ <body>
+ <table width="80%%" align="center">
+ %s
+ </table>
+ </body>
+ </html>
+ """
+ rows = []
+ for rule in load_rules("amqp.org/specs/amqp7.xml"):
+ if rule.tests:
+ tests = ", ".join(rule.tests)
+ else:
+ tests = " "
+ rows.append('<tr bgcolor="#EEEEEE"><td><b>Path:</b> %s</td>'
+ '<td><b>Implement:</b> %s</td>'
+ '<td><b>Tests:</b> %s</td></tr>' %
+ (rule.path[len("/root/amqp"):], rule.implement, tests))
+ rows.append('<tr><td colspan="3">%s</td></tr>' % rule.text)
+ rows.append('<tr><td colspan="3"> </td></tr>')
+
+ print template % "\n".join(rows)
Propchange: incubator/qpid/trunk/qpid/python/qpid/spec.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/qpid/testlib.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/testlib.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/testlib.py (added)
+++ incubator/qpid/trunk/qpid/python/qpid/testlib.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,221 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+# Support library for qpid python tests.
+#
+
+import sys, re, unittest, os, random, logging
+import qpid.client, qpid.spec
+from getopt import getopt, GetoptError
+
+
+def findmodules(root):
+ """Find potential python modules under directory root"""
+ found = []
+ for dirpath, subdirs, files in os.walk(root):
+ modpath = dirpath.replace(os.sep, '.')
+ if not re.match(r'\.svn$', dirpath): # Avoid SVN directories
+ for f in files:
+ match = re.match(r'(.+)\.py$', f)
+ if match and f != '__init__.py':
+ found.append('.'.join([modpath, match.group(1)]))
+ return found
+
+def default(value, default):
+ if (value == None): return default
+ else: return value
+
+class TestRunner:
+ """Runs unit tests.
+
+ Parses command line arguments, provides utility functions for tests,
+ runs the selected test suite.
+ """
+
+ def _die(self, message = None):
+ if message: print message
+ print """
+run-tests [options] [test*]
+The name of a test is package.module.ClassName.testMethod
+Options:
+ -?/-h/--help : this message
+ -s/--spec <spec.xml> : file containing amqp XML spec
+ -b/--broker [<user>[/<password>]@]<host>[:<port>] : broker to connect to
+ -v/--verbose : verbose - lists tests as they are run.
+ -d/--debug : enable debug logging.
+ -i/--ignore <test> : ignore the named test.
+ -I/--ignore-file : file containing patterns to ignore.
+ """
+ sys.exit(1)
+
+ def setBroker(self, broker):
+ rex = re.compile(r"""
+ # [ <user> [ / <password> ] @] <host> [ :<port> ]
+ ^ (?: ([^/]*) (?: / ([^@]*) )? @)? ([^:]+) (?: :([0-9]+))?$""", re.X)
+ match = rex.match(broker)
+ if not match: self._die("'%s' is not a valid broker" % (broker))
+ self.user, self.password, self.host, self.port = match.groups()
+ self.port = int(default(self.port, 5672))
+ self.user = default(self.user, "guest")
+ self.password = default(self.password, "guest")
+
+ def __init__(self):
+ # Defaults
+ self.setBroker("localhost")
+ self.spec = "../specs/amqp-8.0.xml"
+ self.verbose = 1
+ self.ignore = []
+
+ def ignoreFile(self, filename):
+ f = file(filename)
+ for line in f.readlines(): self.ignore.append(line.strip())
+ f.close()
+
+ def _parseargs(self, args):
+ try:
+ opts, self.tests = getopt(args, "s:b:h?dvi:I:", ["help", "spec", "server", "verbose", "ignore", "ignore-file"])
+ except GetoptError, e:
+ self._die(str(e))
+ for opt, value in opts:
+ if opt in ("-?", "-h", "--help"): self._die()
+ if opt in ("-s", "--spec"): self.spec = value
+ if opt in ("-b", "--broker"): self.setBroker(value)
+ if opt in ("-v", "--verbose"): self.verbose = 2
+ if opt in ("-d", "--debug"): logging.basicConfig(level=logging.DEBUG)
+ if opt in ("-i", "--ignore"): self.ignore.append(value)
+ if opt in ("-I", "--ignore-file"): self.ignoreFile(value)
+
+ if len(self.tests) == 0: self.tests=findmodules("tests")
+
+ def testSuite(self):
+ class IgnoringTestSuite(unittest.TestSuite):
+ def addTest(self, test):
+ if isinstance(test, unittest.TestCase) and test.id() in testrunner.ignore:
+ return
+ unittest.TestSuite.addTest(self, test)
+
+ # Use our IgnoringTestSuite in the test loader.
+ unittest.TestLoader.suiteClass = IgnoringTestSuite
+ return unittest.defaultTestLoader.loadTestsFromNames(self.tests)
+
+ def run(self, args=sys.argv[1:]):
+ self._parseargs(args)
+ runner = unittest.TextTestRunner(descriptions=False,
+ verbosity=self.verbose)
+ result = runner.run(self.testSuite())
+ if (self.ignore):
+ print "======================================="
+ print "NOTE: the following tests were ignored:"
+ for t in self.ignore: print t
+ print "======================================="
+ return result.wasSuccessful()
+
+ def connect(self, host=None, port=None, spec=None, user=None, password=None):
+ """Connect to the broker, returns a qpid.client.Client"""
+ host = host or self.host
+ port = port or self.port
+ spec = spec or self.spec
+ user = user or self.user
+ password = password or self.password
+ client = qpid.client.Client(host, port, qpid.spec.load(spec))
+ client.start({"LOGIN": user, "PASSWORD": password})
+ return client
+
+
+# Global instance for tests to call connect.
+testrunner = TestRunner()
+
+
+class TestBase(unittest.TestCase):
+ """Base class for Qpid test cases.
+
+ self.client is automatically connected with channel 1 open before
+ the test methods are run.
+
+ Deletes queues and exchanges after. Tests call
+ self.queue_declare(channel, ...) and self.exchange_declare(chanel,
+ ...) which are wrappers for the Channel functions that note
+ resources to clean up later.
+ """
+
+ def setUp(self):
+ self.queues = []
+ self.exchanges = []
+ self.client = self.connect()
+ self.channel = self.client.channel(1)
+ self.channel.channel_open()
+
+ def tearDown(self):
+ # TODO aconway 2006-09-05: Wrong behaviour here, we should
+ # close all open channels (checking for exceptions on the
+ # channesl) then open a channel to clean up qs and exs,
+ # finally close that channel.
+ for ch, q in self.queues:
+ ch.queue_delete(queue=q)
+ for ch, ex in self.exchanges:
+ ch.exchange_delete(exchange=ex)
+
+ def connect(self, *args, **keys):
+ """Create a new connction, return the Client object"""
+ return testrunner.connect(*args, **keys)
+
+ def queue_declare(self, channel=None, *args, **keys):
+ channel = channel or self.channel
+ reply = channel.queue_declare(*args, **keys)
+ self.queues.append((channel, reply.queue))
+ return reply
+
+ def exchange_declare(self, channel=None, ticket=0, exchange='',
+ type='', passive=False, durable=False,
+ auto_delete=False, internal=False, nowait=False,
+ arguments={}):
+ channel = channel or self.channel
+ reply = channel.exchange_declare(ticket, exchange, type, passive, durable, auto_delete, internal, nowait, arguments)
+ # TODO aconway 2006-09-14: Don't add exchange on failure.
+ self.exchanges.append((channel,exchange))
+ return reply
+
+ def assertPublishConsume(self, queue="", exchange="", routing_key=""):
+ """
+ Publish a message and consume it, assert it comes back intact.
+
+ queue can be a single queue name or a list of queue names.
+ For a list assert the message appears on all queues.
+ Crude attempt to make unique messages so we can't consume
+ a message not really meant for us.
+ """
+ body = "TestMessage("+str(random.randint(999999, 1000000))+")"
+ self.channel.basic_publish(exchange=exchange,
+ content=qpid.content.Content(body),
+ routing_key=routing_key)
+ if not isinstance(queue, list): queue = [queue]
+ for q in queue:
+ reply = self.channel.basic_consume(queue=q, no_ack=True)
+ msg = self.client.queue(reply.consumer_tag).get(timeout=2)
+ self.assertEqual(body, msg.content.body)
+
+
+ def assertChannelException(self, expectedCode, message):
+ self.assertEqual(message.method.klass.name, "channel")
+ self.assertEqual(message.method.name, "close")
+ self.assertEqual(message.reply_code, expectedCode)
+
+
+ def assertConnectionException(self, expectedCode, message):
+ self.assertEqual(message.method.klass.name, "connection")
+ self.assertEqual(message.method.name, "close")
+ self.assertEqual(message.reply_code, expectedCode)
+
Propchange: incubator/qpid/trunk/qpid/python/qpid/testlib.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/qpid/xmlutil.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/xmlutil.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/xmlutil.py (added)
+++ incubator/qpid/trunk/qpid/python/qpid/xmlutil.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,116 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+"""
+XML utilities used by spec.py
+"""
+
+import xml.sax
+from xml.sax.handler import ContentHandler
+
+def parse(file):
+ doc = Node("root")
+ xml.sax.parse(file, Builder(doc))
+ return doc
+
+class Node:
+
+ def __init__(self, name, attrs = None, text = None, parent = None):
+ self.name = name
+ self.attrs = attrs
+ self.text = text
+ self.parent = parent
+ self.children = []
+ if parent != None:
+ parent.children.append(self)
+
+ def get_bool(self, key, default = False):
+ v = self.get(key)
+ if v == None:
+ return default
+ else:
+ return bool(int(v))
+
+ def index(self):
+ if self.parent:
+ return self.parent.children.index(self)
+ else:
+ return 0
+
+ def has(self, key):
+ try:
+ result = self[key]
+ return True
+ except KeyError:
+ return False
+ except IndexError:
+ return False
+
+ def get(self, key, default = None):
+ if self.has(key):
+ return self[key]
+ else:
+ return default
+
+ def __getitem__(self, key):
+ if callable(key):
+ return filter(key, self.children)
+ else:
+ t = key.__class__
+ meth = "__get%s__" % t.__name__
+ if hasattr(self, meth):
+ return getattr(self, meth)(key)
+ else:
+ raise KeyError(key)
+
+ def __getstr__(self, name):
+ if name[:1] == "@":
+ return self.attrs[name[1:]]
+ else:
+ return self[lambda nd: nd.name == name]
+
+ def __getint__(self, index):
+ return self.children[index]
+
+ def __iter__(self):
+ return iter(self.children)
+
+ def path(self):
+ if self.parent == None:
+ return "/%s" % self.name
+ else:
+ return "%s/%s" % (self.parent.path(), self.name)
+
+class Builder(ContentHandler):
+
+ def __init__(self, start = None):
+ self.node = start
+
+ def __setitem__(self, element, type):
+ self.types[element] = type
+
+ def startElement(self, name, attrs):
+ self.node = Node(name, attrs, None, self.node)
+
+ def endElement(self, name):
+ self.node = self.node.parent
+
+ def characters(self, content):
+ if self.node.text == None:
+ self.node.text = content
+ else:
+ self.node.text += content
+
Propchange: incubator/qpid/trunk/qpid/python/qpid/xmlutil.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/rule2test
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/rule2test?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/rule2test (added)
+++ incubator/qpid/trunk/qpid/python/rule2test Tue Sep 19 15:06:50 2006
@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+
+#
+# Convert rules to tests
+#
+import sys, re, os.path
+from getopt import getopt, GetoptError
+from string import capitalize
+from xml import dom
+from xml.dom.minidom import parse
+
+def camelcase(s):
+ """Convert 'string like this' to 'StringLikeThis'"""
+ return "".join([capitalize(w) for w in re.split(re.compile("\W*"), s)])
+
+def uncapitalize(s): return s[0].lower()+s[1:]
+
+def ancestors(node):
+ "Return iterator of ancestors from top-level element to node"
+ def generator(node):
+ while node and node.parentNode:
+ yield node
+ node = node.parentNode
+ return reversed(list(generator(node)))
+
+def tagAndName(element):
+ nameAttr = element.getAttribute("name");
+ if (nameAttr) : return camelcase(nameAttr) + camelcase(element.tagName)
+ else: return camelcase(element.tagName)
+
+def nodeText(n):
+ """Recursively collect text from all text nodes under n"""
+ if n.nodeType == dom.Node.TEXT_NODE:
+ return n.data
+ if n.childNodes:
+ return reduce(lambda t, c: t + nodeText(c), n.childNodes, "")
+ return ""
+
+def cleanup(docString, level=8):
+ unindent = re.sub("\n[ \t]*", "\n", docString.strip())
+ emptyLines = re.sub("\n\n\n", "\n\n", unindent)
+ indented = re.sub("\n", "\n"+level*" ", emptyLines)
+ return level*" " + indented
+
+def printTest(test, docstring):
+ print "class %s(TestBase):" % test
+ print ' """'
+ print docstring
+ print ' """'
+ print
+ print
+
+def printTests(doc, module):
+ """Returns dictionary { classname : [ (methodname, docstring)* ] * }"""
+ tests = {}
+ rules = doc.getElementsByTagName("rule")
+ for r in rules:
+ path = list(ancestors(r))
+ if module == path[1].getAttribute("name").lower():
+ test = "".join(map(tagAndName, path[2:])) + "Tests"
+ docstring = cleanup(nodeText(r), 4)
+ printTest(test, docstring)
+
+def usage(message=None):
+ if message: print >>sys.stderr, message
+ print >>sys.stderr, """
+rule2test [options] <amqpclass>
+
+Print test classes for each rule for the amqpclass in amqp.xml.
+
+Options:
+ -?/-h/--help : this message
+ -s/--spec <spec.xml> : file containing amqp XML spec
+"""
+ return 1
+
+def main(argv):
+ try: opts, args = getopt(argv[1:], "h?s:", ["help", "spec="])
+ except GetoptError, e: return usage(e)
+ spec = "../specs/amqp.xml" # Default
+ for opt, val in opts:
+ if (opt in ("-h", "-?", "--help")): return usage()
+ if (opt in ("-s", "--spec")): spec = val
+ doc = parse(spec)
+ if len(args) == 0: return usage()
+ printTests(doc, args[0])
+ return 0
+
+if (__name__ == "__main__"): sys.exit(main(sys.argv))
Propchange: incubator/qpid/trunk/qpid/python/rule2test
------------------------------------------------------------------------------
svn:executable = *
Added: incubator/qpid/trunk/qpid/python/run-tests
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/run-tests?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/run-tests (added)
+++ incubator/qpid/trunk/qpid/python/run-tests Tue Sep 19 15:06:50 2006
@@ -0,0 +1,24 @@
+#!/usr/bin/env python
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+import sys
+from qpid.testlib import testrunner
+
+if not testrunner.run(): sys.exit(1)
+
+
+
Propchange: incubator/qpid/trunk/qpid/python/run-tests
------------------------------------------------------------------------------
svn:executable = *
Added: incubator/qpid/trunk/qpid/python/tests/__init__.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/__init__.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/__init__.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/__init__.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1 @@
+# Do not delete - marks this directory as a python package.
Propchange: incubator/qpid/trunk/qpid/python/tests/__init__.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/tests/basic.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/basic.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/basic.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/basic.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,115 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+from qpid.client import Client, Closed
+from qpid.queue import Empty
+from qpid.content import Content
+from qpid.testlib import testrunner, TestBase
+
+class BasicTests(TestBase):
+ """Tests for 'methods' on the amqp basic 'class'"""
+
+ def test_consume_no_local(self):
+ """
+ Test that the no_local flag is honoured in the consume method
+ """
+ channel = self.channel
+ #setup, declare two queues:
+ channel.queue_declare(queue="test-queue-1a", exclusive=True)
+ channel.queue_declare(queue="test-queue-1b", exclusive=True)
+ #establish two consumers one of which excludes delivery of locally sent messages
+ channel.basic_consume(consumer_tag="local_included", queue="test-queue-1a")
+ channel.basic_consume(consumer_tag="local_excluded", queue="test-queue-1b", no_local=True)
+
+ #send a message
+ channel.basic_publish(exchange="amq.direct", routing_key="test-queue-1a", content=Content("consume_no_local"))
+ channel.basic_publish(exchange="amq.direct", routing_key="test-queue-1b", content=Content("consume_no_local"))
+
+ #check the queues of the two consumers
+ excluded = self.client.queue("local_excluded")
+ included = self.client.queue("local_included")
+ msg = included.get(timeout=1)
+ self.assertEqual("consume_no_local", msg.content.body)
+ try:
+ excluded.get(timeout=1)
+ self.fail("Received locally published message though no_local=true")
+ except Empty: None
+
+
+ def test_consume_exclusive(self):
+ """
+ Test that the exclusive flag is honoured in the consume method
+ """
+ channel = self.channel
+ #setup, declare a queue:
+ channel.queue_declare(queue="test-queue-2", exclusive=True)
+
+ #check that an exclusive consumer prevents other consumer being created:
+ channel.basic_consume(consumer_tag="first", queue="test-queue-2", exclusive=True)
+ try:
+ channel.basic_consume(consumer_tag="second", queue="test-queue-2")
+ self.fail("Expected consume request to fail due to previous exclusive consumer")
+ except Closed, e:
+ self.assertChannelException(403, e.args[0])
+
+ #open new channel and cleanup last consumer:
+ channel = self.client.channel(2)
+ channel.channel_open()
+
+ #check that an exclusive consumer cannot be created if a consumer already exists:
+ channel.basic_consume(consumer_tag="first", queue="test-queue-2")
+ try:
+ channel.basic_consume(consumer_tag="second", queue="test-queue-2", exclusive=True)
+ self.fail("Expected exclusive consume request to fail due to previous consumer")
+ except Closed, e:
+ self.assertChannelException(403, e.args[0])
+
+ def test_consume_queue_errors(self):
+ """
+ Test error conditions associated with the queue field of the consume method:
+ """
+ channel = self.channel
+ try:
+ #queue specified but doesn't exist:
+ channel.basic_consume(queue="invalid-queue")
+ self.fail("Expected failure when consuming from non-existent queue")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+ channel = self.client.channel(2)
+ channel.channel_open()
+ try:
+ #queue not specified and none previously declared for channel:
+ channel.basic_consume(queue="")
+ self.fail("Expected failure when consuming from unspecified queue")
+ except Closed, e:
+ self.assertConnectionException(530, e.args[0])
+
+ def test_consume_unique_consumers(self):
+ """
+ Ensure unique consumer tags are enforced
+ """
+ channel = self.channel
+ #setup, declare a queue:
+ channel.queue_declare(queue="test-queue-3", exclusive=True)
+
+ #check that attempts to use duplicate tags are detected and prevented:
+ channel.basic_consume(consumer_tag="first", queue="test-queue-3")
+ try:
+ channel.basic_consume(consumer_tag="first", queue="test-queue-3")
+ self.fail("Expected consume request to fail due to non-unique tag")
+ except Closed, e:
+ self.assertConnectionException(530, e.args[0])
+
Propchange: incubator/qpid/trunk/qpid/python/tests/basic.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/tests/broker.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/broker.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/broker.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/broker.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,84 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+from qpid.queue import Empty
+from qpid.content import Content
+from qpid.testlib import testrunner, TestBase
+
+class BrokerTests(TestBase):
+ """Tests for basic Broker functionality"""
+
+ def test_amqp_basic_13(self):
+ """
+ First, this test tries to receive a message with a no-ack
+ consumer. Second, this test tries to explicitely receive and
+ acknowledge a message with an acknowledging consumer.
+ """
+ ch = self.channel
+ self.queue_declare(ch, queue = "myqueue")
+
+ # No ack consumer
+ ctag = ch.basic_consume(queue = "myqueue", no_ack = True).consumer_tag
+ body = "test no-ack"
+ ch.basic_publish(routing_key = "myqueue", content = Content(body))
+ msg = self.client.queue(ctag).get(timeout = 5)
+ self.assert_(msg.content.body == body)
+
+ # Acknowleding consumer
+ self.queue_declare(ch, queue = "otherqueue")
+ ctag = ch.basic_consume(queue = "otherqueue", no_ack = False).consumer_tag
+ body = "test ack"
+ ch.basic_publish(routing_key = "otherqueue", content = Content(body))
+ msg = self.client.queue(ctag).get(timeout = 5)
+ ch.basic_ack(delivery_tag = msg.delivery_tag)
+ self.assert_(msg.content.body == body)
+
+ # TODO: Ensure we get a failure if an ack consumer doesn't ack.
+
+ def test_basic_delivery_immediate(self):
+ """
+ Test basic message delivery where consume is issued before publish
+ """
+ channel = self.channel
+ self.exchange_declare(channel, exchange="test-exchange", type="direct")
+ self.queue_declare(channel, queue="test-queue")
+ channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
+ reply = channel.basic_consume(queue="test-queue", no_ack=True)
+ queue = self.client.queue(reply.consumer_tag)
+
+ body = "Immediate Delivery"
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content(body), immediate=True)
+ msg = queue.get(timeout=5)
+ self.assert_(msg.content.body == body)
+
+ # TODO: Ensure we fail if immediate=True and there's no consumer.
+
+
+ def test_basic_delivery_queued(self):
+ """
+ Test basic message delivery where publish is issued before consume
+ (i.e. requires queueing of the message)
+ """
+ channel = self.channel
+ self.exchange_declare(channel, exchange="test-exchange", type="direct")
+ self.queue_declare(channel, queue="test-queue")
+ channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
+ body = "Queued Delivery"
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content(body))
+ reply = channel.basic_consume(queue="test-queue", no_ack=True)
+ queue = self.client.queue(reply.consumer_tag)
+ msg = queue.get(timeout=5)
+ self.assert_(msg.content.body == body)
+
Propchange: incubator/qpid/trunk/qpid/python/tests/broker.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/tests/example.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/example.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/example.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/example.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,91 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+from qpid.content import Content
+from qpid.testlib import testrunner, TestBase
+
+class ExampleTest (TestBase):
+ """
+ An example Qpid test, illustrating the unittest frameowkr and the
+ python Qpid client. The test class must inherit TestCase. The
+ test code uses the Qpid client to interact with a qpid broker and
+ verify it behaves as expected.
+ """
+
+ def test_example(self):
+ """
+ An example test. Note that test functions must start with 'test_'
+ to be recognized by the test framework.
+ """
+
+ # By inheriting TestBase, self.client is automatically connected
+ # and self.channel is automatically opened as channel(1)
+ # Other channel methods mimic the protocol.
+ channel = self.channel
+
+ # Now we can send regular commands. If you want to see what the method
+ # arguments mean or what other commands are available, you can use the
+ # python builtin help() method. For example:
+ #help(chan)
+ #help(chan.exchange_declare)
+
+ # If you want browse the available protocol methods without being
+ # connected to a live server you can use the amqp-doc utility:
+ #
+ # Usage amqp-doc [<options>] <spec> [<pattern_1> ... <pattern_n>]
+ #
+ # Options:
+ # -e, --regexp use regex instead of glob when matching
+
+ # Now that we know what commands are available we can use them to
+ # interact with the server.
+
+ # Here we use ordinal arguments.
+ self.exchange_declare(channel, 0, "test", "direct")
+
+ # Here we use keyword arguments.
+ self.queue_declare(channel, queue="test-queue")
+ channel.queue_bind(queue="test-queue", exchange="test", routing_key="key")
+
+ # Call Channel.basic_consume to register as a consumer.
+ # All the protocol methods return a message object. The message object
+ # has fields corresponding to the reply method fields, plus a content
+ # field that is filled if the reply includes content. In this case the
+ # interesting field is the consumer_tag.
+ reply = channel.basic_consume(queue="test-queue")
+
+ # We can use the Client.queue(...) method to access the queue
+ # corresponding to our consumer_tag.
+ queue = self.client.queue(reply.consumer_tag)
+
+ # Now lets publish a message and see if our consumer gets it. To do
+ # this we need to import the Content class.
+ body = "Hello World!"
+ channel.basic_publish(exchange="test",
+ routing_key="key",
+ content=Content(body))
+
+ # Now we'll wait for the message to arrive. We can use the timeout
+ # argument in case the server hangs. By default queue.get() will wait
+ # until a message arrives or the connection to the server dies.
+ msg = queue.get(timeout=10)
+
+ # And check that we got the right response with assertEqual
+ self.assertEqual(body, msg.content.body)
+
+ # Now acknowledge the message.
+ channel.basic_ack(msg.delivery_tag, True)
+
Propchange: incubator/qpid/trunk/qpid/python/tests/example.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/tests/exchange.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/exchange.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/exchange.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/exchange.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,234 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+"""
+Tests for exchange behaviour.
+
+Test classes ending in 'RuleTests' are derived from rules in amqp.xml.
+"""
+
+import logging, Queue
+from qpid.testlib import TestBase
+from qpid.content import Content
+
+
+# TODO aconway 2006-09-01: Investigate and add tests as appropriate.
+# Observered on C++:
+#
+# No exception raised for basic_consume on non-existent queue name.
+# No exception for basic_publish with bad routing key.
+# No exception for binding to non-existent exchange?
+# queue_bind hangs with invalid exchange name
+#
+# Do server exceptions get propagated properly?
+# Do Java exceptions propagate with any data (or just Closed())
+
+class StandardExchangeVerifier:
+ """Verifies standard exchange behavior.
+
+ Used as base class for classes that test standard exchanges."""
+
+ def verifyDirectExchange(self, ex):
+ """Verify that ex behaves like a direct exchange."""
+ self.queue_declare(queue="q")
+ self.channel.queue_bind(queue="q", exchange=ex, routing_key="k")
+ self.assertPublishConsume(exchange=ex, queue="q", routing_key="k")
+ try:
+ self.assertPublishConsume(exchange=ex, queue="q", routing_key="kk")
+ self.fail("Expected Empty exception")
+ except Queue.Empty: None # Expected
+
+ def verifyFanOutExchange(self, ex):
+ """Verify that ex behaves like a fanout exchange."""
+ self.queue_declare(queue="q")
+ self.channel.queue_bind(queue="q", exchange=ex)
+ self.queue_declare(queue="p")
+ self.channel.queue_bind(queue="p", exchange=ex)
+ self.assertPublishConsume(exchange=ex, queue=["q","p"])
+
+
+class RecommendedTypesRuleTests(TestBase, StandardExchangeVerifier):
+ """
+ The server SHOULD implement these standard exchange types: topic, headers.
+
+ Client attempts to declare an exchange with each of these standard types.
+ """
+
+ def testDirect(self):
+ """Declare and test a direct exchange"""
+ self.exchange_declare(0, exchange="d", type="direct")
+ self.verifyDirectExchange("d")
+
+ def testFanout(self):
+ """Declare and test a fanout exchange"""
+ self.exchange_declare(0, exchange="f", type="fanout")
+ self.verifyFanOutExchange("f")
+
+
+class RequiredInstancesRuleTests(TestBase, StandardExchangeVerifier):
+ """
+ The server MUST, in each virtual host, pre-declare an exchange instance
+ for each standard exchange type that it implements, where the name of the
+ exchange instance is amq. followed by the exchange type name.
+
+ Client creates a temporary queue and attempts to bind to each required
+ exchange instance (amq.fanout, amq.direct, and amq.topic, amq.headers if
+ those types are defined).
+ """
+ # TODO aconway 2006-09-01: Add tests for 3.1.3.1:
+ # - Test auto binding by q name
+ # - Test the nameless "default publish" exchange.
+ # - Auto created amq.fanout exchange
+
+ def testAmqDirect(self): self.verifyDirectExchange("amq.direct")
+
+ def testAmqFanOut(self): self.verifyFanOutExchange("amq.fanout")
+
+ def testAmqTopic(self):
+ self.exchange_declare(0, exchange="amq.topic", passive="true")
+ # TODO aconway 2006-09-14: verify topic behavior
+
+ def testAmqHeaders(self):
+ self.exchange_declare(0, exchange="amq.headers", passive="true")
+ # TODO aconway 2006-09-14: verify headers behavior
+
+class DefaultExchangeRuleTests(TestBase):
+ """
+ The server MUST predeclare a direct exchange to act as the default exchange
+ for content Publish methods and for default queue bindings.
+
+ Client checks that the default exchange is active by specifying a queue
+ binding with no exchange name, and publishing a message with a suitable
+ routing key but without specifying the exchange name, then ensuring that
+ the message arrives in the queue correctly.
+ """
+
+
+class DefaultAccessRuleTests(TestBase):
+ """
+ The server MUST NOT allow clients to access the default exchange except
+ by specifying an empty exchange name in the Queue.Bind and content Publish
+ methods.
+ """
+
+
+class ExtensionsRuleTests(TestBase):
+ """
+ The server MAY implement other exchange types as wanted.
+ """
+
+
+class DeclareMethodMinimumRuleTests(TestBase):
+ """
+ The server SHOULD support a minimum of 16 exchanges per virtual host and
+ ideally, impose no limit except as defined by available resources.
+
+ The client creates as many exchanges as it can until the server reports
+ an error; the number of exchanges successfuly created must be at least
+ sixteen.
+ """
+
+
+class DeclareMethodTicketFieldValidityRuleTests(TestBase):
+ """
+ The client MUST provide a valid access ticket giving "active" access to
+ the realm in which the exchange exists or will be created, or "passive"
+ access if the if-exists flag is set.
+
+ Client creates access ticket with wrong access rights and attempts to use
+ in this method.
+ """
+
+
+class DeclareMethodExchangeFieldReservedRuleTests(TestBase):
+ """
+ Exchange names starting with "amq." are reserved for predeclared and
+ standardised exchanges. The client MUST NOT attempt to create an exchange
+ starting with "amq.".
+
+
+ """
+
+
+class DeclareMethodTypeFieldTypedRuleTests(TestBase):
+ """
+ Exchanges cannot be redeclared with different types. The client MUST not
+ attempt to redeclare an existing exchange with a different type than used
+ in the original Exchange.Declare method.
+
+
+ """
+
+
+class DeclareMethodTypeFieldSupportRuleTests(TestBase):
+ """
+ The client MUST NOT attempt to create an exchange with a type that the
+ server does not support.
+
+
+ """
+
+
+class DeclareMethodPassiveFieldNotFoundRuleTests(TestBase):
+ """
+ If set, and the exchange does not already exist, the server MUST raise a
+ channel exception with reply code 404 (not found).
+
+
+ """
+
+
+class DeclareMethodDurableFieldSupportRuleTests(TestBase):
+ """
+ The server MUST support both durable and transient exchanges.
+
+
+ """
+
+
+class DeclareMethodDurableFieldStickyRuleTests(TestBase):
+ """
+ The server MUST ignore the durable field if the exchange already exists.
+
+
+ """
+
+
+class DeclareMethodAutoDeleteFieldStickyRuleTests(TestBase):
+ """
+ The server MUST ignore the auto-delete field if the exchange already
+ exists.
+
+
+ """
+
+
+class DeleteMethodTicketFieldValidityRuleTests(TestBase):
+ """
+ The client MUST provide a valid access ticket giving "active" access
+ rights to the exchange's access realm.
+
+ Client creates access ticket with wrong access rights and attempts to use
+ in this method.
+ """
+
+
+class DeleteMethodExchangeFieldExistsRuleTests(TestBase):
+ """
+ The client MUST NOT attempt to delete an exchange that does not exist.
+ """
+
+
Propchange: incubator/qpid/trunk/qpid/python/tests/exchange.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/python/tests/queue.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests/queue.py?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests/queue.py (added)
+++ incubator/qpid/trunk/qpid/python/tests/queue.py Tue Sep 19 15:06:50 2006
@@ -0,0 +1,254 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+from qpid.client import Client, Closed
+from qpid.queue import Empty
+from qpid.content import Content
+from qpid.testlib import testrunner, TestBase
+
+class QueueTests(TestBase):
+ """Tests for 'methods' on the amqp queue 'class'"""
+
+ def test_purge(self):
+ """
+ Test that the purge method removes messages from the queue
+ """
+ channel = self.channel
+ #setup, declare a queue and add some messages to it:
+ channel.exchange_declare(exchange="test-exchange", type="direct")
+ channel.queue_declare(queue="test-queue", exclusive=True)
+ channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content("one"))
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content("two"))
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content("three"))
+
+ #check that the queue now reports 3 messages:
+ reply = channel.queue_declare(queue="test-queue")
+ self.assertEqual(3, reply.message_count)
+
+ #now do the purge, then test that three messages are purged and the count drops to 0
+ reply = channel.queue_purge(queue="test-queue");
+ self.assertEqual(3, reply.message_count)
+ reply = channel.queue_declare(queue="test-queue")
+ self.assertEqual(0, reply.message_count)
+
+ #send a further message and consume it, ensuring that the other messages are really gone
+ channel.basic_publish(exchange="test-exchange", routing_key="key", content=Content("four"))
+ reply = channel.basic_consume(queue="test-queue", no_ack=True)
+ queue = self.client.queue(reply.consumer_tag)
+ msg = queue.get(timeout=1)
+ self.assertEqual("four", msg.content.body)
+
+ #check error conditions (use new channels):
+ channel = self.client.channel(2)
+ channel.channel_open()
+ try:
+ #queue specified but doesn't exist:
+ channel.queue_purge(queue="invalid-queue")
+ self.fail("Expected failure when purging non-existent queue")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+ channel = self.client.channel(3)
+ channel.channel_open()
+ try:
+ #queue not specified and none previously declared for channel:
+ channel.queue_purge()
+ self.fail("Expected failure when purging unspecified queue")
+ except Closed, e:
+ self.assertConnectionException(530, e.args[0])
+
+ #cleanup
+ channel = self.client.channel(4)
+ channel.channel_open()
+ channel.exchange_delete(exchange="test-exchange")
+
+ def test_declare_exclusive(self):
+ """
+ Test that the exclusive field is honoured in queue.declare
+ """
+ # TestBase.setUp has already opened channel(1)
+ c1 = self.channel
+ # Here we open a second separate connection:
+ other = self.connect()
+ c2 = other.channel(1)
+ c2.channel_open()
+
+ #declare an exclusive queue:
+ c1.queue_declare(queue="exclusive-queue", exclusive="True")
+ try:
+ #other connection should not be allowed to declare this:
+ c2.queue_declare(queue="exclusive-queue", exclusive="True")
+ self.fail("Expected second exclusive queue_declare to raise a channel exception")
+ except Closed, e:
+ self.assertChannelException(405, e.args[0])
+
+
+ def test_declare_passive(self):
+ """
+ Test that the passive field is honoured in queue.declare
+ """
+ channel = self.channel
+ #declare an exclusive queue:
+ channel.queue_declare(queue="passive-queue-1", exclusive="True")
+ channel.queue_declare(queue="passive-queue-1", passive="True")
+ try:
+ #other connection should not be allowed to declare this:
+ channel.queue_declare(queue="passive-queue-2", passive="True")
+ self.fail("Expected passive declaration of non-existant queue to raise a channel exception")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+
+ def test_bind(self):
+ """
+ Test various permutations of the queue.bind method
+ """
+ channel = self.channel
+ channel.queue_declare(queue="queue-1", exclusive="True")
+
+ #straightforward case, both exchange & queue exist so no errors expected:
+ channel.queue_bind(queue="queue-1", exchange="amq.direct", routing_key="key1")
+
+ #bind the default queue for the channel (i.e. last one declared):
+ channel.queue_bind(exchange="amq.direct", routing_key="key2")
+
+ #use the queue name where neither routing key nor queue are specified:
+ channel.queue_bind(exchange="amq.direct")
+
+ #try and bind to non-existant exchange
+ try:
+ channel.queue_bind(queue="queue-1", exchange="an-invalid-exchange", routing_key="key1")
+ self.fail("Expected bind to non-existant exchange to fail")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+ #need to reopen a channel:
+ channel = self.client.channel(2)
+ channel.channel_open()
+
+ #try and bind non-existant queue:
+ try:
+ channel.queue_bind(queue="queue-2", exchange="amq.direct", routing_key="key1")
+ self.fail("Expected bind of non-existant queue to fail")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+
+ def test_delete_simple(self):
+ """
+ Test basic queue deletion
+ """
+ channel = self.client.channel(1)
+ channel.channel_open()
+
+ #straight-forward case:
+ channel.queue_declare(queue="delete-me")
+ channel.basic_publish(exchange="amq.direct", routing_key="delete-me", content=Content("a"))
+ channel.basic_publish(exchange="amq.direct", routing_key="delete-me", content=Content("b"))
+ channel.basic_publish(exchange="amq.direct", routing_key="delete-me", content=Content("c"))
+ reply = channel.queue_delete(queue="delete-me")
+ self.assertEqual(3, reply.message_count)
+ #check that it has gone be declaring passively
+ try:
+ channel.queue_declare(queue="delete-me", passive="True")
+ self.fail("Queue has not been deleted")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+ #check attempted deletion of non-existant queue is handled correctly:
+ channel = self.client.channel(2)
+ channel.channel_open()
+ try:
+ channel.queue_delete(queue="i-dont-exist", if_empty="True")
+ self.fail("Expected delete of non-existant queue to fail")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+
+
+ def test_delete_ifempty(self):
+ """
+ Test that if_empty field of queue_delete is honoured
+ """
+ channel = self.client.channel(1)
+ channel.channel_open()
+
+ #create a queue and add a message to it (use default binding):
+ channel.queue_declare(queue="delete-me-2")
+ channel.queue_declare(queue="delete-me-2", passive="True")
+ channel.basic_publish(exchange="amq.direct", routing_key="delete-me-2", content=Content("message"))
+
+ #try to delete, but only if empty:
+ try:
+ channel.queue_delete(queue="delete-me-2", if_empty="True")
+ self.fail("Expected delete if_empty to fail for non-empty queue")
+ except Closed, e:
+ self.assertChannelException(406, e.args[0])
+
+ #need new channel now:
+ channel = self.client.channel(2)
+ channel.channel_open()
+
+ #empty queue:
+ reply = channel.basic_consume(queue="delete-me-2", no_ack=True)
+ queue = self.client.queue(reply.consumer_tag)
+ msg = queue.get(timeout=1)
+ self.assertEqual("message", msg.content.body)
+ channel.basic_cancel(consumer_tag=reply.consumer_tag)
+
+ #retry deletion on empty queue:
+ channel.queue_delete(queue="delete-me-2", if_empty="True")
+
+ #check that it has gone by declaring passively:
+ try:
+ channel.queue_declare(queue="delete-me-2", passive="True")
+ self.fail("Queue has not been deleted")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+ def test_delete_ifunused(self):
+ """
+ Test that if_unused field of queue_delete is honoured
+ """
+ channel = self.client.channel(1)
+ channel.channel_open()
+
+ #create a queue and register a consumer:
+ channel.queue_declare(queue="delete-me-3")
+ channel.queue_declare(queue="delete-me-3", passive="True")
+ reply = channel.basic_consume(queue="delete-me-3", no_ack=True)
+
+ #need new channel now:
+ channel2 = self.client.channel(2)
+ channel2.channel_open()
+ #try to delete, but only if empty:
+ try:
+ channel2.queue_delete(queue="delete-me-3", if_unused="True")
+ self.fail("Expected delete if_unused to fail for queue with existing consumer")
+ except Closed, e:
+ self.assertChannelException(406, e.args[0])
+
+
+ channel.basic_cancel(consumer_tag=reply.consumer_tag)
+ channel.queue_delete(queue="delete-me-3", if_unused="True")
+ #check that it has gone by declaring passively:
+ try:
+ channel.queue_declare(queue="delete-me-3", passive="True")
+ self.fail("Queue has not been deleted")
+ except Closed, e:
+ self.assertChannelException(404, e.args[0])
+
+
Propchange: incubator/qpid/trunk/qpid/python/tests/queue.py
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/qpid/trunk/qpid/ruby/client.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/client.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/client.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/client.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,106 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+require "peer"
+require "thread"
+require "queue"
+
+module Qpid
+
+ class Client
+ def initialize(host, port, spec, vhost = nil)
+ @host = host
+ @port = port
+ @spec = spec
+ @vhost = if vhost.nil?; host else vhost end
+
+ @mechanism = nil
+ @response = nil
+ @locale = nil
+
+ @queues = {}
+ @mutex = Mutex.new()
+
+ @closed = false
+ @started = ConditionVariable.new()
+
+ @conn = Connection.new(@host, @port, @spec)
+ @peer = Peer.new(@conn, ClientDelegate.new(self))
+ end
+
+ attr_reader :mechanism, :response, :locale
+
+ def closed?; @closed end
+
+ def wait()
+ @mutex.synchronize do
+ @started.wait(@mutex)
+ end
+ raise EOFError.new() if closed?
+ end
+
+ def signal_start()
+ @started.broadcast()
+ end
+
+ def queue(key)
+ @mutex.synchronize do
+ q = @queues[key]
+ if q.nil?
+ q = Queue.new()
+ @queues[key] = q
+ end
+ return q
+ end
+ end
+
+ def start(response, mechanism="AMQPLAIN", locale="en_US")
+ @response = response
+ @mechanism = mechanism
+ @locale = locale
+
+ @conn.connect()
+ @conn.init()
+ @peer.start()
+ wait()
+ channel(0).connection_open(@vhost)
+ end
+
+ def channel(id)
+ return @peer.channel(id)
+ end
+ end
+
+ class ClientDelegate
+ include Delegate
+
+ def initialize(client)
+ @client = client
+ end
+
+ def connection_start(ch, msg)
+ ch.connection_start_ok(:mechanism => @client.mechanism,
+ :response => @client.response,
+ :locale => @client.locale)
+ end
+
+ def connection_tune(ch, msg)
+ ch.connection_tune_ok(*msg.fields)
+ @client.signal_start()
+ end
+ end
+
+end
Added: incubator/qpid/trunk/qpid/ruby/codec.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/codec.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/codec.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/codec.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,253 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+module Codec
+ # is there a better way to do this?
+ class StringWriter
+
+ def initialize(str = "")
+ @str = str
+ end
+
+ def write(value)
+ @str << value
+ end
+
+ def to_s()
+ return @str
+ end
+
+ end
+
+ class EOF < Exception; end
+
+ class Encoder
+
+ def initialize(out)
+ @out = out
+ @bits = []
+ end
+
+ attr_reader(:out)
+
+ def encode(type, value)
+ send(type, value)
+ end
+
+ def bit(b)
+ @bits << b
+ end
+
+ def octet(o)
+ pack("C", o)
+ end
+
+ def short(s)
+ pack("n", s)
+ end
+
+ def long(l)
+ pack("N", l)
+ end
+
+ def longlong(l)
+ # is this the right byte order?
+ lower = l & 0xffffffff
+ upper = (l & ~0xffffffff) >> 32
+ long(lower)
+ long(upper)
+ end
+
+ def shortstr(s)
+ # shortstr is actually octetstr
+ octet(s.length)
+ write(s)
+ end
+
+ def longstr(s)
+ case s
+ when Hash
+ table(s)
+ else
+ long(s.length)
+ write(s)
+ end
+ end
+
+ def table(t)
+ t = {} if t.nil?
+ enc = Encoder.new(StringWriter.new())
+ t.each {|key, value|
+ enc.shortstr(key)
+ # I offer this chicken to the gods of polymorphism. May they
+ # choke on it.
+ case value
+ when String
+ type = :longstr
+ desc = "S"
+ when Numeric
+ type = :long
+ desc = "I"
+ else
+ raise Exception.new("unknown table value: #{value.class}")
+ end
+ enc.write(desc)
+ enc.encode(type, value)
+ }
+ longstr(enc.out.to_s())
+ end
+
+ def write(str)
+ flushbits()
+ @out.write(str)
+ end
+
+ def pack(fmt, *args)
+ write(args.pack(fmt))
+ end
+
+ def flush()
+ flushbits()
+ end
+
+ private
+
+ def flushbits()
+ if @bits.empty? then return end
+
+ bytes = []
+ index = 0
+ @bits.each {|b|
+ bytes << 0 if index == 0
+ if b then bytes[-1] |= 1 << index end
+ index = (index + 1) % 8
+ }
+ @bits.clear()
+ bytes.each {|b|
+ octet(b)
+ }
+ end
+
+ end
+
+ class StringReader
+
+ def initialize(str)
+ @str = str
+ @index = 0
+ end
+
+ def read(n)
+ result = @str[@index, n]
+ @index += result.length
+ return result
+ end
+
+ end
+
+ class Decoder
+
+ def initialize(_in)
+ @in = _in
+ @bits = []
+ end
+
+ def decode(type)
+ return send(type)
+ end
+
+ def bit()
+ if @bits.empty?
+ byte = octet()
+ 7.downto(0) {|i|
+ @bits << (byte[i] == 1)
+ }
+ end
+ return @bits.pop()
+ end
+
+ def octet()
+ return unpack("C", 1)
+ end
+
+ def short()
+ return unpack("n", 2)
+ end
+
+ def long()
+ return unpack("N", 4)
+ end
+
+ def longlong()
+ upper = long()
+ lower = long()
+ return upper << 32 | lower
+ end
+
+ def shortstr()
+ # shortstr is actually octetstr
+ return read(octet())
+ end
+
+ def longstr()
+ return read(long())
+ end
+
+ def table()
+ dec = Decoder.new(StringReader.new(longstr()))
+ result = {}
+ while true
+ begin
+ key = dec.shortstr()
+ rescue EOF
+ break
+ end
+ desc = dec.read(1)
+ case desc
+ when "S"
+ value = dec.longstr()
+ when "I"
+ value = dec.long()
+ else
+ raise Exception.new("unrecognized descriminator: #{desc.inspect()}")
+ end
+ result[key] = value
+ end
+ return result
+ end
+
+ def read(n)
+ return "" if n == 0
+ result = @in.read(n)
+ if result.nil? or result.empty?
+ raise EOF.new()
+ else
+ return result
+ end
+ end
+
+ def unpack(fmt, size)
+ result = read(size).unpack(fmt)
+ if result.length == 1
+ return result[0]
+ else
+ return result
+ end
+ end
+
+ end
+
+end
Added: incubator/qpid/trunk/qpid/ruby/connection.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/connection.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/connection.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/connection.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,142 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+require "socket"
+require "codec"
+
+include Codec
+
+module Qpid
+
+ class Connection
+
+ def initialize(host, port, spec)
+ @host = host
+ @port = port
+ @spec = spec
+ end
+
+ attr_reader(:host, :port, :spec)
+
+ def connect()
+ @sock = TCPSocket.open(@host, @port)
+ @out = Encoder.new(@sock)
+ @in = Decoder.new(@sock)
+ end
+
+ def init()
+ @out.write("AMQP")
+ [1, 1, @spec.major, @spec.minor].each {|o|
+ @out.octet(o)
+ }
+ end
+
+ def write(frame)
+ @out.octet(@spec.constants[frame.payload.type].id)
+ @out.short(frame.channel)
+ frame.payload.encode(@out)
+ @out.octet(frame_end)
+ end
+
+ def read()
+ type = @spec.constants[@in.octet()].name
+ channel = @in.short()
+ payload = Payload.decode(type, @spec, @in)
+ oct = @in.octet()
+ if oct != frame_end
+ raise Exception.new("framing error: expected #{frame_end}, got #{oct}")
+ end
+ Frame.new(channel, payload)
+ end
+
+ private
+
+ def frame_end
+ @spec.constants[:"frame end"].id
+ end
+
+ end
+
+ class Frame
+
+ def initialize(channel, payload)
+ @channel = channel
+ @payload = payload
+ end
+
+ attr_reader(:channel, :payload)
+
+ end
+
+ class Payload
+
+ TYPES = {}
+
+ def Payload.singleton_method_added(name)
+ if name == :type
+ TYPES[type] = self
+ end
+ end
+
+ def Payload.decode(type, spec, dec)
+ klass = TYPES[type]
+ klass.decode(spec, dec)
+ end
+
+ end
+
+ class Method < Payload
+
+ def initialize(method, args)
+ if args.size != method.fields.size
+ raise ArgumentError.new("argument mismatch #{method} #{args}")
+ end
+ @method = method
+ @args = args
+ end
+
+ attr_reader(:method, :args)
+
+ def Method.type
+ :"frame method"
+ end
+
+ def type; Method.type end
+
+ def encode(encoder)
+ buf = StringWriter.new()
+ enc = Encoder.new(buf)
+ enc.short(@method.parent.id)
+ enc.short(@method.id)
+ @method.fields.zip(self.args).each {|f, a|
+ enc.encode(f.type, a)
+ }
+ enc.flush()
+ encoder.longstr(buf.to_s)
+ end
+
+ def Method.decode(spec, decoder)
+ buf = decoder.longstr()
+ dec = Decoder.new(StringReader.new(buf))
+ klass = spec.classes[dec.short()]
+ meth = klass.methods[dec.short()]
+ args = meth.fields.map {|f| dec.decode(f.type)}
+ return Method.new(meth, args)
+ end
+
+ end
+
+end
Added: incubator/qpid/trunk/qpid/ruby/diff.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/diff.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/diff.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/diff.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,57 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+require "spec"
+
+spec = Spec.load(File.new($*[0]))
+
+class Symbol
+ def <=>(sym)
+ to_s <=> sym.to_s
+ end
+end
+
+def diff(classes)
+ sets = classes.map {|c| yield(c).to_set}
+ common = sets[0]
+ sets[1..-1].each {|s|
+ common = common & s
+ }
+
+ sep = "\n "
+
+ puts "Common:\n #{common.to_a.sort.join(sep)}"
+ classes.zip(sets).each {|c, s|
+ specific = (s - common).to_a.sort
+ puts "\n#{c.name}:\n #{specific.join(sep)}"
+ }
+end
+
+classes = $*[1..-1].map {|c|
+ spec.classes[c]
+}
+
+diff(classes) {|c|
+ result = []
+ c.methods.each {|m| m.fields.each {|f| result << :"#{m.name}.#{f.name}"}}
+ result
+}
+
+puts
+
+diff(classes) {|c|
+ c.fields.map {|f| f.name}
+}
Added: incubator/qpid/trunk/qpid/ruby/fields.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/fields.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/fields.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/fields.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,46 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+class Class
+ def fields(*fields)
+ module_eval {
+ def initialize(*args, &block)
+ args = init_fields(*args)
+
+ if respond_to? :init
+ init(*args) {|*a| yield(*a)}
+ elsif args.any?
+ raise ArgumentException.new("extra arguments: #{args}")
+ end
+ end
+ }
+
+ vars = fields.map {|f| :"@#{f.to_s().chomp("?")}"}
+
+ define_method(:init_fields) {|*args|
+ vars.each {|v|
+ instance_variable_set(v, args.shift())
+ }
+ args
+ }
+
+ vars.each_index {|i|
+ define_method(fields[i]) {
+ instance_variable_get(vars[i])
+ }
+ }
+ end
+end
Added: incubator/qpid/trunk/qpid/ruby/peer.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/peer.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/peer.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/peer.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,246 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+require "thread"
+require "queue"
+require "connection"
+require "fields"
+
+module Qpid
+
+ class Peer
+
+ def initialize(conn, delegate)
+ @conn = conn
+ @delegate = delegate
+ @outgoing = Queue.new()
+ @work = Queue.new()
+ @channels = {}
+ @mutex = Mutex.new()
+ end
+
+ def channel(id)
+ @mutex.synchronize do
+ ch = @channels[id]
+ if ch.nil?
+ ch = Channel.new(id, @outgoing, @conn.spec)
+ @channels[id] = ch
+ end
+ return ch
+ end
+ end
+
+ def start()
+ spawn(:writer)
+ spawn(:reader)
+ spawn(:worker)
+ end
+
+ private
+
+ def spawn(method, *args)
+ Thread.new do
+ begin
+ send(method, *args)
+ # is this the standard way to catch any exception?
+ rescue Object => e
+ print e
+ e.backtrace.each do |line|
+ print "\n ", line
+ end
+ print "\n"
+ end
+ end
+ end
+
+ def reader()
+ while true
+ frame = @conn.read()
+ ch = channel(frame.channel)
+ ch.dispatch(frame, @work)
+ end
+ end
+
+ def writer()
+ while true
+ @conn.write(@outgoing.pop())
+ end
+ end
+
+ def worker()
+ while true
+ dispatch(@work.pop())
+ end
+ end
+
+ def dispatch(queue)
+ frame = queue.pop()
+ ch = channel(frame.channel)
+ payload = frame.payload
+ if payload.method.content?
+ content = read_content(queue)
+ else
+ content = nil
+ end
+
+ message = Message.new(payload.method, payload.args, content)
+ @delegate.dispatch(ch, message)
+ end
+
+ end
+
+ class Channel
+ def initialize(id, outgoing, spec)
+ @id = id
+ @outgoing = outgoing
+ @spec = spec
+ @incoming = Queue.new()
+ @responses = Queue.new()
+ @queue = nil
+ @closed = false
+ @reason = nil
+ end
+
+ def closed?; @closed end
+
+ def close(reason)
+ return if closed?
+ @closed = true
+ @reason = reason
+ @incoming.close()
+ @responses .close()
+ end
+
+ def dispatch(frame, work)
+ payload = frame.payload
+ case payload
+ when Method
+ if payload.method.response?
+ @queue = @responses
+ else
+ @queue = @incoming
+ work << @incoming
+ end
+ end
+ @queue << frame
+ end
+
+ def method_missing(name, *args)
+ method = @spec.ruby_method(name)
+ if method.nil?
+ raise NoMethodError.new("undefined method '#{name}' for #{self}:#{self.class}")
+ end
+
+ if args.size == 1 and args[0].instance_of? Hash
+ kwargs = args[0]
+ invoke_args = method.fields.map do |f|
+ kwargs[f.ruby_name]
+ end
+ content = kwargs[:content]
+ else
+ invoke_args = []
+ method.fields.each do |f|
+ if args.any?
+ invoke_args << args.shift()
+ else
+ invoke_args << f.default
+ end
+ end
+ if method.content? and args.any?
+ content = args.shift()
+ else
+ content = nil
+ end
+ if args.any? then raise ArgumentError.new("#{args.size} extr arguments") end
+ end
+ return invoke(method, invoke_args, content)
+ end
+
+ def invoke(method, args, content = nil)
+ raise Closed(@reason) if closed?
+ frame = Frame.new(@id, Method.new(method, args))
+ @outgoing << frame
+
+ if method.content?
+ content = Content.new() if content.nil?
+ write_content(method.klass, content, @outgoing)
+ end
+
+ nowait = false
+ f = method.fields[:"nowait"]
+ nowait = args[method.fields.index(f)] unless f.nil?
+
+ unless nowait or method.responses.empty?
+ resp = @responses.pop().payload
+ if resp.method.content?
+ content = read_content(@responses)
+ else
+ content = nil
+ end
+ if method.responses.include? resp.method
+ return Message.new(resp.method, resp.args, content)
+ else
+ # XXX: ValueError doesn't actually exist
+ raise ValueError.new(resp)
+ end
+ end
+ end
+
+ def write_content(klass, content, queue)
+ size = content.size
+ header = Frame.new(@id, Header.new(klass, content.weight, size))
+ queue << header
+ content.children.each {|child| write_content(klass, child, queue)}
+ queue << Frame.new(@id, Body.new(content.body)) if size > 0
+ end
+
+ end
+
+ def read_content(queue)
+ frame = queue.pop()
+ header = frame.payload
+ children = []
+ 1.upto(header.weight) { children << read_content(queue) }
+ size = header.size
+ read = 0
+ buf = ""
+ while read << size
+ body = queue.get()
+ content = body.payload.content
+ buf << content
+ read += content.size
+ end
+ buf.freeze()
+ return Content.new(buf, children, header.properties.clone())
+ end
+
+ class Message
+ fields(:method, :args, :content)
+
+ alias fields args
+
+ def inspect()
+ "#{method.ruby_name}(#{args.join(", ")})"
+ end
+ end
+
+ module Delegate
+ def dispatch(ch, msg)
+ send(msg.method.ruby_name, ch, msg)
+ end
+ end
+
+end
Added: incubator/qpid/trunk/qpid/ruby/queue.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/ruby/queue.rb?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/ruby/queue.rb (added)
+++ incubator/qpid/trunk/qpid/ruby/queue.rb Tue Sep 19 15:06:50 2006
@@ -0,0 +1,49 @@
+#
+# Copyright (c) 2006 The Apache Software Foundation
+#
+# Licensed 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.
+#
+
+require "thread"
+
+module Qpid
+
+ class Closed < Exception; end
+
+ class Queue < Queue
+
+ @@END = Object.new()
+
+ def close()
+ # sentinal to indicate the end of the queue
+ self << @@END
+ end
+
+ def pop(*args)
+ result = super(*args)
+ if @@END.equal? result
+ # we put another sentinal on the end in case there are
+ # subsequent calls to pop by this or other threads
+ self << @@END
+ raise Closed.new()
+ else
+ return result
+ end
+ end
+
+ alias shift pop
+ alias deq pop
+
+ end
+
+end