You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by sh...@apache.org on 2011/10/21 03:20:13 UTC
svn commit: r1187150 [3/43] - in /qpid/branches/QPID-2519: ./ bin/ cpp/
cpp/bindings/ cpp/bindings/qmf/python/ cpp/bindings/qmf/ruby/
cpp/bindings/qmf/tests/ cpp/bindings/qmf2/ cpp/bindings/qmf2/examples/cpp/
cpp/bindings/qmf2/python/ cpp/bindings/qmf2...
Modified: qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i Fri Oct 21 01:19:00 2011
@@ -21,357 +21,21 @@
%include "std_string.i"
%include "../../swig_python_typemaps.i"
-/* Needed for get/setPriority methods. Surprising SWIG 1.3.40 doesn't
- * convert uint8_t by default. */
-%apply unsigned char { uint8_t };
-
-
-/*
- * Exceptions
- *
- * The convention below is that exceptions in _cqpid.so have the same
- * names as in the C++ library. They get renamed to their Python
- * equivalents when brought into the Python wrapping
- */
-%{
-static PyObject* pNoMessageAvailable;
-static PyObject* pTargetCapacityExceeded;
-static PyObject* pNotFound;
-static PyObject* pTransportFailure;
-%}
-
-%init %{
- pNoMessageAvailable = PyErr_NewException(
- "_cqpid.NoMessageAvailable", NULL, NULL);
- Py_INCREF(pNoMessageAvailable);
- PyModule_AddObject(m, "NoMessageAvailable", pNoMessageAvailable);
-
- pTargetCapacityExceeded = PyErr_NewException(
- "_cqpid.TargetCapacityExceeded", NULL, NULL);
- Py_INCREF(pTargetCapacityExceeded);
- PyModule_AddObject(m, "TargetCapacityExceeded", pTargetCapacityExceeded);
-
- pNotFound = PyErr_NewException(
- "_cqpid.NotFound", NULL, NULL);
- Py_INCREF(pNotFound);
- PyModule_AddObject(m, "NotFound", pNotFound);
-
- pTransportFailure = PyErr_NewException(
- "_cqpid.TransportFailure", NULL, NULL);
- Py_INCREF(pTransportFailure);
- PyModule_AddObject(m, "TransportFailure", pTransportFailure);
-%}
-
-%pythoncode %{
- Empty = _cqpid.NoMessageAvailable
- TargetCapacityExceeded = _cqpid.TargetCapacityExceeded
- NotFound = _cqpid.NotFound
- ConnectError = _cqpid.TransportFailure
-%}
-
/* Define the general-purpose exception handling */
%exception {
- PyObject * pExceptionType = NULL;
std::string error;
Py_BEGIN_ALLOW_THREADS;
try {
$action
- } catch (qpid::messaging::NoMessageAvailable & ex) {
- pExceptionType = pNoMessageAvailable;
- error = ex.what();
- } catch (qpid::messaging::TargetCapacityExceeded & ex) {
- pExceptionType = pTargetCapacityExceeded;
- error = ex.what();
- } catch (qpid::messaging::NotFound & ex) {
- pExceptionType = pNotFound;
- error = ex.what();
- } catch (qpid::messaging::TransportFailure & ex) {
- pExceptionType = pTransportFailure;
- error = ex.what();
} catch (qpid::types::Exception& ex) {
- pExceptionType = PyExc_RuntimeError;
error = ex.what();
}
Py_END_ALLOW_THREADS;
if (!error.empty()) {
- PyErr_SetString(pExceptionType, error.c_str());
+ PyErr_SetString(PyExc_RuntimeError, error.c_str());
return NULL;
}
}
-
-/* This only renames the non-const version (I believe). Then again, I
- * don't even know why there is a non-const version of the method. */
-%rename(opened) qpid::messaging::Connection::isOpen();
-%rename(receiver) qpid::messaging::Session::createReceiver;
-%rename(sender) qpid::messaging::Session::createSender;
-%rename(_acknowledge_all) qpid::messaging::Session::acknowledge(bool);
-%rename(_acknowledge_msg) qpid::messaging::Session::acknowledge(
- Message &, bool);
-
-%rename(_fetch) qpid::messaging::Receiver::fetch;
-%rename(unsettled) qpid::messaging::Receiver::getUnsettled;
-%rename(available) qpid::messaging::Receiver::getAvailable;
-
-%rename(unsettled) qpid::messaging::Sender::getUnsettled;
-%rename(available) qpid::messaging::Sender::getAvailable;
-%rename(_send) qpid::messaging::Sender::send;
-
-%rename(_getReplyTo) qpid::messaging::Message::getReplyTo;
-%rename(_setReplyTo) qpid::messaging::Message::setReplyTo;
-%rename(_getTtl) qpid::messaging::Message::getTtl;
-%rename(_setTtl) qpid::messaging::Message::setTtl;
-
-
%include "../qpid.i"
-%extend qpid::messaging::Connection {
- %pythoncode %{
- # Handle the different options by converting underscores to hyphens.
- # Also, the sasl_mechanisms option in Python has no direct
- # equivalent in C++, so we will translate them to sasl_mechanism
- # when possible.
- def __init__(self, url=None, **options):
- args = [url] if url else []
- if options :
- if "sasl_mechanisms" in options :
- if ' ' in options.get("sasl_mechanisms",'') :
- raise Exception(
- "C++ Connection objects are unable to handle "
- "multiple sasl-mechanisms")
- options["sasl_mechanism"] = options.pop("sasl_mechanisms")
- args.append(options)
- this = _cqpid.new_Connection(*args)
- try: self.this.append(this)
- except: self.this = this
- %}
-
- /* Return a pre-existing session with the given name, if one
- * exists, otherwise return a new one. (Note that if a
- * pre-existing session exists, the transactional argument is
- * ignored, and the returned session might not satisfy the desired
- * setting. */
- qpid::messaging::Session _session(const std::string & name,
- bool transactional) {
- if (!name.empty()) {
- try {
- return self->getSession(name);
- }
- catch (const qpid::messaging::KeyError &) {
- }
- }
- if (transactional) {
- return self->createTransactionalSession(name);
- }
- else {
- return self->createSession(name);
- }
- }
-
- %pythoncode %{
- def session(self, name=None, transactional=False) :
- if name is None :
- name = ''
- return self._session(name, transactional)
- %}
-
- %pythoncode %{
- @staticmethod
- def establish(url=None, **options) :
- conn = Connection(url, **options)
- conn.open()
- return conn
- %}
-}
-
-%extend qpid::messaging::Session {
- %pythoncode %{
- def acknowledge(self, message=None, disposition=None, sync=True) :
- if disposition :
- raise Exception("SWIG does not support dispositions yet. Use "
- "Session.reject and Session.release instead")
- if message :
- self._acknowledge_msg(message, sync)
- else :
- self._acknowledge_all(sync)
-
- __swig_getmethods__["connection"] = getConnection
- if _newclass: connection = _swig_property(getConnection)
- %}
-}
-
-
-%extend qpid::messaging::Receiver {
- %pythoncode %{
- __swig_getmethods__["capacity"] = getCapacity
- __swig_setmethods__["capacity"] = setCapacity
- if _newclass: capacity = _swig_property(getCapacity, setCapacity)
-
- __swig_getmethods__["session"] = getSession
- if _newclass: session = _swig_property(getSession)
- %}
-
- %pythoncode %{
- def fetch(self, timeout=None) :
- if timeout is None :
- return self._fetch()
- else :
- # Python API uses timeouts in seconds,
- # but C++ API uses milliseconds
- return self._fetch(Duration(int(1000*timeout)))
- %}
-}
-
-%extend qpid::messaging::Sender {
- %pythoncode %{
- def send(self, object, sync=True) :
- if isinstance(object, Message):
- message = object
- else:
- message = Message(object)
- return self._send(message, sync)
-
- __swig_getmethods__["capacity"] = getCapacity
- __swig_setmethods__["capacity"] = setCapacity
- if _newclass: capacity = _swig_property(getCapacity, setCapacity)
-
- __swig_getmethods__["session"] = getSession
- if _newclass: session = _swig_property(getSession)
- %}
-}
-
-
-%extend qpid::messaging::Message {
- %pythoncode %{
- # UNSPECIFIED was module level before, but I do not
- # know how to insert python code at the top of the module.
- # (A bare "%pythoncode" inserts at the end.
- UNSPECIFIED=object()
- def __init__(self, content=None, content_type=UNSPECIFIED, id=None,
- subject=None, user_id=None, reply_to=None,
- correlation_id=None, durable=None, priority=None,
- ttl=None, properties=None):
- this = _cqpid.new_Message('')
- try: self.this.append(this)
- except: self.this = this
- if content :
- self.content = content
- if content_type != UNSPECIFIED :
- self.content_type = content_type
- if id is not None :
- self.id = id
- if subject is not None :
- self.subject = subject
- if user_id is not None :
- self.user_id = user_id
- if reply_to is not None :
- self.reply_to = reply_to
- if correlation_id is not None :
- self.correlation_id = correlation_id
- if durable is not None :
- self.durable = durable
- if priority is not None :
- self.priority = priority
- if ttl is not None :
- self.ttl = ttl
- if properties is not None :
- # Can't set properties via (inst).getProperties, because
- # the typemaps make a copy of the underlying properties.
- # Instead, set via setProperty for the time-being
- for k, v in properties.iteritems() :
- self.setProperty(k, v)
-
- def _get_content(self) :
- if self.content_type == "amqp/list" :
- return decodeList(self)
- if self.content_type == "amqp/map" :
- return decodeMap(self)
- return self.getContent()
- def _set_content(self, content) :
- if isinstance(content, basestring) :
- self.setContent(content)
- elif isinstance(content, list) or isinstance(content, dict) :
- encode(content, self)
- else :
- # Not a type we can handle. Try setting it anyway,
- # although this will probably lead to a swig error
- self.setContent(content)
- __swig_getmethods__["content"] = _get_content
- __swig_setmethods__["content"] = _set_content
- if _newclass: content = _swig_property(_get_content, _set_content)
-
- __swig_getmethods__["content_type"] = getContentType
- __swig_setmethods__["content_type"] = setContentType
- if _newclass: content_type = _swig_property(getContentType,
- setContentType)
-
- __swig_getmethods__["id"] = getMessageId
- __swig_setmethods__["id"] = setMessageId
- if _newclass: id = _swig_property(getMessageId, setMessageId)
-
- __swig_getmethods__["subject"] = getSubject
- __swig_setmethods__["subject"] = setSubject
- if _newclass: subject = _swig_property(getSubject, setSubject)
-
- __swig_getmethods__["priority"] = getPriority
- __swig_setmethods__["priority"] = setPriority
- if _newclass: priority = _swig_property(getPriority, setPriority)
-
- def getTtl(self) :
- return self._getTtl().getMilliseconds()/1000.0
- def setTtl(self, duration) :
- self._setTtl(Duration(int(1000*duration)))
- __swig_getmethods__["ttl"] = getTtl
- __swig_setmethods__["ttl"] = setTtl
- if _newclass: ttl = _swig_property(getTtl, setTtl)
-
- __swig_getmethods__["user_id"] = getUserId
- __swig_setmethods__["user_id"] = setUserId
- if _newclass: user_id = _swig_property(getUserId, setUserId)
-
- __swig_getmethods__["correlation_id"] = getCorrelationId
- __swig_setmethods__["correlation_id"] = setCorrelationId
- if _newclass: correlation_id = _swig_property(getCorrelationId,
- setCorrelationId)
-
- __swig_getmethods__["redelivered"] = getRedelivered
- __swig_setmethods__["redelivered"] = setRedelivered
- if _newclass: redelivered = _swig_property(getRedelivered,
- setRedelivered)
-
- __swig_getmethods__["durable"] = getDurable
- __swig_setmethods__["durable"] = setDurable
- if _newclass: durable = _swig_property(getDurable, setDurable)
-
- __swig_getmethods__["properties"] = getProperties
- if _newclass: properties = _swig_property(getProperties)
-
- def getReplyTo(self) :
- return self._getReplyTo().str()
- def setReplyTo(self, address_str) :
- self._setReplyTo(Address(address_str))
- __swig_getmethods__["reply_to"] = getReplyTo
- __swig_setmethods__["reply_to"] = setReplyTo
- if _newclass: reply_to = _swig_property(getReplyTo, setReplyTo)
-
- def __repr__(self):
- args = []
- for name in ["id", "subject", "user_id", "reply_to",
- "correlation_id", "priority", "ttl",
- "durable", "redelivered", "properties",
- "content_type"] :
- value = getattr(self, name)
- if value : args.append("%s=%r" % (name, value))
- if self.content is not None:
- if args:
- args.append("content=%r" % self.content)
- else:
- args.append(repr(self.content))
- return "Message(%s)" % ", ".join(args)
- %}
-}
-
-%pythoncode %{
-# Bring into module scope
-UNSPECIFIED = Message.UNSPECIFIED
-%}
Modified: qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i Fri Oct 21 01:19:00 2011
@@ -27,7 +27,6 @@
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Duration.h>
-#include <qpid/messaging/FailoverUpdates.h>
//
// Wrapper functions for map-decode and list-decode. This allows us to avoid
@@ -49,7 +48,6 @@ qpid::types::Variant::List& decodeList(c
%}
-%include <qpid/ImportExport.h>
%include <qpid/messaging/ImportExport.h>
%include <qpid/messaging/Address.h>
%include <qpid/messaging/Duration.h>
@@ -58,7 +56,6 @@ qpid::types::Variant::List& decodeList(c
%include <qpid/messaging/Sender.h>
%include <qpid/messaging/Session.h>
%include <qpid/messaging/Connection.h>
-%include <qpid/messaging/FailoverUpdates.h>
qpid::types::Variant::Map& decodeMap(const qpid::messaging::Message&);
qpid::types::Variant::List& decodeList(const qpid::messaging::Message&);
Modified: qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am Fri Oct 21 01:19:00 2011
@@ -21,7 +21,7 @@ if HAVE_RUBY_DEVEL
INCLUDES = -I$(top_srcdir)/include -I$(top_builddir)/include -I$(top_srcdir)/src -I$(top_builddir)/src
-EXTRA_DIST = CMakeLists.txt ruby.i
+EXTRA_DIST = ruby.i
BUILT_SOURCES = cqpid.cpp
SWIG_FLAGS = -w362,401
@@ -33,10 +33,10 @@ cqpid.cpp: $(srcdir)/ruby.i $(srcdir)/..
rubylibarchdir = $(RUBY_LIB_ARCH)
rubylibarch_LTLIBRARIES = cqpid.la
-cqpid_la_LDFLAGS = -avoid-version -module -shared -shrext ".$(RUBY_DLEXT)"
+cqpid_la_LDFLAGS = -avoid-version -module -shrext ".$(RUBY_DLEXT)"
cqpid_la_LIBADD = $(RUBY_LIBS) -L$(top_builddir)/src/.libs -lqpidmessaging -lqpidtypes \
$(top_builddir)/src/libqpidmessaging.la $(top_builddir)/src/libqpidtypes.la
-cqpid_la_CXXFLAGS = $(INCLUDES) -I$(RUBY_INC) -I$(RUBY_INC_ARCH) -fno-strict-aliasing
+cqpid_la_CXXFLAGS = $(INCLUDES) -I$(RUBY_INC) -I$(RUBY_INC_ARCH)
nodist_cqpid_la_SOURCES = cqpid.cpp
CLEANFILES = cqpid.cpp
Modified: qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i Fri Oct 21 01:19:00 2011
@@ -17,25 +17,6 @@
* under the License.
*/
-/* For UUID objects, to convert them to Python uuid.UUID objects,
- * we'll need a reference to the uuid module.
- */
-%{
-static PyObject* pUuidModule;
-%}
-
-%init %{
- pUuidModule = PyImport_ImportModule("uuid");
-
- /* Although it is not required, we'll publish the uuid module in our
- * module, as if this module was a python module and we called
- * "import uuid"
- */
- Py_INCREF(pUuidModule);
- PyModule_AddObject(m, "uuid", pUuidModule);
-%}
-
-
%wrapper %{
#if PY_VERSION_HEX < 0x02050000 && !defined(PY_SSIZE_T_MIN)
@@ -47,7 +28,6 @@ typedef int Py_ssize_t;
PyObject* MapToPy(const qpid::types::Variant::Map*);
PyObject* ListToPy(const qpid::types::Variant::List*);
- PyObject* UuidToPy(const qpid::types::Uuid*);
void PyToMap(PyObject*, qpid::types::Variant::Map*);
void PyToList(PyObject*, qpid::types::Variant::List*);
@@ -124,9 +104,6 @@ typedef int Py_ssize_t;
break;
}
case qpid::types::VAR_UUID : {
- qpid::types::Uuid uuid = v->asUuid();
- result = UuidToPy(&uuid);
- break;
}
}
} catch (qpid::types::Exception& ex) {
@@ -166,30 +143,6 @@ typedef int Py_ssize_t;
return result;
}
- PyObject* UuidToPy(const qpid::types::Uuid * uuid) {
- PyObject* pUuidClass = PyObject_GetAttrString(pUuidModule, "UUID");
- if (!pUuidClass) {
- // Failed to get UUID class
- return 0;
- }
-
- PyObject* pArgs = PyTuple_New(0);
- PyObject* pKw = PyDict_New();
- PyObject* pData = PyString_FromStringAndSize(
- (const char*)(uuid->data()), 16);
- PyDict_SetItemString(pKw, "bytes", pData);
-
- PyObject* result = PyObject_Call(pUuidClass, pArgs, pKw);
-
- Py_DECREF(pData);
- Py_DECREF(pKw);
- Py_DECREF(pArgs);
- Py_DECREF(pUuidClass);
-
- return result;
- }
-
-
void PyToMap(PyObject* obj, qpid::types::Variant::Map* map) {
map->clear();
Py_ssize_t iter(0);
@@ -351,15 +304,6 @@ typedef int Py_ssize_t;
Py_INCREF($result);
}
-/*
- * UUID type: C++ --> Python
- */
-%typemap(out) qpid::types::UUID & {
- $result = UuidToPy($1);
- if ($result)
- Py_INCREF($result);
-}
-
/*
* Variant types: Ruby --> C++
Modified: qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i Fri Oct 21 01:19:00 2011
@@ -49,7 +49,7 @@
}
VALUE VariantToRb(const qpid::types::Variant* v) {
- VALUE result = Qnil;
+ VALUE result;
try {
switch (v->getType()) {
case qpid::types::VAR_VOID: {
Modified: qpid/branches/QPID-2519/cpp/bld-winsdk.ps1
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/bld-winsdk.ps1?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bld-winsdk.ps1 (original)
+++ qpid/branches/QPID-2519/cpp/bld-winsdk.ps1 Fri Oct 21 01:19:00 2011
@@ -186,6 +186,9 @@ function BuildAPlatform
'examples/qmf-console',
'examples/request-response',
'examples/tradedemo',
+ 'examples/old-examples.sln',
+ 'examples/README.*',
+ 'examples/verify*',
'include',
'plugins')
Modified: qpid/branches/QPID-2519/cpp/configure.ac
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/configure.ac?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/configure.ac (original)
+++ qpid/branches/QPID-2519/cpp/configure.ac Fri Oct 21 01:19:00 2011
@@ -68,10 +68,8 @@ if test x$GXX = xyes; then
# The following warnings are deliberately omitted, they warn on valid code.
# -Wunreachable-code -Wpadded -Winline
# -Wshadow - warns about boost headers.
- # Can't test for -Werror as whether it fails or not depends on what's in
- # CFLAGS/CXXFLAGS. In any case it's been in gcc for a long time (since 2.95 at least)
if test "${enableval}" = yes; then
- COMPILER_FLAGS="-Werror"
+ gl_COMPILER_FLAGS(-Werror)
gl_COMPILER_FLAGS(-pedantic)
gl_COMPILER_FLAGS(-Wall)
gl_COMPILER_FLAGS(-Wextra)
@@ -523,19 +521,18 @@ AM_PATH_PYTHON()
builddir_lib_suffix="/.libs"
AC_SUBST([builddir_lib_suffix])
-# Files to generate
+# Files to generate
AC_CONFIG_FILES([
Makefile
examples/Makefile
- examples/old_api/Makefile
- examples/old_api/direct/Makefile
- examples/old_api/fanout/Makefile
- examples/old_api/pub-sub/Makefile
- examples/old_api/request-response/Makefile
- examples/old_api/failover/Makefile
- examples/old_api/xml-exchange/Makefile
+ examples/direct/Makefile
+ examples/fanout/Makefile
+ examples/pub-sub/Makefile
+ examples/request-response/Makefile
+ examples/failover/Makefile
+ examples/xml-exchange/Makefile
examples/qmf-console/Makefile
- examples/old_api/tradedemo/Makefile
+ examples/tradedemo/Makefile
examples/messaging/Makefile
bindings/qpid/Makefile
bindings/qpid/ruby/Makefile
Modified: qpid/branches/QPID-2519/cpp/design_docs/new-cluster-design.txt
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/design_docs/new-cluster-design.txt?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/design_docs/new-cluster-design.txt (original)
+++ qpid/branches/QPID-2519/cpp/design_docs/new-cluster-design.txt Fri Oct 21 01:19:00 2011
@@ -17,6 +17,7 @@
# under the License.
* A new design for Qpid clustering.
+
** Issues with current design.
The cluster is based on virtual synchrony: each broker multicasts
@@ -94,9 +95,8 @@ Use a moving queue ownership protocol to
No longer relies on identical state and lock-step behavior to cause
identical dequeues on each broker.
-Use multiple CPG groups to process different queues in parallel. Use a
-fixed set of groups and hash queue names to choose the group for each
-queue.
+Each queue has an associated thread-context. Events for a queue are executed
+in that queues context, in parallel with events for other queues.
*** Requirements
@@ -149,7 +149,7 @@ a release-queue event, allowing another
ownership.
*** Asynchronous completion of accept
-
+### HERE
In acknowledged mode a message is not forgotten until it is accepted,
to allow for requeue on rejection or crash. The accept should not be
completed till the message has been forgotten.
@@ -162,32 +162,19 @@ On receiving an accept the broker:
NOTE: The message store does not currently implement asynchronous
completions of accept, this is a bug.
-*** Multiple CPG groups.
-
-The old cluster was bottlenecked by processing everything in a single
-CPG deliver thread.
-
-The new cluster uses a set of CPG groups, one per core. Queue names
-are hashed to give group indexes, so statistically queues are likely
-to be spread over the set of groups.
-
-Operations on a given queue always use the same group, so we have
-order within each queue, but operations on different queues can use
-different groups giving greater throughput sending to CPG and multiple
-handler threads to process CPG messages.
-
** Inconsistent errors.
-An inconsistent error means that after multicasting an enqueue, accept
-or dequeue, some brokers succeed in processing it and others fail.
+The new design eliminates most sources of inconsistent errors
+(connections, sessions, security, management etc.) The only points
+where inconsistent errors can occur are at enqueue and dequeue (most
+likely store-related errors.)
+
+The new design can use the exisiting error-handling protocol with one
+major improvement: since brokers are no longer required to maintain
+identical state they do not have to stall processing while an error is
+being resolved.
-The new design eliminates most sources of inconsistent errors in the
-old broker: connections, sessions, security, management etc. Only
-store journal errors remain.
-
-The new inconsistent error protocol is similar to the old one with one
-major improvement: brokers do not have to stall processing while an
-error is being resolved.
+#TODO: The only source of dequeue errors is probably an unrecoverable journal failure.
** Updating new members
@@ -206,44 +193,60 @@ catch up (which is not guaranteed to hap
With the new cluster design only exchanges, queues, bindings and
messages need to be replicated.
-We update individual objects (queues and exchanges) independently.
-- create queues first, then update all queues and exchanges in parallel.
-- multiple updater threads, per queue/exchange.
-
-Queue updater:
-- marks the queue position at the sync point
-- sends messages starting from the sync point working towards the head of the queue.
-- send "done" message.
-
-Queue updatee:
-- enqueues received from CPG: add to back of queue as normal.
-- dequeues received from CPG: apply if found, else save to check at end of update.
-- messages from updater: add to the *front* of the queue.
-- update complete: apply any saved dequeues.
-
-Exchange updater:
-- updater: send snapshot of exchange as it was at the sync point.
-
-Exchange updatee:
-- queue exchange operations after the sync point.
-- when snapshot is received: apply saved operations.
-
-Note:
-- Updater is active throughout, no stalling.
-- Consuming clients actually reduce the size of the update.
-- Updatee stalls clients until the update completes.
- (Note: May be possible to avoid updatee stall as well, needs thought)
-
-** Internal cluster interface
-
-The new cluster interface is similar to the MessageStore interface, but
-provides more detail (message positions) and some additional call
-points (e.g. acquire)
+Update of wiring (exchanges, queues, bindings) is the same as current
+design.
+
+Update of messages is different:
+- per-queue rather than per-broker, separate queues can be updated in parallel.
+- updates queues in reverse order to eliminate unbounded catch-up
+- does not require updater & updatee to stall during update.
+
+Replication events, multicast to cluster:
+- enqueue(q,m): message m pushed on back of queue q .
+- acquire(q,m): mark m acquired
+- dequeue(q,m): forget m.
+Messages sent on update connection:
+- update_front(q,m): during update, receiver pushes m to *front* of q
+- update_done(q): during update, update of q is complete.
+
+Updater:
+- when updatee joins set iterator i = q.end()
+- while i != q.begin(): --i; send update_front(q,*i) to updatee
+- send update_done(q) to updatee
+
+Updatee:
+- q initially in locked state, can't dequeue locally.
+- start processing replication events for q immediately (enqueue, dequeue, acquire etc.)
+- receive update_front(q,m): q.push_front(m)
+- receive update_done(q): q can be unlocked for local dequeing.
+
+Benefits:
+- Stall only for wiring update: updater & updatee can process multicast messages while messages are updated.
+- No unbounded catch-up: update consists of at most N update_front() messages where N=q.size() at start of update.
+- During update consumers actually help by removing messages before they need to be updated.
+- Needs no separate "work to do" queue, only the broker queues themselves.
+
+# TODO how can we recover from updater crashing before update complete?
+# Clear queues that are not updated & send request for udpates on those queues?
+
+# TODO updatee may receive a dequeue for a message it has not yet seen, needs
+# to hold on to that so it can drop the message when it is seen.
+# Similar problem exists for wiring?
+
+** Cluster API
+
+The new cluster API is similar to the MessageStore interface.
+(Initially I thought it would be an extension of the MessageStore interface,
+but as the design develops it seems better to make it a separate interface.)
The cluster interface captures these events:
- wiring changes: queue/exchange declare/bind
- message enqueued/acquired/released/rejected/dequeued.
-- transactional events.
+
+The cluster will require some extensions to the Queue:
+- Queues can be "locked", locked queues are ignored by IO-driven output.
+- Cluster must be able to apply queue events from the cluster to a queue.
+ These appear to fit into existing queue operations.
** Maintainability
@@ -270,48 +273,106 @@ A number of specific ways the code will
** Performance
-The standalone broker processes _connections_ concurrently, so CPU
-usage increases as you add more connections.
-
-The new cluster processes _queues_ concurrently, so CPU usage increases as you
-add more queues.
-
-In both cases, CPU usage peaks when the number of "units of
- concurrency" (connections or queues) goes above the number of cores.
+The only way to verify the relative performance of the new design is
+to prototype & profile. The following points suggest the new design
+may scale/perform better:
+
+Some work moved from virtual synchrony thread to connection threads:
+- All connection/session logic moves to connection thread.
+- Exchange routing logic moves to connection thread.
+- On local broker dequeueing is done in connection thread
+- Local broker dequeue is IO driven as for a standalone broker.
+
+For queues with all consumers on a single node dequeue is all
+IO-driven in connection thread. Pay for time-sharing only if queue has
+consumers on multiple brokers.
+
+Doing work for different queues in parallel scales on multi-core boxes when
+there are multiple queues.
+
+One difference works against performance, thre is an extra
+encode/decode. The old design multicasts raw client data and decodes
+it in the virtual synchrony thread. The new design would decode
+messages in the connection thread, re-encode them for multicast, and
+decode (on non-local brokers) in the virtual synchrony thread. There
+is extra work here, but only in the *connection* thread: on a
+multi-core machine this happens in parallel for every connection, so
+it probably is not a bottleneck. There may be scope to optimize
+decode/re-encode by re-using some of the original encoded data, this
+could also benefit the stand-alone broker.
+
+** Asynchronous queue replication
+
+The existing "asynchronous queue replication" feature maintains a
+passive backup passive backup of queues on a remote broker over a TCP
+connection.
+
+The new cluster replication protocol could be re-used to implement
+asynchronous queue replication: its just a special case where the
+active broker is always the queue owner and the enqueue/dequeue
+messages are sent over a TCP connection rather than multicast.
+
+The new update update mechanism could also work with 'asynchronous
+queue replication', allowing such replication (over a TCP connection
+on a WAN say) to be initiated after the queue had already been created
+and been in use (one of the key missing features).
+
+** Increasing Concurrency and load sharing
+
+The current cluster is bottlenecked by processing everything in the
+CPG deliver thread. By removing the need for identical operation on
+each broker, we open up the possiblility of greater concurrency.
+
+Handling multicast enqueue, acquire, accpet, release etc: concurrency
+per queue. Operatons on different queues can be done in different
+threads.
+
+The new design does not force each broker to do all the work in the
+CPG thread so spreading load across cluster members should give some
+scale-up.
+
+** Misc outstanding issues & notes
+
+Replicating wiring
+- Need async completion of wiring commands?
+- qpid.sequence_counter: need extra work to support in new design, do we care?
+
+Cluster+persistence:
+- finish async completion: dequeue completion for store & cluster
+- cluster restart from store: clean stores *not* identical, pick 1, all others update.
+- need to generate cluster ids for messages recovered from store.
+
+Live updates: we don't need to stall brokers during an update!
+- update on queue-by-queue basis.
+- updatee locks queues during update, no dequeue.
+- update in reverse: don't update messages dequeued during update.
+- updatee adds update messages at front (as normal), replicated messages at back.
+- updater starts from back, sends "update done" when it hits front of queue.
+
+Flow control: need to throttle multicasting
+1. bound the number of outstanding multicasts.
+2. ensure the entire cluster keeps up, no unbounded "lag"
+The existing design uses read-credit to solve 1., and does not solve 2.
+New design should stop reading on all connections while flow control
+condition exists?
+
+Can federation also be unified, at least in configuration?
+
+Consider queues (and exchanges?) as having "reliability" attributes:
+- persistent: is the message stored on disk.
+- backed-up (to another broker): active/passive async replication.
+- replicated (to a cluster): active/active multicast replication to cluster.
+- federated: federation link to a queue/exchange on another broker.
+
+"Reliability" seems right for the first 3 but not for federation, is
+there a better term?
+
+Clustering and scalability: new design may give us the flexibility to
+address scalability as part of cluster design. Think about
+relationship to federation and "fragmented queues" idea.
-When all consumers on a queue are connected to the same broker the new
-cluster uses the same messagea allocation threading/logic as a
-standalone broker, with a little extra asynchronous book-keeping.
-
-If a queue has multiple consumers connected to multiple brokers, the
-new cluster time-shares the queue which is less efficient than having
-all consumers on a queue connected to the same broker.
-
-** Flow control
-New design does not queue up CPG delivered messages, they are
-processed immediately in the CPG deliver thread. This means that CPG's
-flow control is sufficient for qpid.
-
-** Live upgrades
-
-Live upgrades refers to the ability to upgrade a cluster while it is
-running, with no downtime. Each brokers in the cluster is shut down,
-and then re-started with a new version of the broker code.
-
-To achieve this
-- Cluster protocl XML file has a new element <version number=N> attached
- to each method. This is the version at which the method was added.
-- New versions can only add methods, existing methods cannot be changed.
-- The cluster handshake for new members includes the protocol version
- at each member.
-- The cluster's version is the lowest version among its members.
-- A newer broker can join and older cluster. When it does, it must restrict
- itself to speaking the older version protocol.
-- When the cluster version increases (because the lowest version member has left)
- the remaining members may move up to the new version.
+* Design debates/descisions
-
-* Design debates
** Active/active vs. active passive
An active-active cluster can be used in an active-passive mode. In
@@ -324,7 +385,7 @@ An active/passive implementation allows
- can do immediate local enqueue and still guarantee order.
Active/passive introduces a few extra requirements:
-- Exactly one broker has to take over if primary fails.
+- Exactly one broker hast to take over if primary fails.
- Passive members must refuse client connections.
- On failover, clients must re-try all known addresses till they find the active member.
@@ -332,17 +393,43 @@ Active/active benefits:
- A broker failure only affects the subset of clients connected to that broker.
- Clients can switch to any other broker on failover
- Backup brokers are immediately available on failover.
-- As long as a client can connect to any broker in the cluster, it can be served.
+- Some load sharing: reading from client + multicast only done on direct node.
+
+Active/active drawbacks:
+- Co-ordinating message acquisition may impact performance (not tested)
+- Code may be more complex that active/passive.
Active/passive benefits:
-- Don't need to replicate message allocation, can feed consumers at top speed.
+- Don't need message allocation strategy, can feed consumers at top speed.
+- Code may be simpler than active/active.
Active/passive drawbacks:
- All clients on one node so a failure affects every client in the system.
- After a failure there is a "reconnect storm" as every client reconnects to the new active node.
- After a failure there is a period where no broker is active, until the other brokers realize the primary is gone and agree on the new primary.
- Clients must find the single active node, may involve multiple connect attempts.
-- No service if a partition separates a client from the active broker,
- even if the client can see other brokers.
+** Total ordering.
+
+Initial thinking: allow message ordering to differ between brokers.
+New thinking: use CPG total ordering, get identical ordering on all brokers.
+- Allowing variation in order introduces too much chance of unexpected behavior.
+- Usign total order allows other optimizations, see Message Identifiers below.
+
+** Message identifiers.
+
+Initial thinking: message ID = CPG node id + 64 bit sequence number.
+This involves a lot of mapping between cluster IDs and broker messsages.
+
+New thinking: message ID = queue name + queue position.
+- Removes most of the mapping and memory management for cluster code.
+- Requires total ordering of messages (see above)
+
+** Message rejection
+
+Initial thinking: add special reject/rejected points to cluster interface so
+rejected messages could be re-queued without multicast.
+New thinking: treat re-queueing after reject as entirely new message.
+- Simplifies cluster interface & implementation
+- Not on the critical path.
Modified: qpid/branches/QPID-2519/cpp/design_docs/new-cluster-plan.txt
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/design_docs/new-cluster-plan.txt?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/design_docs/new-cluster-plan.txt (original)
+++ qpid/branches/QPID-2519/cpp/design_docs/new-cluster-plan.txt Fri Oct 21 01:19:00 2011
@@ -17,150 +17,376 @@
# specific language governing permissions and limitations
# under the License.
-* Status of impementation
-Meaning of priorities:
-[#A] Essential for basic functioning.
-[#B] Required for first release.
-[#C] Can be addressed in a later release.
+Notes on new cluster implementation. See also: new-cluster-design.txt
-The existig prototype is bare bones to do performance benchmarks:
-- Implements publish and consumer locking protocol.
-- Defered delivery and asynchronous completion of message.
-- Optimize the case all consumers are on the same node.
-- No new member updates, no failover updates, no transactions, no persistence etc.
+* Implementation plan.
-Prototype code is on branch qpid-2920-active, in cpp/src/qpid/cluster/exp/
-
-** Similarities to existing cluster.
-
-/Active-active/: the new cluster can be a drop-in replacement for the
-old, existing tests & customer deployment configurations are still
-valid.
-
-/Virtual synchrony/: Uses corosync to co-ordinate activity of members.
-
-/XML controls/: Uses XML to define the primitives multicast to the
-cluster.
-
-** Differences with existing cluster.
-
-/Report rather than predict consumption/: brokers explicitly tell each
-other which messages have been acquired or dequeued. This removes the
-major cause of bugs in the existing cluster.
-
-/Queue consumer locking/: to avoid duplicates only one broker can acquire or
-dequeue messages at a time - while has the consume-lock on the
-queue. If multiple brokers are consuming from the same queue the lock
-is passed around to time-share access to the queue.
-
-/Per-queue concurrency/: uses a fixed-size set of CPG groups (reflecting
-the concurrency of the host) to allow concurrent processing on
-different queues. Queues are hashed onto the groups.
-
-* Completed tasks
-** DONE [#A] Minimal POC: publish/acquire/dequeue protocol.
- CLOSED: [2011-10-05 Wed 16:03]
-
-Defines broker::Cluster interface and call points.
-Initial interface commite
-
-Main classes
-Core: central object holding cluster classes together (replaces cluster::Cluster)
-BrokerContext: implements broker::Cluster interface.
-QueueContext: Attached to a broker::Queue, holds cluster status.
-MessageHolder:holds local messages while they are being enqueued.
-
-Implements multiple CPG groups for better concurrency.
-
-** DONE [#A] Large message replication.
- CLOSED: [2011-10-05 Wed 17:22]
-Multicast using fixed-size (64k) buffers, allow fragmetation of messages across buffers (frame by frame)
-
-* Open questions
-
-** TODO [#A] Queue sequence numbers vs. independant message IDs.
- SCHEDULED: <2011-10-07 Fri>
-
-Current prototype uses queue sequence numbers to identify
-message. This is tricky for updating new members as the sequence
-numbers are only known on delivery.
-
-Independent message IDs that can be generated and sent with the message simplify
-this and potentially allow performance benefits by relaxing total ordering.
-However they imply additional map lookups that might hurt performance.
-
-- [ ] Prototype independent message IDs, check performance.
-
-* Outstanding Tasks
-** TODO [#A] Defer and async completion of wiring commands.
-
-Testing requirement: Many tests assume wiring changes are visible
-across the cluster once the commad completes.
-
-Name clashes: need to avoid race if same name queue/exchange declared
-on 2 brokers simultaneously
-
-** TODO [#A] Passing all existing cluster tests.
-
-The new cluster should be a drop-in replacement for the old, so it
-should be able to pass all the existing tests.
+Co-existence with old cluster code and tests:
+- Separate plugin cluster2, options --cluster2-*. Eventually renamed to replace cluster.
+- Double up tests with old version/new version as the new code develops.
+
+Minimal POC for message delivery & perf test.
+- no wiring replication, no updates, no failover, no persistence, no async completion.
+- just implement publish and acquire/dequeue locking protocol.
+- optimize the special case where all consumers are on the same node.
+- measure performance: compare active-passive and active-active modes of use.
+
+Full implementation of transient cluster
+- Update (based on existing update), async completion etc.
+- Passing all existing transient cluster tests.
+
+Persistent cluster
+- Make sure async completion works correctly.
+- InitialStatus protoocl etc. to support persistent start-up (existing code)
+- cluster restart from store: stores not identical. Load one, update the rest.
+ - assign cluster ID's to messages recovered from store, don't replicate.
+
+Improved update protocol
+- per-queue, less stalling, bounded catch-up.
+
+* Task list
+
+** TODO [#A] Minimal POC: publish/acquire/dequeue protocol.
+
+NOTE: as implementation questions arise, take the easiest option and make
+a note for later optimization/improvement.
+
+*** Tests
+- python test: 4 senders, numbered messages, 4 receivers, verify message set.
+- acquire then release messages: verify can be dequeued on any member
+- acquire then kill broker: verify can be dequeued other members.
+- acquire then reject: verify goes on alt-exchange once only.
+
+*** DONE broker::Cluster interface and call points.
+
+Initial interface commited.
+
+*** Main classes
+
+BrokerHandler:
+- implements broker::Cluster intercept points.
+- sends mcast events to inform cluster of local actions.
+- thread safe, called in connection threads.
+
+LocalMessageMap:
+- Holds local messages while they are being enqueued.
+- thread safe: called by both BrokerHandler and MessageHandler
+
+MessageHandler:
+- handles delivered mcast messages related to messages.
+- initiates local actions in response to mcast events.
+- thread unsafe, only called in deliver thread.
+- maintains view of cluster state regarding messages.
+
+QueueOwnerHandler:
+- handles delivered mcast messages related to queue consumer ownership.
+- thread safe, called in deliver, connection and timer threads.
+- maintains view of cluster state regarding queue ownership.
+
+cluster::Core: class to hold new cluster together (replaces cluster::Cluster)
+- thread safe: manage state used by both MessageHandler and BrokerHandler
+
+The following code sketch illustrates only the "happy path" error handling
+is omitted.
+
+*** BrokerHandler
+Types:
+- struct QueuedMessage { Message msg; QueueName q; SequenceNumber position; }
+- struct
+
+NOTE:
+- Messages on queues are identified by a queue name + a position.
+- Messages being routed are identified by a sequence number.
+
+Members:
+- thread_local bool noReplicate // suppress replication.
+- thread_local bool isRouting // suppress operations while routing
+- Message localMessage[SequenceNumber] // local messages being routed.
+- thread_local SequenceNumber routingSequence
+
+NOTE: localMessage is also modified by MessageHandler.
+
+broker::Cluster intercept functions:
+
+routing(msg)
+ if noReplicate: return
+ # Supress everything except enqueues while we are routing.
+ # We don't want to replicate acquires & dequeues caused by an enqueu,
+ # e.g. removal of messages from ring/LV queues.
+ isRouting = true
+
+enqueue(qmsg):
+ if noReplicate: return
+ if routingSequence == 0 # thread local
+ routingSequence = nextRoutingSequence()
+ mcast create(encode(qmsg.msg),routingSeq)
+ mcast enqueue(qmsg.q,routingSeq)
+
+routed(msg):
+ if noReplicate: return
+ isRouting = false
+
+acquire(qmsg):
+ if noReplicate: return
+ if isRouting: return # Ignore while we are routing a message.
+ if msg.id: mcast acquire(qmsg)
+
+release(QueuedMessage)
+ if noReplicate: return
+ if isRouting: return # Ignore while we are routing a message.
+ mcast release(qmsg)
+
+accept(QueuedMessage):
+ if noReplicate: return
+ if isRouting: return # Ignore while we are routing a message.
+ mcast accept(qmsg)
+
+reject(QueuedMessage):
+ isRejecting = true
+ mcast reject(qmsg)
+
+# FIXME no longer needed?
+drop(QueuedMessage)
+ cleanup(qmsg)
+
+*** MessageHandler and mcast messages
+Types:
+- struct QueueEntry { QueuedMessage qmsg; NodeId acquired; }
+- struct QueueKey { MessageId id; QueueName q; }
+- typedef map<QueueKey, QueueEntry> Queue
+- struct Node { Message routing[SequenceNumber]; list<QueueKey> acquired; }
+
+Members:
+- QueueEntry enqueued[QueueKey]
+- Node node[NodeId]
+
+Mcast messages in Message class:
+
+create(msg,seq)
+ if sender != self: node[sender].routing[seq] = decode(msg)
+
+enqueue(q,seq):
+ id = (sender,seq)
+ if sender == self:
+ enqueued[id,q] = (localMessage[seq], acquired=None)
+ else:
+ msg = sender.routing[seq]
+ enqueued[id,q] = (qmsg, acquired=None)
+ with noReplicate=true: qmsg = broker.getQueue(q).push(msg)
+
+routed(seq):
+ if sender == self: localMessage.erase(msg.id.seq)
+ else: sender.routing.erase(seq)
+
+acquire(id,q):
+ enqueued[id,q].acquired = sender
+ node[sender].acquired.push_back((id,q))
+ if sender != self:
+ with noReplicate=true: broker.getQueue(q).acquire(enqueued[id,q])
+
+release(id,q)
+ enqueued[id,q].acquired = None
+ node[sender].acquired.erase((id,q))
+ if sender != self
+ with noReplicate=true: broker.getQueue(q).requeue(enqueued[id,q])
+
+reject(id,q):
+ sender.routing[id] = enqueued[id,q] # prepare for re-queueing
+
+rejected(id,q)
+ sender.routing.erase[id]
+
+dequeue(id,q)
+ entry = enqueued[id,q]
+ enqueued.erase[id,q]
+ node[entry.acquired].acquired.erase(id,q)
+ if sender != self:
+ with noReplicate=true: broker.getQueue(q).dequeue(entry.qmsg)
+
+member m leaves cluster:
+ for key in node[m].acquired:
+ release(key.id, key.q)
+ node.erase(m)
+
+*** Queue consumer locking
+
+When a queue is locked it does not deliver messages to its consumers.
+
+New broker::Queue functions:
+- stopConsumers(): set consumersStopped flag, wait for currently busy consumers to exit.
+- startConsumers(): reset consumersStopped flag
+
+Implementation sketch, locking omitted:
+
+void Queue::stopConsumers() {
+ consumersStopped = true;
+ while (consumersBusy) consumersBusyMonitor.wait();
+}
+
+void Queue::startConsumers() {
+ consumersStopped = false;
+ listeners.notify();
+}
+
+bool Queue::dispatch(consumer) {
+ if (consumersStopped) return false;
+ ++consumersBusy;
+ do_regular_dispatch_body()
+ if (--consumersBusy == 0) consumersBusyMonitor.notify();
+}
+
+*** QueueOwnerHandler
+
+Invariants:
+- Each queue is owned by at most one node at any time.
+- Each node is interested in a set of queues at any given time.
+- A queue is un-owned if no node is interested.
+
+The queue owner releases the queue when
+- it loses interest i.e. queue has no consumers with credit.
+- a configured time delay expires and there are other interested nodes.
+
+The owner mcasts release(q). On delivery the new queue owner is the
+next node in node-id order (treating nodes as a circular list)
+starting from the old owner that is interested in the queue.
+
+Queue consumers initially are stopped, only started when we get
+ownership from the cluster.
+
+Thread safety: called by deliver, connection and timer threads, needs locking.
+
+Thread safe object per queue holding queue ownership status.
+Called by deliver, connection and timer threads.
+
+class QueueOwnership {
+ bool owned;
+ Timer timer;
+ BrokerQueue q;
+
+ drop(): # locked
+ if owned:
+ owned = false
+ q.stopConsumers()
+ mcast release(q.name, false)
+ timer.stop()
+
+ take(): # locked
+ if not owned:
+ owned = true
+ q.startConsumers()
+ timer.start(timeout)
+
+ timer.fire(): drop()
+}
+
+Data Members, only modified/examined in deliver thread:
+- typedef set<NodeId> ConsumerSet
+- map<QueueName, ConsumerSet> consumers
+- map<QueueName, NodeId> owner
+
+Thread safe data members, accessed in connection threads (via BrokerHandler):
+- map<QueueName, QueueOwnership> ownership
+
+Multicast messages in QueueOwner class:
+
+consume(q):
+ if sender==self and consumers[q].empty(): ownership[q].take()
+ consumers[q].insert(sender)
+
+release(q):
+ asssert(owner[q] == sender and owner[q] in consumers[q])
+ owner[q] = circular search from sender in consumers[q]
+ if owner==self: ownership[q].take()
+
+cancel(q):
+ assert(queue[q].owner != sender) # sender must release() before cancel()
+ consumers[q].erase(sender)
+
+member-leaves:
+ for q in queue: if owner[q] = left: left.release(q)
+
+Need 2 more intercept points in broker::Cluster:
+
+consume(q,consumer,consumerCount) - Queue::consume()
+ if consumerCount == 1: mcast consume(q)
+
+cancel(q,consumer,consumerCount) - Queue::cancel()
+ if consumerCount == 0:
+ ownership[q].drop()
+ mcast cancel(q)
+
+#TODO: lifecycle, updating cluster data structures when queues are destroyed
+
+*** Increasing concurrency
+The major performance limitation of the old cluster is that it does
+everything in the single CPG deliver thread context.
+
+We can get additional concurrency by creating a thread context _per queue_
+for queue operations: enqueue, acquire, accept etc.
+
+We associate a PollableQueue of queue operations with each AMQP queue.
+The CPG deliver thread would
+- build messages and associate with cluster IDs.
+- push queue ops to the appropriate PollableQueue to be dispatched the queues thread.
+
+Serializing operations on the same queue avoids contention, but takes advantage
+of the independence of operations on separate queues.
+
+*** Re-use of existing cluster code
+- re-use Event
+- re-use Multicaster
+- re-use same PollableQueueSetup (may experiment later)
+- new Core class to replace Cluster.
+- keep design modular, keep threading rules clear.
+
+** TODO [#B] Large message replication.
+Multicast should encode messages in fixed size buffers (64k)?
+Can't assume we can send message in one chunk.
+For 0-10 can use channel numbers & send whole frames packed into larger buffer.
+** TODO [#B] Transaction support.
+Extend broker::Cluster interface to capture transaction context and completion.
+Sequence number to generate per-node tx IDs.
+Replicate transaction completion.
+** TODO [#B] Batch CPG multicast messages
+The new cluster design involves a lot of small multicast messages,
+they need to be batched into larger CPG messages for efficiency.
+** TODO [#B] Genuine async completion
+Replace current synchronous waiting implementation with genuine async completion.
-** TODO [#A] Update to new members joining.
+Test: enhance test_store.cpp to defer enqueueComplete till special message received.
-Need to resolve [[Queue sequence numbers vs. independant message IDs]] first.
-- implicit sequence numbers are more tricky to replicate to new member.
+Async callback uses *requestIOProcessing* to queue action on IO thread.
-Update individual objects (queues and exchanges) independently.
-- create queues first, then update all queues and exchanges in parallel.
-- multiple updater threads, per queue/exchange.
-- updater sends messages to special exchange(s) (not using extended AMQP controls)
+** TODO [#B] Async completion of accept when dequeue completes.
+Interface is already there on broker::Message, just need to ensure
+that store and cluster implementations call it appropriately.
-Queue updater:
-- marks the queue position at the sync point
-- sends messages starting from the sync point working towards the head of the queue.
-- send "done" message.
-Note: updater remains active throughout, consuming clients actually reduce the
-size of the update.
+** TODO [#B] Replicate wiring.
+From messageStore create/destroy/bind, replicate encoded declare/destroy/bind command.
-Queue updatee:
-- enqueues received from CPG: add to back of queue as normal.
-- dequeues received from CPG: apply if found, else save to check at end of update.
-- messages from updater: add to the *front* of the queue.
-- update complete: apply any saved dequeues.
+** TODO [#B] New members joining - first pass
-Exchange updater:
-- updater: send snapshot of exchange as it was at the sync point.
+Re-use update code from old cluster but don't replicate sessions &
+connections.
-Exchange updatee:
-- queue exchange operations after the sync point.
-- when snapshot is received: apply saved operations.
+Need to extend it to send cluster IDs with messages.
-Updater remains active throughout.
-Updatee stalls clients until the update completes.
+Need to replicate the queue ownership data as part of the update.
-Updating queue/exchange/binding objects is via the same encode/decode
-that is used by the store. Updatee to use recovery interfaces to
-recover?
+** TODO [#B] Persistence support.
+InitialStatus protoocl etc. to support persistent start-up (existing code)
-** TODO [#A] Failover updates to client.
-Implement the amq.failover exchange to notify clients of membership.
+Only one broker recovers from store, update to others.
-** TODO [#B] Initial status protocol.
-Handshake to give status of each broker member to new members joining.
-Status includes
-- persistent store state (clean, dirty)
-- cluster protocol version.
+Assign cluster IDs to messages recovered from store, don't replicate. See Queue::recover.
-** TODO [#B] Persistent cluster support.
-Initial status protoocl to support persistent start-up (see existing code)
+** TODO [#B] Handle other ways that messages can leave a queue.
-Only one broker recovers from store, update to others.
+Other ways (other than via a consumer) that messages are take off a queue.
-Assign cluster IDs to messages recovered from store, don't replicate. See Queue::recover.
+NOTE: Not controlled by queue lock, how to make them consistent?
-** TODO [#B] Management support
-Replicate management methods that modify queues - e.g. move, purge.
Target broker may not have all messages on other brokers for purge/destroy.
- Queue::move() - need to wait for lock? Replicate?
- Queue::get() - ???
@@ -169,38 +395,66 @@ Target broker may not have all messages
Need to add callpoints & mcast messages to replicate these?
-** TODO [#B] TX transaction support.
-Extend broker::Cluster interface to capture transaction context and completion.
-Running brokers exchange TX information.
-New broker update includes TX information.
+** TODO [#B] Flow control for internal queues.
- // FIXME aconway 2010-10-18: As things stand the cluster is not
- // compatible with transactions
- // - enqueues occur after routing is complete
- // - no call to Cluster::enqueue, should be in Queue::process?
- // - no transaction context associated with messages in the Cluster interface.
- // - no call to Cluster::accept in Queue::dequeueCommitted
-
-** TODO [#B] DTX transaction support.
-Extend broker::Cluster interface to capture transaction context and completion.
-Running brokers exchange DTX information.
-New broker update includes DTX information.
-
-** TODO [#B] Async completion of accept.
-When this is fixed in the standalone broker, it should be fixed for cluster.
-
-** TODO [#B] Network partitions and quorum.
-Re-use existing implementation.
+Need to bound the size of internal queues: delivery and multicast.
+- stop polling for read on client connections when we reach a bound.
+- restart polling when we get back under it.
+
+That will stop local multicasting, we still have to deal with remote
+multicasting (note existing cluster does not do this.) Something like:
+- when over bounds multicast a flow-control event.
+- on delivery of flow-control all members stop polling to read client connections
+- when back under bounds send flow-control-end, all members resume
+- if flow-controling member dies others resume
+
+** TODO [#B] Integration with transactions.
+Do we want to replicate during transaction & replicate commit/rollback
+or replicate only on commit?
+No integration with DTX transactions.
+** TODO [#B] Make new cluster work with replication exchange.
+Possibly re-use some common logic. Replication exchange is like clustering
+except over TCP.
+** TODO [#B] Better concurrency, scalabiility on multi-cores.
+Introduce PollableQueue of operations per broker queue. Queue up mcast
+operations (enqueue, acquire, accept etc.) to be handled concurrently
+on different queue. Performance testing to verify improved scalability.
+** TODO [#C] Async completion for declare, bind, destroy queues and exchanges.
+Cluster needs to complete these asynchronously to guarantee resources
+exist across the cluster when the command completes.
** TODO [#C] Allow non-replicated exchanges, queues.
-Set qpid.replicate=false in declare arguments, set flag on Exchange, Queue objects.
+Set qpid.replicated=false in declare arguments, set flag on Exchange, Queue objects.
- save replicated status to store.
- support in management tools.
+Replicated exchange: replicate binds to replicated queues.
Replicated queue: replicate all messages.
-Replicated exchange: replicate bindings to replicated queues only.
-Configurable default? Defaults to true.
+** TODO [#C] New members joining - improved.
+
+Replicate wiring like old cluster, stall for wiring but not for
+messages. Update messages on a per-queue basis from back to front.
+
+Updater:
+- stall & push wiring: declare exchanges, queues, bindings.
+- start update iterator thread on each queue.
+- unstall and process normally while iterator threads run.
+
+Update iterator thread:
+- starts at back of updater queue, message m.
+- send update_front(q,m) to updatee and advance towards front
+- at front: send update_done(q)
+
+Updatee:
+- stall, receive wiring, lock all queues, mark queues "updating", unstall
+- update_front(q,m): push m to *front* of q
+- update_done(q): mark queue "ready"
+
+Updatee cannot take the queue consume lock for a queue that is updating.
+Updatee *can* push messages onto a queue that is updating.
+
+TODO: Is there any way to eliminate the stall for wiring?
** TODO [#C] Refactoring of common concerns.
@@ -215,40 +469,9 @@ Look for ways to capitalize on the simil
In particular QueuedEvents (async replication) strongly resembles
cluster replication, but over TCP rather than multicast.
-
+** TODO [#C] Concurrency for enqueue events.
+All enqueue events are being processed in the CPG deliver thread context which
+serializes all the work. We only need ordering on a per queue basis, can we
+enqueue in parallel on different queues and will that improve performance?
** TODO [#C] Handling immediate messages in a cluster
Include remote consumers in descision to deliver an immediate message?
-** TODO [#C] Remove old cluster hacks and workarounds
-The old cluster has workarounds in the broker code that can be removed.
-- [ ] drop code to replicate management model.
-- [ ] drop timer workarounds for TTL, management, heartbeats.
-- [ ] drop "cluster-safe assertions" in broker code.
-- [ ] drop connections, sessions, management from cluster update.
-- [ ] drop security workarounds: cluster code now operates after message decoding.
-- [ ] drop connection tracking in cluster code.
-- [ ] simper inconsistent-error handling code, no need to stall.
-** TODO [#C] Support for live upgrades.
-
-Allow brokers in a running cluster to be replaced one-by-one with a new version.
-
-The old cluster protocol was unstable because any changes in broker
-state caused changes to the cluster protocol.The new design should be
-much more stable.
-
-Points to implement:
-- Brokers should ignore unknown controls (with a warning) rather than an error.
-- Limit logging frequency for unknown control warnings.
-- Add a version number at front of every CPG message. Determines how the
- rest of the message is decoded. (allows for entirely new encodings e.g. AMQP 1.0)
-- Protocol version XML element in cluster.xml, on each control.
-- Initial status protocol to include protocol version number.
-
-** TODO [#C] Support for AMQP 1.0.
-
-* Testing
-** TODO [#A] Pass all existing cluster tests.
-Requires [[Defer and async completion of wiring commands.]]
-** TODO [#A] New cluster tests.
-Stress tests & performance benchmarks focused on changes in new cluster:
-- concurrency by queues rather than connections.
-- different handling shared queues when consuemrs are on different brokers.
Propchange: qpid/branches/QPID-2519/cpp/docs/api/
('svn:mergeinfo' removed)
Modified: qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in Fri Oct 21 01:19:00 2011
@@ -1029,7 +1029,7 @@ INCLUDE_FILE_PATTERNS =
# undefined via #undef or recursively expanded use the := operator
# instead of the = operator.
-PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN= QMFE_EXTERN=
+PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN=
# If the MACRO_EXPANSION and EXPAND_ONLY_PREDEF tags are set to YES then
# this tag can be used to specify a list of macro names that should be expanded.
Modified: qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h Fri Oct 21 01:19:00 2011
@@ -266,46 +266,7 @@
* else
* session.rollback();
* </pre>
- *
- * <h3>Exceptions</h3>
- *
- * All exceptions for the messaging API have MessagingException as
- * their base class.
-
- * A common class of exception are those related to processing
- * addresses used to create senders and/or receivers. These all have
- * AddressError as their base class.
- *
- * Where there is a syntax error in the address itself, a
- * MalformedAddress will be thrown. Where the address is valid, but
- * there is an error in interpreting (i.e. resolving) it, a
- * ResolutionError - or a sub-class of it - will be thrown. If the
- * address has assertions enabled for a given context and the asserted
- * node properties are not in fact correct then AssertionFailed will
- * be thrown. If the node is not found, NotFound will be thrown.
- *
- * The loss of the underlying connection (e.g. the TCP connection)
- * results in TransportFailure being thrown. If automatic reconnect is
- * enabled, this will be caught be the library which will then try to
- * reconnect. If reconnection - as configured by the connection
- * options - fails, then TransportFailure will be thrown. This can
- * occur on any call to the messaging API.
- *
- * Sending a message may also result in an exception
- * (e.g. TargetCapacityExceeded if a queue to which the message is
- * delivered cannot enqueue it due to lack of capacity). For
- * asynchronous send the exception may not be thrown on the send
- * invocation that actually triggers it, but on a subsequent method
- * call on the API.
- *
- * Certain exceptions may render the session invalid; once these
- * occur, subsequent calls on the session will throw the same class of
- * exception. This is not an intrinsic property of the class of
- * exception, but is a result of the current mapping of the API to the
- * underlying AMQP 0-10 protocol. You can test whether the session is
- * valid at any time using the hasError() and/or checkError() methods
- * on Session.
- *
+ *
* <h3>Logging</h3>
*
* The Qpidd broker and C++ clients can both use environment variables to
Modified: qpid/branches/QPID-2519/cpp/docs/api/footer.html
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/api/footer.html?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/footer.html (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/footer.html Fri Oct 21 01:19:00 2011
@@ -25,7 +25,7 @@ Qpid C++ API Reference</small></address>
<address style="text-align: right;">
<small>
-Generated on $date for $projectname by <a href="http://www.doxygen.org/index.html"><img src="doxygen.png" alt="doxygen" align="middle" border="0"></a> $doxygenversion</small>
+Generated on $datetime for $projectname by <a href="http://www.doxygen.org/index.html"><img src="doxygen.png" alt="doxygen" align="middle" border="0"></a> $doxygenversion</small>
</address>
</body>
</html>
Modified: qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in Fri Oct 21 01:19:00 2011
@@ -1021,7 +1021,7 @@ INCLUDE_FILE_PATTERNS =
# undefined via #undef or recursively expanded use the := operator
# instead of the = operator.
-PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN= QMFE_EXTERN=
+PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN=
# If the MACRO_EXPANSION and EXPAND_ONLY_PREDEF tags are set to YES then
# this tag can be used to specify a list of macro names that should be expanded.
Modified: qpid/branches/QPID-2519/cpp/docs/man/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/man/Makefile.am?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/man/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/docs/man/Makefile.am Fri Oct 21 01:19:00 2011
@@ -16,29 +16,10 @@
# specific language governing permissions and limitations
# under the License.
#
-
-# Generate makefile from qpidd --help
-#
-# Note: qiddd.1 is normally a _checked in_ pre-generated file, so that
-# make dist does not have to build the entire source just for the man page.
-#
-# To update the checked-in file (e.g. for a new release) do the following:
-#
-# - start with a completely clean checkout.
-# - make sure there are no modules installed in your configured prefix,
-# we don't want to pick up configuration from optional modules
-# - do bootstrap; configure
-# - in build-dir: cd src; make # build the broker
-# - in source-dir: cd docs/man; rm qpidd.1 # remove checked-in man page.
-# - in build-dir: cd docs/man; make # make new man page
-# - edit qpidd.1 to remove all default values referring to file/directory locations.
-# these values will differ between builds depending on configuration.
-# - if source-dir != build-dir: copy qpidd.1 from build-dir/docs/man to source-dir/docs/man
-
dist_man_MANS = qpidd.1
-man_aux = $(dist_man_MANS:.1=.x)
-EXTRA_DIST = $(man_aux) generate_manpage groffify_options.sed groffify_template.sed
+man_aux = $(dist_man_MANS:.1=.x)
+EXTRA_DIST = $(man_aux) generate_manpage groffify_options.sed groffify_template.sed
DISTCLEANFILES = $(dist_man_MANS)
CLEANFILES=qpidd.1
Modified: qpid/branches/QPID-2519/cpp/docs/man/qpidd.x
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/docs/man/qpidd.x?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/man/qpidd.x (original)
+++ qpid/branches/QPID-2519/cpp/docs/man/qpidd.x Fri Oct 21 01:19:00 2011
@@ -13,8 +13,6 @@ messages using the Advanced Message Queu
[OPTIONS]
-The options below are built-in to qpidd. Installing add-on modules provides additional options. To see the full set of options available type "qpidd --help"
-
Options may be specified via command line, environment variable or configuration file. See FILES and ENVIRONMENT below for details.
[FILES]
Modified: qpid/branches/QPID-2519/cpp/etc/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/etc/Makefile.am?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/etc/Makefile.am Fri Oct 21 01:19:00 2011
@@ -30,7 +30,30 @@ nobase_sysconf_DATA = \
qpidd.conf
if HAVE_SASL
+SASL_DB = qpidd.sasldb
+
nobase_sysconf_DATA += \
$(SASL_CONF)
+sasldbdir = $(localstatedir)/lib/qpidd
+sasldb_DATA = $(SASL_DB)
+
+# Setup the default sasldb file with a single user, guest, with an
+# obvious password. This user and password are the default for many
+# clients.
+#
+# The realm specified by -u is very important, and QPID is the default
+# for the broker so we use it here. The realm is important because it
+# defaults to the local hostname of the machine running the
+# broker. This may not seem to bad at first glance, but it means that
+# the sasldb has to be tailored to each machine that would be running
+# a broker, and if the machine ever changed its name the
+# authentication would stop working until the sasldb was updated. For
+# these reasons we always want the broker to specify a realm where its
+# users live, and we want the users to exist in that realm as well.
+$(SASL_DB):
+ echo guest | $(SASL_PASSWD) -c -p -f $(SASL_DB) -u QPID guest
+
+CLEANFILES=$(SASL_DB)
+
endif
Modified: qpid/branches/QPID-2519/cpp/etc/qpidd.conf
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/etc/qpidd.conf?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/qpidd.conf (original)
+++ qpid/branches/QPID-2519/cpp/etc/qpidd.conf Fri Oct 21 01:19:00 2011
@@ -21,4 +21,4 @@
#
# (Note: no spaces on either side of '='). Using default settings:
# "qpidd --help" or "man qpidd" for more details.
-cluster-mechanism=DIGEST-MD5 ANONYMOUS
+cluster-mechanism=ANONYMOUS
Modified: qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf (original)
+++ qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf Fri Oct 21 01:19:00 2011
@@ -17,8 +17,8 @@
# under the License.
#
#
-# This configuation allows for either SASL ANONYMOUS or DIGEST-MD5
-# authentication. The DIGEST-MD5 authentication is done on a
+# This configuation allows for either SASL PLAIN or ANONYMOUS
+# authentication. The PLAIN authentication is done on a
# username+password, which is stored in the sasldb_path
# file. Usernames and passwords can be added to the file using the
# command:
@@ -39,7 +39,6 @@
pwcheck_method: auxprop
auxprop_plugin: sasldb
sasldb_path: /var/lib/qpidd/qpidd.sasldb
-mech_list: DIGEST-MD5 ANONYMOUS
#following line stops spurious 'sql_select option missing' errors when
#cyrus-sql-sasl plugin is installed
Modified: qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt (original)
+++ qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt Fri Oct 21 01:19:00 2011
@@ -77,14 +77,25 @@ macro(add_example subdir example)
endmacro(add_example)
install (FILES ${CMAKE_CURRENT_SOURCE_DIR}/README.txt
+ ${CMAKE_CURRENT_SOURCE_DIR}/README.verify
+ ${CMAKE_CURRENT_SOURCE_DIR}/verify
+ ${CMAKE_CURRENT_SOURCE_DIR}/verify_all
DESTINATION ${QPID_INSTALL_EXAMPLESDIR}
COMPONENT ${QPID_COMPONENT_EXAMPLES})
if (MSVC)
install (FILES ${CMAKE_CURRENT_SOURCE_DIR}/examples.sln
+ ${CMAKE_CURRENT_SOURCE_DIR}/old-examples.sln
DESTINATION ${QPID_INSTALL_EXAMPLESDIR}
COMPONENT ${QPID_COMPONENT_EXAMPLES})
endif (MSVC)
+add_subdirectory(direct)
+add_subdirectory(failover)
+add_subdirectory(fanout)
+add_subdirectory(pub-sub)
+#add_subdirectory(qmf-agent)
add_subdirectory(qmf-console)
+add_subdirectory(request-response)
+add_subdirectory(tradedemo)
+add_subdirectory(xml-exchange)
add_subdirectory(messaging)
-add_subdirectory(old_api)
Modified: qpid/branches/QPID-2519/cpp/examples/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/Makefile.am?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/examples/Makefile.am Fri Oct 21 01:19:00 2011
@@ -16,7 +16,15 @@
# specific language governing permissions and limitations
# under the License.
#
-SUBDIRS = qmf-console messaging old_api
+SUBDIRS = direct fanout pub-sub request-response failover qmf-console tradedemo messaging
+if HAVE_XML
+ SUBDIRS += xml-exchange
+ broker_args = "--no-module-dir --data-dir \"\" --auth no --load-module $(top_builddir)/src/.libs/xml.so"
+endif
+if !HAVE_XML
+ exclude_examples_regexp="xml" # Exclude XML examples.
+ broker_args = "--no-module-dir --data-dir \"\" --auth no"
+endif
MAKEDIST=.libs/Makefile
@@ -29,9 +37,13 @@ $(MAKEDIST): Makefile
examplesdir=$(pkgdatadir)/examples
dist_examples_DATA = README.txt $(MAKEDIST)
-EXTRA_DIST = examples.sln CMakeLists.txt
+EXTRA_DIST = README.verify verify verify_all examples.sln CMakeLists.txt
# For older versions of automake
abs_top_srcdir = @abs_top_srcdir@
abs_top_builddir = @abs_top_builddir@
+# Verify the examples in the buid tree.
+check-local:
+ $(srcdir)/verify_all $(abs_top_srcdir)/.. $(abs_top_builddir) $(broker_args) $(exclude_examples_regexp)
+
Propchange: qpid/branches/QPID-2519/cpp/examples/direct/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,5 @@
+declare_queues
+direct_producer
+listener
+Makefile.in
+Makefile
Modified: qpid/branches/QPID-2519/cpp/examples/examples.sln
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/examples.sln?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/examples.sln (original)
+++ qpid/branches/QPID-2519/cpp/examples/examples.sln Fri Oct 21 01:19:00 2011
@@ -32,14 +32,6 @@ Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C9
EndProject
Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "messaging_spout", "messaging\messaging_spout.vcproj", "{D3115AC9-91C4-4D79-BCAC-DE837C70F1EA}"
EndProject
-Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "qmf_console_console", "qmf-console\qmf-console_console.vcproj", "{490473E1-FECA-1BAD-2E13-3FFA2B8669C3}"
-EndProject
-Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "qmf_console_ping", "qmf-console\qmf-console_ping.vcproj", "{C1FFDE95-3442-49AE-9985-7EEE3D45B4A3}"
-EndProject
-Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "qmf_console_printevents", "qmf-console\qmf-console_printevents.vcproj", "{72C74624-FECA-1BAD-2E13-3FFA2B8669C3}"
-EndProject
-Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "qmf_console_queuestats", "qmf-console\qmf-console_queuestats.vcproj", "{B21825EA-FECA-1BAD-2E13-3FFA2B8669C3}"
-EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Win32 = Debug|Win32
@@ -96,22 +88,6 @@ Global
{D3115AC9-91C4-4D79-BCAC-DE837C70F1EA}.Release|Win32.Build.0 = Release|Win32
{D3115AC9-91C4-4D79-BCAC-DE837C70F1EA}.Release|x64.ActiveCfg = Release|x64
{D3115AC9-91C4-4D79-BCAC-DE837C70F1EA}.Release|x64.Build.0 = Release|x64
- {490473E1-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.ActiveCfg = Debug|Win32
- {490473E1-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.Build.0 = Debug|Win32
- {490473E1-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.ActiveCfg = Release|Win32
- {490473E1-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.Build.0 = Release|Win32
- {C1FFDE95-3442-49AE-9985-7EEE3D45B4A3}.Debug|Win32.ActiveCfg = Debug|Win32
- {C1FFDE95-3442-49AE-9985-7EEE3D45B4A3}.Debug|Win32.Build.0 = Debug|Win32
- {C1FFDE95-3442-49AE-9985-7EEE3D45B4A3}.Release|Win32.ActiveCfg = Release|Win32
- {C1FFDE95-3442-49AE-9985-7EEE3D45B4A3}.Release|Win32.Build.0 = Release|Win32
- {72C74624-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.ActiveCfg = Debug|Win32
- {72C74624-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.Build.0 = Debug|Win32
- {72C74624-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.ActiveCfg = Release|Win32
- {72C74624-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.Build.0 = Release|Win32
- {B21825EA-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.ActiveCfg = Debug|Win32
- {B21825EA-FECA-1BAD-2E13-3FFA2B8669C3}.Debug|Win32.Build.0 = Debug|Win32
- {B21825EA-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.ActiveCfg = Release|Win32
- {B21825EA-FECA-1BAD-2E13-3FFA2B8669C3}.Release|Win32.Build.0 = Release|Win32
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
Propchange: qpid/branches/QPID-2519/cpp/examples/failover/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,7 @@
+Makefile.in
+listener
+Makefile
+declare_queues
+direct_producer
+resuming_receiver
+replaying_sender
Propchange: qpid/branches/QPID-2519/cpp/examples/fanout/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,5 @@
+declare_queues
+fanout_producer
+listener
+Makefile.in
+Makefile
Modified: qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp Fri Oct 21 01:19:00 2011
@@ -45,12 +45,12 @@ struct Options : OptionParser
url("127.0.0.1"),
timeout(0),
forever(false),
- count(0)
+ count(1)
{
add("broker,b", url, "url of broker to connect to");
add("timeout,t", timeout, "timeout in seconds to wait before exiting");
add("forever,f", forever, "ignore timeout and wait forever");
- add("connection-options", connectionOptions, "connection options string in the form {name1:value1, name2:value2}");
+ add("connection-options", connectionOptions, "connection options string in the form {name1=value1, name2=value2}");
add("count,c", count, "number of messages to read before exiting");
}
Modified: qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp Fri Oct 21 01:19:00 2011
@@ -39,8 +39,8 @@ using std::string;
int main(int argc, char** argv) {
const char* url = argc>1 ? argv[1] : "amqp:tcp:127.0.0.1:5672";
- std::string connectionOptions = argc > 2 ? argv[2] : "";
-
+ std::string connectionOptions = argc > 3 ? argv[3] : "";
+
Connection connection(url, connectionOptions);
try {
connection.open();
Modified: qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp Fri Oct 21 01:19:00 2011
@@ -65,7 +65,7 @@ struct Options : OptionParser
add("property,P", properties, "specify message property");
add("map,M", entries, "specify entry for map content");
add("content", content, "specify textual content");
- add("connection-options", connectionOptions, "connection options string in the form {name1:value1, name2:value2}");
+ add("connection-options", connectionOptions, "connection options string in the form {name1=value1, name2=value2}");
}
static bool nameval(const std::string& in, std::string& name, std::string& value)
Propchange: qpid/branches/QPID-2519/cpp/examples/pub-sub/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,5 @@
+topic_config_queues
+topic_listener
+topic_publisher
+Makefile.in
+Makefile
Modified: qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp Fri Oct 21 01:19:00 2011
@@ -31,7 +31,9 @@ using namespace qpid::console;
int main_int(int /*argc*/, char** /*argv*/)
{
//
- // Declare connection settings for the messaging broker.
+ // Declare connection settings for the messaging broker. The settings default to
+ // localhost:5672 with user guest (password guest). Refer to the header file
+ // <qpid/client/ConnectionSettings.h> for full details.
//
qpid::client::ConnectionSettings connSettings;
Modified: qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp Fri Oct 21 01:19:00 2011
@@ -64,7 +64,9 @@ struct Main {
Listener listener;
//
- // Declare connection settings for the messaging broker.
+ // Declare connection settings for the messaging broker. The settings default to
+ // localhost:5672 with user guest (password guest). Refer to the header file
+ // <qpid/client/ConnectionSettings.h> for full details.
//
qpid::client::ConnectionSettings connSettings;
Propchange: qpid/branches/QPID-2519/cpp/examples/request-response/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,4 @@
+client
+server
+Makefile.in
+Makefile
Propchange: qpid/branches/QPID-2519/cpp/examples/tradedemo/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,6 @@
+Makefile.in
+Makefile
+topic_listener
+declare_queues
+topic_publisher
+
Propchange: qpid/branches/QPID-2519/cpp/examples/xml-exchange/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Fri Oct 21 01:19:00 2011
@@ -0,0 +1,5 @@
+declare_queues
+listener
+xml_producer
+Makefile.in
+Makefile
Modified: qpid/branches/QPID-2519/cpp/include/qmf/Agent.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/include/qmf/Agent.h?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/include/qmf/Agent.h (original)
+++ qpid/branches/QPID-2519/cpp/include/qmf/Agent.h Fri Oct 21 01:19:00 2011
@@ -42,7 +42,7 @@ namespace qmf {
class SchemaId;
class Schema;
- class QMF_CLASS_EXTERN Agent : public qmf::Handle<AgentImpl> {
+ class Agent : public qmf::Handle<AgentImpl> {
public:
QMF_EXTERN Agent(AgentImpl* impl = 0);
QMF_EXTERN Agent(const Agent&);
Modified: qpid/branches/QPID-2519/cpp/include/qmf/AgentEvent.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/include/qmf/AgentEvent.h?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/include/qmf/AgentEvent.h (original)
+++ qpid/branches/QPID-2519/cpp/include/qmf/AgentEvent.h Fri Oct 21 01:19:00 2011
@@ -46,7 +46,7 @@ namespace qmf {
AGENT_THREAD_FAILED = 8
};
- class QMF_CLASS_EXTERN AgentEvent : public qmf::Handle<AgentEventImpl> {
+ class AgentEvent : public qmf::Handle<AgentEventImpl> {
public:
QMF_EXTERN AgentEvent(AgentEventImpl* impl = 0);
QMF_EXTERN AgentEvent(const AgentEvent&);
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org