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/20 20:43:26 UTC
svn commit: r1186990 [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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/python/python.i Thu Oct 20 18:42:46 2011
@@ -21,21 +21,357 @@
%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(PyExc_RuntimeError, error.c_str());
+ PyErr_SetString(pExceptionType, 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/qpid.i Thu Oct 20 18:42:46 2011
@@ -27,6 +27,7 @@
#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
@@ -48,6 +49,7 @@ 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>
@@ -56,6 +58,7 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/bindings/qpid/ruby/Makefile.am Thu Oct 20 18:42:46 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 = ruby.i
+EXTRA_DIST = CMakeLists.txt 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 -shrext ".$(RUBY_DLEXT)"
+cqpid_la_LDFLAGS = -avoid-version -module -shared -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)
+cqpid_la_CXXFLAGS = $(INCLUDES) -I$(RUBY_INC) -I$(RUBY_INC_ARCH) -fno-strict-aliasing
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/swig_python_typemaps.i Thu Oct 20 18:42:46 2011
@@ -17,6 +17,25 @@
* 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)
@@ -28,6 +47,7 @@ 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*);
@@ -104,6 +124,9 @@ 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) {
@@ -143,6 +166,30 @@ 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);
@@ -304,6 +351,15 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i (original)
+++ qpid/branches/QPID-2519/cpp/bindings/swig_ruby_typemaps.i Thu Oct 20 18:42:46 2011
@@ -49,7 +49,7 @@
}
VALUE VariantToRb(const qpid::types::Variant* v) {
- VALUE result;
+ VALUE result = Qnil;
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/bld-winsdk.ps1 (original)
+++ qpid/branches/QPID-2519/cpp/bld-winsdk.ps1 Thu Oct 20 18:42:46 2011
@@ -186,9 +186,6 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/configure.ac (original)
+++ qpid/branches/QPID-2519/cpp/configure.ac Thu Oct 20 18:42:46 2011
@@ -68,8 +68,10 @@ 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
- gl_COMPILER_FLAGS(-Werror)
+ COMPILER_FLAGS="-Werror"
gl_COMPILER_FLAGS(-pedantic)
gl_COMPILER_FLAGS(-Wall)
gl_COMPILER_FLAGS(-Wextra)
@@ -521,18 +523,19 @@ 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/direct/Makefile
- examples/fanout/Makefile
- examples/pub-sub/Makefile
- examples/request-response/Makefile
- examples/failover/Makefile
- examples/xml-exchange/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/qmf-console/Makefile
- examples/tradedemo/Makefile
+ examples/old_api/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=1186990&r1=1186989&r2=1186990&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 Thu Oct 20 18:42:46 2011
@@ -17,7 +17,6 @@
# under the License.
* A new design for Qpid clustering.
-
** Issues with current design.
The cluster is based on virtual synchrony: each broker multicasts
@@ -95,8 +94,9 @@ Use a moving queue ownership protocol to
No longer relies on identical state and lock-step behavior to cause
identical dequeues on each broker.
-Each queue has an associated thread-context. Events for a queue are executed
-in that queues context, in parallel with events for other queues.
+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.
*** 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,19 +162,32 @@ 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.
-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.
+An inconsistent error means that after multicasting an enqueue, accept
+or dequeue, some brokers succeed in processing it and others fail.
-#TODO: The only source of dequeue errors is probably an unrecoverable journal failure.
+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.
** Updating new members
@@ -193,60 +206,44 @@ catch up (which is not guaranteed to hap
With the new cluster design only exchanges, queues, bindings and
messages need to be replicated.
-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.)
+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)
The cluster interface captures these events:
- wiring changes: queue/exchange declare/bind
- message enqueued/acquired/released/rejected/dequeued.
-
-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.
+- transactional events.
** Maintainability
@@ -273,106 +270,48 @@ A number of specific ways the code will
** Performance
-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.
+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.
-* Design debates/descisions
+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
** Active/active vs. active passive
An active-active cluster can be used in an active-passive mode. In
@@ -385,7 +324,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 hast to take over if primary fails.
+- Exactly one broker has 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.
@@ -393,43 +332,17 @@ 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.
-- 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.
+- As long as a client can connect to any broker in the cluster, it can be served.
Active/passive benefits:
-- Don't need message allocation strategy, can feed consumers at top speed.
-- Code may be simpler than active/active.
+- Don't need to replicate message allocation, can feed consumers at top speed.
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=1186990&r1=1186989&r2=1186990&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 Thu Oct 20 18:42:46 2011
@@ -17,376 +17,150 @@
# specific language governing permissions and limitations
# under the License.
+* Status of impementation
-Notes on new cluster implementation. See also: new-cluster-design.txt
+Meaning of priorities:
+[#A] Essential for basic functioning.
+[#B] Required for first release.
+[#C] Can be addressed in a later release.
-* Implementation plan.
+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.
-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.
+Prototype code is on branch qpid-2920-active, in cpp/src/qpid/cluster/exp/
-Test: enhance test_store.cpp to defer enqueueComplete till special message received.
+** Similarities to existing cluster.
-Async callback uses *requestIOProcessing* to queue action on IO thread.
+/Active-active/: the new cluster can be a drop-in replacement for the
+old, existing tests & customer deployment configurations are still
+valid.
-** 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.
+/Virtual synchrony/: Uses corosync to co-ordinate activity of members.
-** TODO [#B] Replicate wiring.
-From messageStore create/destroy/bind, replicate encoded declare/destroy/bind command.
+/XML controls/: Uses XML to define the primitives multicast to the
+cluster.
-** TODO [#B] New members joining - first pass
+** Differences with existing cluster.
-Re-use update code from old cluster but don't replicate sessions &
-connections.
+/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.
-Need to extend it to send cluster IDs with messages.
+/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.
-Need to replicate the queue ownership data as part of the update.
+/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.
-** TODO [#B] Persistence support.
-InitialStatus protoocl etc. to support persistent start-up (existing code)
+* Completed tasks
+** DONE [#A] Minimal POC: publish/acquire/dequeue protocol.
+ CLOSED: [2011-10-05 Wed 16:03]
-Only one broker recovers from store, update to others.
+Defines broker::Cluster interface and call points.
+Initial interface commite
-Assign cluster IDs to messages recovered from store, don't replicate. See Queue::recover.
+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.
+
+** TODO [#A] Update to new members joining.
+
+Need to resolve [[Queue sequence numbers vs. independant message IDs]] first.
+- implicit sequence numbers are more tricky to replicate to new member.
+
+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)
+
+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] Handle other ways that messages can leave a queue.
+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.
-Other ways (other than via a consumer) that messages are take off a queue.
+Exchange updater:
+- updater: send snapshot of exchange as it was at the sync point.
-NOTE: Not controlled by queue lock, how to make them consistent?
+Exchange updatee:
+- queue exchange operations after the sync point.
+- when snapshot is received: apply saved operations.
+Updater remains active throughout.
+Updatee stalls clients until the update completes.
+
+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 [#A] Failover updates to client.
+Implement the amq.failover exchange to notify clients of membership.
+
+** 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.
+
+** TODO [#B] Persistent cluster support.
+Initial status protoocl to support persistent start-up (see existing code)
+
+Only one broker recovers from store, update to others.
+
+Assign cluster IDs to messages recovered from store, don't replicate. See Queue::recover.
+
+** 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() - ???
@@ -395,66 +169,38 @@ Target broker may not have all messages
Need to add callpoints & mcast messages to replicate these?
-** TODO [#B] Flow control for internal queues.
+** 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.
-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.
+ // 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.
** TODO [#C] Allow non-replicated exchanges, queues.
-Set qpid.replicated=false in declare arguments, set flag on Exchange, Queue objects.
+Set qpid.replicate=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.
-** 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?
+Configurable default? Defaults to true.
** TODO [#C] Refactoring of common concerns.
@@ -469,9 +215,40 @@ 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 (added)
+++ svn:mergeinfo Thu Oct 20 18:42:46 2011
@@ -0,0 +1,8 @@
+/qpid/branches/0.10/qpid/cpp/docs/api:1103088
+/qpid/branches/0.5.x-dev/qpid/cpp/docs/api:892761,894875
+/qpid/branches/0.6-release-windows-installer/cpp/docs/api:926803
+/qpid/branches/0.6-release-windows-installer/qpid/cpp/docs/api:926803,927233
+/qpid/branches/java-network-refactor/qpid/cpp/docs/api:805429-825319
+/qpid/branches/qpid-2935/qpid/cpp/docs/api:1061302-1072333
+/qpid/branches/qpid-3346/qpid/cpp/docs/api:1144319-1179855
+/qpid/trunk/qpid/cpp/docs/api:1072051-1185907
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/developer.doxygen.in Thu Oct 20 18:42:46 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=
+PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN= QMFE_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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/doxygen_mainpage.h Thu Oct 20 18:42:46 2011
@@ -266,7 +266,46 @@
* 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/footer.html (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/footer.html Thu Oct 20 18:42:46 2011
@@ -25,7 +25,7 @@ Qpid C++ API Reference</small></address>
<address style="text-align: right;">
<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>
+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>
</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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in (original)
+++ qpid/branches/QPID-2519/cpp/docs/api/user.doxygen.in Thu Oct 20 18:42:46 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=
+PREDEFINED = QPID_CLIENT_EXTERN= QPID_COMMON_EXTERN= QPID_CONSOLE_EXTERN= QPID_BROKER_EXTERN= QPID_MESSAGING_EXTERN= QMF_EXTERN= QMFE_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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/man/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/docs/man/Makefile.am Thu Oct 20 18:42:46 2011
@@ -16,10 +16,29 @@
# 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/docs/man/qpidd.x (original)
+++ qpid/branches/QPID-2519/cpp/docs/man/qpidd.x Thu Oct 20 18:42:46 2011
@@ -13,6 +13,8 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/etc/Makefile.am Thu Oct 20 18:42:46 2011
@@ -30,30 +30,7 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/qpidd.conf (original)
+++ qpid/branches/QPID-2519/cpp/etc/qpidd.conf Thu Oct 20 18:42:46 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=ANONYMOUS
+cluster-mechanism=DIGEST-MD5 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf (original)
+++ qpid/branches/QPID-2519/cpp/etc/sasl2/qpidd.conf Thu Oct 20 18:42:46 2011
@@ -17,8 +17,8 @@
# under the License.
#
#
-# This configuation allows for either SASL PLAIN or ANONYMOUS
-# authentication. The PLAIN authentication is done on a
+# This configuation allows for either SASL ANONYMOUS or DIGEST-MD5
+# authentication. The DIGEST-MD5 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,6 +39,7 @@
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt (original)
+++ qpid/branches/QPID-2519/cpp/examples/CMakeLists.txt Thu Oct 20 18:42:46 2011
@@ -77,25 +77,14 @@ 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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/Makefile.am (original)
+++ qpid/branches/QPID-2519/cpp/examples/Makefile.am Thu Oct 20 18:42:46 2011
@@ -16,15 +16,7 @@
# specific language governing permissions and limitations
# under the License.
#
-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
+SUBDIRS = qmf-console messaging old_api
MAKEDIST=.libs/Makefile
@@ -37,13 +29,9 @@ $(MAKEDIST): Makefile
examplesdir=$(pkgdatadir)/examples
dist_examples_DATA = README.txt $(MAKEDIST)
-EXTRA_DIST = README.verify verify verify_all examples.sln CMakeLists.txt
+EXTRA_DIST = 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)
-
Modified: qpid/branches/QPID-2519/cpp/examples/examples.sln
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/cpp/examples/examples.sln?rev=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/examples.sln (original)
+++ qpid/branches/QPID-2519/cpp/examples/examples.sln Thu Oct 20 18:42:46 2011
@@ -32,6 +32,14 @@ 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
@@ -88,6 +96,22 @@ 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
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/drain.cpp Thu Oct 20 18:42:46 2011
@@ -45,12 +45,12 @@ struct Options : OptionParser
url("127.0.0.1"),
timeout(0),
forever(false),
- count(1)
+ count(0)
{
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/server.cpp Thu Oct 20 18:42:46 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 > 3 ? argv[3] : "";
-
+ std::string connectionOptions = argc > 2 ? argv[2] : "";
+
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/messaging/spout.cpp Thu Oct 20 18:42:46 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/old_api/direct/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,5 @@
+declare_queues
+direct_producer
+listener
+Makefile.in
+Makefile
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/failover/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,7 @@
+Makefile.in
+listener
+Makefile
+declare_queues
+direct_producer
+resuming_receiver
+replaying_sender
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/fanout/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,5 @@
+declare_queues
+fanout_producer
+listener
+Makefile.in
+Makefile
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/pub-sub/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,5 @@
+topic_config_queues
+topic_listener
+topic_publisher
+Makefile.in
+Makefile
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/request-response/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,4 @@
+client
+server
+Makefile.in
+Makefile
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/tradedemo/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,6 @@
+Makefile.in
+Makefile
+topic_listener
+declare_queues
+topic_publisher
+
Propchange: qpid/branches/QPID-2519/cpp/examples/old_api/xml-exchange/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Thu Oct 20 18:42:46 2011
@@ -0,0 +1,5 @@
+declare_queues
+listener
+xml_producer
+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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/qmf-console/ping.cpp Thu Oct 20 18:42:46 2011
@@ -31,9 +31,7 @@ using namespace qpid::console;
int main_int(int /*argc*/, char** /*argv*/)
{
//
- // 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.
+ // Declare connection settings for the messaging broker.
//
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp (original)
+++ qpid/branches/QPID-2519/cpp/examples/qmf-console/printevents.cpp Thu Oct 20 18:42:46 2011
@@ -64,9 +64,7 @@ struct Main {
Listener listener;
//
- // 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.
+ // Declare connection settings for the messaging broker.
//
qpid::client::ConnectionSettings connSettings;
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=1186990&r1=1186989&r2=1186990&view=diff
==============================================================================
--- qpid/branches/QPID-2519/cpp/include/qmf/Agent.h (original)
+++ qpid/branches/QPID-2519/cpp/include/qmf/Agent.h Thu Oct 20 18:42:46 2011
@@ -42,7 +42,7 @@ namespace qmf {
class SchemaId;
class Schema;
- class Agent : public qmf::Handle<AgentImpl> {
+ class QMF_CLASS_EXTERN Agent : public qmf::Handle<AgentImpl> {
public:
QMF_EXTERN Agent(AgentImpl* impl = 0);
QMF_EXTERN Agent(const Agent&);
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org