You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2015/01/27 16:00:16 UTC
svn commit: r1655057 [2/8] - in /qpid/branches/QPID-6262-JavaBrokerNIO: ./
qpid/ qpid/cpp/examples/messaging/ qpid/cpp/src/
qpid/cpp/src/qpid/amqp_0_10/ qpid/cpp/src/qpid/broker/
qpid/cpp/src/qpid/broker/amqp/ qpid/cpp/src/qpid/client/amqp0_10/ qpid/cp...
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp Tue Jan 27 15:00:13 2015
@@ -73,6 +73,7 @@ class ProtocolImpl : public BrokerContex
qpid::sys::ConnectionCodec* create(const qpid::framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> translate(const qpid::broker::Message&);
boost::shared_ptr<RecoverableMessage> recover(qpid::framing::Buffer&);
+ qpid::framing::ProtocolVersion supportedVersion() const;
private:
};
@@ -158,5 +159,10 @@ boost::shared_ptr<RecoverableMessage> Pr
}
}
+qpid::framing::ProtocolVersion ProtocolImpl::supportedVersion() const
+{
+ return qpid::framing::ProtocolVersion(1,0);
+}
+
}}} // namespace qpid::broker::amqp
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.cpp Tue Jan 27 15:00:13 2015
@@ -23,6 +23,7 @@
#include "qpid/log/Statement.h"
#include <algorithm>
#include <string.h>
+#include "config.h"
namespace qpid {
namespace broker {
@@ -126,7 +127,13 @@ bool OutgoingFromRelay::doWork()
{
relay->check();
relay->setCredit(pn_link_credit(link));
- return relay->send(link);
+ bool worked = relay->send(link);
+ pn_delivery_t *d = pn_link_current(link);
+ if (d && pn_delivery_writable(d)) {
+ handle(d);
+ return true;
+ }
+ return worked;
}
/**
* Called when a delivery is writable
@@ -163,7 +170,7 @@ void OutgoingFromRelay::handle(pn_delive
/**
* Signals that this link has been detached
*/
-void OutgoingFromRelay::detached()
+void OutgoingFromRelay::detached(bool /*closed*/)
{
relay->detached(this);
}
@@ -221,7 +228,7 @@ uint32_t IncomingToRelay::getCredit()
return relay->getCredit();
}
-void IncomingToRelay::detached()
+void IncomingToRelay::detached(bool /*closed*/)
{
relay->detached(this);
}
@@ -238,7 +245,11 @@ void BufferedTransfer::initIn(pn_link_t*
//copy delivery tag
pn_delivery_tag_t dt = pn_delivery_tag(d);
tag.resize(dt.size);
+#ifdef NO_PROTON_DELIVERY_TAG_T
+ ::memmove(&tag[0], dt.start, dt.size);
+#else
::memmove(&tag[0], dt.bytes, dt.size);
+#endif
//set context
pn_delivery_set_context(d, this);
@@ -258,7 +269,11 @@ bool BufferedTransfer::settle()
void BufferedTransfer::initOut(pn_link_t* link)
{
pn_delivery_tag_t dt;
+#ifdef NO_PROTON_DELIVERY_TAG_T
+ dt.start = &tag[0];
+#else
dt.bytes = &tag[0];
+#endif
dt.size = tag.size();
out.handle = pn_delivery(link, dt);
//set context
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Relay.h Tue Jan 27 15:00:13 2015
@@ -100,7 +100,7 @@ class OutgoingFromRelay : public Outgoin
const std::string& target, const std::string& name, boost::shared_ptr<Relay>);
bool doWork();
void handle(pn_delivery_t* delivery);
- void detached();
+ void detached(bool closed);
void init();
void setSubjectFilter(const std::string&);
void setSelectorFilter(const std::string&);
@@ -118,7 +118,7 @@ class IncomingToRelay : public Incoming
bool settle();
bool doWork();
bool haveWork();
- void detached();
+ void detached(bool closed);
void readable(pn_delivery_t* delivery);
uint32_t getCredit();
private:
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Session.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Session.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Session.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/broker/amqp/Session.cpp Tue Jan 27 15:00:13 2015
@@ -48,6 +48,7 @@
#include "qpid/framing/MessageTransferBody.h"
#include "qpid/log/Statement.h"
#include "qpid/amqp_0_10/Codecs.h"
+#include "config.h"
#include <boost/intrusive_ptr.hpp>
#include <boost/format.hpp>
#include <map>
@@ -505,10 +506,11 @@ void Session::setupOutgoing(pn_link_t* l
if (!settings.autodelete) settings.autodelete = autodelete;
altExchange = node.topic->getAlternateExchange();
}
- if (!settings.autoDeleteDelay) {
+ if (settings.original.find("qpid.auto_delete_timeout") == settings.original.end()) {
//only use delay from link if policy didn't specify one
settings.autoDeleteDelay = pn_terminus_get_timeout(source);
- settings.original["qpid.auto_delete_timeout"] = settings.autoDeleteDelay;
+ if (settings.autoDeleteDelay)
+ settings.original["qpid.auto_delete_timeout"] = settings.autoDeleteDelay;
}
if (settings.autoDeleteDelay) {
settings.autodelete = true;
@@ -577,7 +579,7 @@ void Session::detach(pn_link_t* link)
if (pn_link_is_sender(link)) {
OutgoingLinks::iterator i = outgoing.find(link);
if (i != outgoing.end()) {
- i->second->detached();
+ i->second->detached(true/*TODO: checked whether actually closed; see PROTON-773*/);
boost::shared_ptr<Queue> q = OutgoingFromQueue::getExclusiveSubscriptionQueue(i->second.get());
if (q && !q->isAutoDelete() && !q->isDeleted()) {
connection.getBroker().deleteQueue(q->getName(), connection.getUserId(), connection.getMgmtId());
@@ -588,7 +590,7 @@ void Session::detach(pn_link_t* link)
} else {
IncomingLinks::iterator i = incoming.find(link);
if (i != incoming.end()) {
- i->second->detached();
+ i->second->detached(true/*TODO: checked whether actually closed; see PROTON-773*/);
incoming.erase(i);
QPID_LOG(debug, "Incoming link detached");
}
@@ -615,7 +617,11 @@ void Session::accepted(pn_delivery_t* de
void Session::readable(pn_link_t* link, pn_delivery_t* delivery)
{
pn_delivery_tag_t tag = pn_delivery_tag(delivery);
+#ifdef NO_PROTON_DELIVERY_TAG_T
+ QPID_LOG(debug, "received delivery: " << std::string(tag.start, tag.size));
+#else
QPID_LOG(debug, "received delivery: " << std::string(tag.bytes, tag.size));
+#endif
incomingMessageReceived();
IncomingLinks::iterator target = incoming.find(link);
if (target == incoming.end()) {
@@ -653,7 +659,7 @@ bool Session::dispatch()
pn_condition_set_name(error, e.symbol());
pn_condition_set_description(error, e.what());
pn_link_close(s->first);
- s->second->detached();
+ s->second->detached(true);
outgoing.erase(s++);
output = true;
}
@@ -678,7 +684,7 @@ bool Session::dispatch()
pn_condition_set_name(error, e.symbol());
pn_condition_set_description(error, e.what());
pn_link_close(i->first);
- i->second->detached();
+ i->second->detached(true);
incoming.erase(i++);
output = true;
}
@@ -690,10 +696,10 @@ bool Session::dispatch()
void Session::close()
{
for (OutgoingLinks::iterator i = outgoing.begin(); i != outgoing.end(); ++i) {
- i->second->detached();
+ i->second->detached(false);
}
for (IncomingLinks::iterator i = incoming.begin(); i != incoming.end(); ++i) {
- i->second->detached();
+ i->second->detached(false);
}
outgoing.clear();
incoming.clear();
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp Tue Jan 27 15:00:13 2015
@@ -55,6 +55,7 @@ using qpid::messaging::AssertionFailed;
using qpid::framing::ExchangeBoundResult;
using qpid::framing::ExchangeQueryResult;
using qpid::framing::FieldTable;
+using qpid::framing::FieldValue;
using qpid::framing::QueueQueryResult;
using qpid::framing::ReplyTo;
using qpid::framing::Uuid;
@@ -140,6 +141,11 @@ const std::string PREFIX_AMQ("amq.");
const std::string PREFIX_QPID("qpid.");
const Verifier verifier;
+
+bool areEquivalent(const FieldValue& a, const FieldValue& b)
+{
+ return ((a == b) || (a.convertsTo<int64_t>() && b.convertsTo<int64_t>() && a.get<int64_t>() == b.get<int64_t>()));
+}
}
struct Binding
@@ -534,19 +540,19 @@ Subscription::Subscription(const Address
reliable(durable ? !AddressResolution::is_unreliable(address) : AddressResolution::is_reliable(address)),
actualType(type.empty() ? (specifiedType.empty() ? TOPIC_EXCHANGE : specifiedType) : type),
exclusiveQueue((Opt(address)/LINK/X_DECLARE/EXCLUSIVE).asBool(true)),
- autoDeleteQueue((Opt(address)/LINK/X_DECLARE/AUTO_DELETE).asBool(true)),
+ autoDeleteQueue((Opt(address)/LINK/X_DECLARE/AUTO_DELETE).asBool(!(durable || reliable))),
exclusiveSubscription((Opt(address)/LINK/X_SUBSCRIBE/EXCLUSIVE).asBool(exclusiveQueue)),
alternateExchange((Opt(address)/LINK/X_DECLARE/ALTERNATE_EXCHANGE).str())
{
- const Variant* timeout = (Opt(address)/LINK/TIMEOUT).value;
- if (timeout) {
+
+ if ((Opt(address)/LINK).hasKey(TIMEOUT)) {
+ const Variant* timeout = (Opt(address)/LINK/TIMEOUT).value;
if (timeout->asUint32()) queueOptions.setInt("qpid.auto_delete_timeout", timeout->asUint32());
- } else if (durable && !(Opt(address)/LINK/RELIABILITY).value) {
- //if durable but not explicitly reliable, then set a non-zero
- //default for the autodelete timeout (previously this would
- //have defaulted to autodelete immediately anyway, so the risk
- //of the change causing problems is mitigated)
- queueOptions.setInt("qpid.auto_delete_delay", 15*60);
+ } else if (durable && !reliable && !(Opt(address)/LINK/X_DECLARE).hasKey(AUTO_DELETE)) {
+ //if durable but not reliable, and auto-delete not
+ //explicitly set, then set a non-zero default for the
+ //autodelete timeout
+ queueOptions.setInt("qpid.auto_delete_timeout", 2*60);
}
(Opt(address)/LINK/X_DECLARE/ARGUMENTS).collect(queueOptions);
(Opt(address)/LINK/X_SUBSCRIBE/ARGUMENTS).collect(subscriptionOptions);
@@ -609,7 +615,7 @@ void Subscription::subscribe(qpid::clien
//create subscription queue:
session.queueDeclare(arg::queue=queue, arg::exclusive=exclusiveQueue,
- arg::autoDelete=autoDeleteQueue && (!(durable || reliable)), arg::durable=durable,
+ arg::autoDelete=autoDeleteQueue, arg::durable=durable,
arg::alternateExchange=alternateExchange,
arg::arguments=queueOptions);
//'default' binding:
@@ -806,7 +812,7 @@ void Queue::checkAssert(qpid::client::As
FieldTable::ValuePtr v = result.getArguments().get(i->first);
if (!v) {
throw AssertionFailed((boost::format("Option %1% not set for %2%") % i->first % name).str());
- } else if (*i->second != *v) {
+ } else if (!areEquivalent(*i->second, *v)) {
throw AssertionFailed((boost::format("Option %1% does not match for %2%, expected %3%, got %4%")
% i->first % name % *(i->second) % *v).str());
}
@@ -906,7 +912,7 @@ void Exchange::checkAssert(qpid::client:
FieldTable::ValuePtr v = result.getArguments().get(i->first);
if (!v) {
throw AssertionFailed((boost::format("Option %1% not set for %2%") % i->first % name).str());
- } else if (*i->second != *v) {
+ } else if (!areEquivalent(*i->second, *v)) {
throw AssertionFailed((boost::format("Option %1% does not match for %2%, expected %3%, got %4%")
% i->first % name % *(i->second) % *v).str());
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/framing/SequenceSet.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/framing/SequenceSet.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/framing/SequenceSet.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/framing/SequenceSet.cpp Tue Jan 27 15:00:13 2015
@@ -33,7 +33,18 @@ namespace framing {
namespace {
//each range contains 2 numbers, 4 bytes each
-uint16_t RANGE_SIZE = 2 * 4;
+uint16_t RANGE_SIZE = 2 * 4;
+int32_t MAX_RANGE = 2147483647;//2^31-1
+
+int32_t gap(const SequenceNumber& a, const SequenceNumber& b)
+{
+ return a < b ? b - a : a - b;
+}
+
+bool is_max_range(const SequenceNumber& a, const SequenceNumber& b)
+{
+ return gap(a, b) == MAX_RANGE;
+}
}
void SequenceSet::encode(Buffer& buffer) const
@@ -54,7 +65,21 @@ void SequenceSet::decode(Buffer& buffer)
throw IllegalArgumentException(QPID_MSG("Invalid size for sequence set: " << size));
for (uint16_t i = 0; i < count; i++) {
- add(SequenceNumber(buffer.getLong()), SequenceNumber(buffer.getLong()));
+ SequenceNumber a(buffer.getLong());
+ SequenceNumber b(buffer.getLong());
+ if (b < a)
+ throw IllegalArgumentException(QPID_MSG("Invalid range in sequence set: " << a << " -> " << b));
+ if (is_max_range(a, b)) {
+ //RangeSet holds 'half-closed' ranges, where the end is
+ //one past the 'highest' value in the range. So if the
+ //range is already the maximum expressable with a 32bit
+ //sequence number, we can't represent it as a
+ //'half-closed' range, so we represent it as two ranges.
+ add(a, b-1);
+ add(b);
+ } else {
+ add(a, b);
+ }
}
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/ha/Primary.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/ha/Primary.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/ha/Primary.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/ha/Primary.cpp Tue Jan 27 15:00:13 2015
@@ -482,6 +482,7 @@ shared_ptr<PrimaryTxObserver> Primary::m
{
shared_ptr<PrimaryTxObserver> observer =
PrimaryTxObserver::create(*this, haBroker, txBuffer);
+ sys::Mutex::ScopedLock l(lock);
txMap[observer->getTxQueue()->getName()] = observer;
return observer;
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/ISSUES
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/ISSUES?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/ISSUES (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/ISSUES Tue Jan 27 15:00:13 2015
@@ -25,7 +25,7 @@ Current/pending:
------ ------- ----------------------
5359 - Linearstore: Implement new management schema and wire into store
5360 - Linearstore: Evaluate and rework logging to produce a consistent log output
- 5361 1145359 Linearstore: No tests for linearstore functionality currently exist
+* 5361 1145359 Linearstore: No tests for linearstore functionality currently exist
svn r.1564893 2014-02-05: Added tx-test-soak.sh
svn r.1564935 2014-02-05: Added license text to tx-test-soak.sh
svn r.1625283 2014-09-16: Basic python tests from legacystore ported over to linearstore
@@ -37,15 +37,22 @@ Current/pending:
** Basic performance tests
5464 - [linearstore] Incompletely created journal files accumulate in EFP
- 1088944 [Linearstore] store does not return all files to EFP after purging big queue <queue purge issue>
- 6043 1066256 [LinearStore] changing efp size after using store broke the new durable nodes creation
+* - 1066256 [LinearStore] changing efp size after using store broke the new durable nodes creation
- 1067480 [LinearStore] Provide a way to limit max count/size of empty files in EFP
- 1067429 [LinearStore] last file from deleted queue is not moved to EFP <queue delete issue>
- - 1067482 [LinearStore] Provide a way to prealocate empty pages in EFP
-* 5671 [linearstore] Add ability to use disk partitions and select per-queue EFPs
- svn r.1636598 2014-11-04: WIP: New EFP and journal dir structure using symlinks
- svn r.1637985 2014-11-10: WIP: Auto-upgrade from old dir structure to new
- 6230 1165200 [linearstore] qpid-qls-analyze fails when analyzing empty journal
- svn r.1643053 2014-11-18: Proposed fix
+ - 1067482 [LinearStore] Provide a way to preallocate empty pages in EFP
+* 6303 1180660 [linearstore] Roll back auto-upgrade of store directory structure
+* 5362 1145363 Linearstore: No store tools exist for examining the journals
+ svn r.1556888 2014-01-09: WIP checkin for linearstore version of qpid_qls_analyze. Needs testing and tidy-up.
+ svn r.1560530 2014-01-22: Bugfixes for qpid_qls_analyze
+ svn r.1561848 2014-01-27: Bugfixes and enhancements for qpid_qls_analyze
+ svn r.1564808 2014-02-05: Bugfixes and enhancements for qpid_qls_analyze
+ svn r.1578899 2014-03-18: Bugfixes and enhancements for qpid_qls_analyze
+ svn r.1583778 2014-04-01: Bugfix for qpid_qls_analyze
+ * Store analysis and status
+ * Recovery/reading of message content
+ * Empty file pool status and management
+
@@ -115,16 +122,6 @@ NO-JIRA - Added missing Apache co
5651 - [C++ broker] segfault in qpid::linearstore::journal::jdir::clear_dir when declaring durable queue
svn r.1582730 2014-03-28 Proposed fix by Pavel Moravec
* Bug introduced by r.1578899.
- 5362 1145363 Linearstore: No store tools exist for examining the journals
- svn r.1556888 2014-01-09: WIP checkin for linearstore version of qpid_qls_analyze. Needs testing and tidy-up.
- svn r.1560530 2014-01-22: Bugfixes for qpid_qls_analyze
- svn r.1561848 2014-01-27: Bugfixes and enhancements for qpid_qls_analyze
- svn r.1564808 2014-02-05: Bugfixes and enhancements for qpid_qls_analyze
- svn r.1578899 2014-03-18: Bugfixes and enhancements for qpid_qls_analyze
- svn r.1583778 2014-04-01: Bugfix for qpid_qls_analyze
- * Store analysis and status
- * Recovery/reading of message content
- * Empty file pool status and management
5661 - [linearstore] Set default cmake build to exclude linearstore
svn r.1584379 2014-04-03 Proposed solution.
* Run ccmake, select BUILD_LINEARSTORE to change its value to ON to build.
@@ -147,8 +144,16 @@ NO-JIRA - Added missing Apache co
svn r.1631360 2014-10-13 Proposed solution
6157 1150397 linearstore: segfault when 2 journals request new journal file from empty EFP
svn r.1632504 2014-10-17 Proposed solution by pmoravec
- 6248 1167911 [linearstore] Symlink creation fails if store dir path is not absolute
+ 6230 1165200 [linearstore] qpid-qls-analyze fails when analyzing empty journal
+ svn r.1643053 2014-11-18: Proposed fix
+ 6248 1167911 [linearstore] Symlink creation fails if store dir path is not absolute
svn r.1641689 2014-11-25 Proposed solution
+ 5671 1160367 [linearstore] Add ability to use disk partitions and select per-queue EFPs
+ svn r.1636598 2014-11-04: WIP: New EFP and journal dir structure using symlinks
+ svn r.1637985 2014-11-10: WIP: Auto-upgrade from old dir structure to new
+ svn r.1649081 2015-01-02: WIP: Specify new queue using qpid-config --durable together with --efp-partition-num and/or --efp-pool-file-size. Needs testing.
+ - 1148807 [linearstore] Restarting broker with empty journal raises confusing warning
+ Fixed by svn r.1649081 of bug 5671 / 1160367 above
Ordered checkin list:
@@ -187,14 +192,17 @@ no. svn r Q-JIRA RHBZ Date
28. 1596509 5767 1098118 2014-05-21 0.22-mrg (pmoravec)
29. 1596633 NO-JIRA 1078937 2014-05-21 (includes tools install update)
30. 1599243 5767 1098118 2014-06-02 0.22-mrg
-30. 1599243 5767 1098118 2014-06-02
-31. 1614665 5924 1124906 2014-07-30
-32. 1620426 6043 1089652 2014-08-25
-33. 1631360 6147 1152012 2014-10-13 (pmoravec)
-34. 1632504 6157 1150397 2014-10-17 (pmoravec)
-35. 1636598 5671 2014-11-04
-36. 1637985 5671 2014-11-10
-37. 1641689 6248 1167911 2014-11-25
+31. 1599243 5767 1098118 2014-06-02
+32. 1614665 5924 1124906 2014-07-30
+33. 1620426 6043 1089652 2014-08-25
+34. 1631360 6147 1152012 2014-10-13 (pmoravec)
+35. 1632504 6157 1150397 2014-10-17 (pmoravec)
+36. 1636598 5671 1160367 2014-11-04
+37. 1637985 5671 1160367 2014-11-10
+38. 1643053 6230 1165200 2014-11-18
+39. 1641689 6248 1167911 2014-11-25
+40. 1649081 5671 1160367 2015-01-02
+41. 1649082 NO-JIRA - 2015-01-02
See above sections for details on these checkins.
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/MessageStoreImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/MessageStoreImpl.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/MessageStoreImpl.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/MessageStoreImpl.cpp Tue Jan 27 15:00:13 2015
@@ -148,7 +148,7 @@ void MessageStoreImpl::initManagement ()
mgmtObject = qmf::org::apache::qpid::linearstore::Store::shared_ptr (
new qmf::org::apache::qpid::linearstore::Store(agent, this, broker));
- mgmtObject->set_location(storeDir);
+ mgmtObject->set_storeDir(storeDir);
mgmtObject->set_tplIsInitialized(false);
mgmtObject->set_tplDirectory(getTplBaseDir());
mgmtObject->set_tplWritePageSize(tplWCachePgSizeSblks * QLS_SBLK_SIZE_BYTES);
@@ -406,7 +406,7 @@ void MessageStoreImpl::create(qpid::brok
if (queue_.getName().size() == 0)
{
- QLS_LOG(error, "Cannot create store for empty (null) queue name - ignoring and attempting to continue.");
+ QLS_LOG(error, "Cannot create store for empty (null) queue name - queue create ignored.");
return;
}
@@ -449,15 +449,15 @@ qpid::linearstore::journal::EmptyFilePoo
MessageStoreImpl::getEmptyFilePool(const qpid::framing::FieldTable& args_) {
qpid::framing::FieldTable::ValuePtr value;
qpid::linearstore::journal::efpPartitionNumber_t localEfpPartition = defaultEfpPartitionNumber;
- value = args_.get("qpid.efp_partition");
+ value = args_.get("qpid.efp_partition_num");
if (value.get() != 0 && !value->empty() && value->convertsTo<int>()) {
- localEfpPartition = chkEfpPartition((uint32_t)value->get<int>(), "qpid.efp_partition");
+ localEfpPartition = chkEfpPartition((uint32_t)value->get<int>(), "qpid.efp_partition_num");
}
qpid::linearstore::journal::efpDataSize_kib_t localEfpFileSizeKib = defaultEfpFileSize_kib;
- value = args_.get("qpid.efp_file_size");
+ value = args_.get("qpid.efp_pool_file_size");
if (value.get() != 0 && !value->empty() && value->convertsTo<int>()) {
- localEfpFileSizeKib = chkEfpFileSizeKiB((uint32_t)value->get<int>(),"qpid.efp_file_size" );
+ localEfpFileSizeKib = chkEfpFileSizeKiB((uint32_t)value->get<int>(), "qpid.efp_pool_file_size");
}
return getEmptyFilePool(localEfpPartition, localEfpFileSizeKib);
}
@@ -1488,21 +1488,21 @@ std::string MessageStoreImpl::getStoreTo
std::string MessageStoreImpl::getJrnlBaseDir()
{
std::ostringstream dir;
- dir << storeDir << "/" << storeTopLevelDir << "/jrnl/" ;
+ dir << storeDir << "/" << storeTopLevelDir << "/jrnl2/" ;
return dir.str();
}
std::string MessageStoreImpl::getBdbBaseDir()
{
std::ostringstream dir;
- dir << storeDir << "/" << storeTopLevelDir << "/dat/" ;
+ dir << storeDir << "/" << storeTopLevelDir << "/dat2/" ;
return dir.str();
}
std::string MessageStoreImpl::getTplBaseDir()
{
std::ostringstream dir;
- dir << storeDir << "/" << storeTopLevelDir << "/tpl/" ;
+ dir << storeDir << "/" << storeTopLevelDir << "/tpl2/" ;
return dir.str();
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp Tue Jan 27 15:00:13 2015
@@ -59,10 +59,16 @@ EmptyFilePool::EmptyFilePool(const std::
EmptyFilePool::~EmptyFilePool() {}
void EmptyFilePool::initialize() {
-//std::cout << "*** Initializing EFP " << efpDataSize_kib_ << "k in partition " << partitionPtr_->getPartitionNumber() << "; efpDirectory=" << efpDirectory_ << std::endl; // DEBUG
- std::vector<std::string> dirList;
+ if (::mkdir(efpDirectory_.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH)) { // Create EFP dir if it does not yet exist
+ if (errno != EEXIST) {
+ std::ostringstream oss;
+ oss << "directory=" << efpDirectory_ << " " << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_EFP_MKDIR, oss.str(), "EmptyFilePool", "initialize");
+ }
+ }
// Process empty files in main dir
+ std::vector<std::string> dirList;
jdir::read_dir(efpDirectory_, dirList, false, true, false, false);
for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
size_t dotPos = i->rfind(".");
@@ -122,14 +128,14 @@ const efpIdentity_t EmptyFilePool::getId
std::string EmptyFilePool::takeEmptyFile(const std::string& destDirectory) {
std::string emptyFileName = popEmptyFile();
- std::string newFileName = efpDirectory_ + "/" + s_inuseFileDirectory_ + emptyFileName.substr(emptyFileName.rfind('/'));
+ std::string newFileName = efpDirectory_ + "/" + s_inuseFileDirectory_ + emptyFileName.substr(emptyFileName.rfind('/')); // NOTE: substr() includes leading '/'
std::string symlinkName = destDirectory + emptyFileName.substr(emptyFileName.rfind('/')); // NOTE: substr() includes leading '/'
- if (moveFile(emptyFileName, newFileName)) {
+ if (!moveFile(emptyFileName, newFileName)) {
// Try again with new UUID for file name
newFileName = efpDirectory_ + "/" + s_inuseFileDirectory_ + "/" + getEfpFileName();
- if (moveFile(emptyFileName, newFileName)) {
+ if (!moveFile(emptyFileName, newFileName)) {
//std::cerr << "*** DEBUG: pushEmptyFile " << emptyFileName << "from EmptyFilePool::takeEmptyFile()" << std::endl; // DEBUG
- pushEmptyFile(emptyFileName);
+ pushEmptyFile(emptyFileName); // Return empty file to pool
std::ostringstream oss;
oss << "file=\"" << emptyFileName << "\" dest=\"" << newFileName << "\"" << FORMAT_SYSERR(errno);
throw jexception(jerrno::JERR_JDIR_FMOVE, oss.str(), "EmptyFilePool", "takeEmptyFile");
@@ -138,7 +144,7 @@ std::string EmptyFilePool::takeEmptyFile
if (createSymLink(newFileName, symlinkName)) {
std::ostringstream oss;
oss << "file=\"" << emptyFileName << "\" dest=\"" << newFileName << "\" symlink=\"" << symlinkName << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_EFP_SYMLINK, oss.str(), "EmptyFilePool", "takeEmptyFile");
+ throw jexception(jerrno::JERR__SYMLINK, oss.str(), "EmptyFilePool", "takeEmptyFile");
}
return symlinkName;
}
@@ -189,12 +195,27 @@ efpDataSize_kib_t EmptyFilePool::dataSiz
}
// --- protected functions ---
+void EmptyFilePool::checkIosState(std::ofstream& ofs,
+ const uint32_t jerrno,
+ const std::string& fqFileName,
+ const std::string& operation,
+ const std::string& errorMessage,
+ const std::string& className,
+ const std::string& fnName) {
+ if (!ofs.good()) {
+ if (ofs.is_open()) {
+ ofs.close();
+ }
+ std::ostringstream oss;
+ oss << "IO failure: eofbit=" << (ofs.eof()?"T":"F") << " failbit=" << (ofs.fail()?"T":"F") << " badbit="
+ << (ofs.bad()?"T":"F") << " file=" << fqFileName << " operation=" << operation << ": " << errorMessage;
+ throw jexception(jerrno, oss.str(), className, fnName);
+ }
+}
std::string EmptyFilePool::createEmptyFile() {
std::string efpfn = getEfpFileName();
- if (!overwriteFileContents(efpfn)) {
- // TODO: handle failure to prepare new file here
- }
+ overwriteFileContents(efpfn);
return efpfn;
}
@@ -226,24 +247,20 @@ void EmptyFilePool::initializeSubDirecto
}
}
-bool EmptyFilePool::overwriteFileContents(const std::string& fqFileName) {
+void EmptyFilePool::overwriteFileContents(const std::string& fqFileName) {
::file_hdr_t fh;
::file_hdr_create(&fh, QLS_FILE_MAGIC, QLS_JRNL_VERSION, QLS_JRNL_FHDR_RES_SIZE_SBLKS, partitionPtr_->getPartitionNumber(), efpDataSize_kib_);
std::ofstream ofs(fqFileName.c_str(), std::ofstream::out | std::ofstream::binary);
- if (ofs.good()) {
- ofs.write((char*)&fh, sizeof(::file_hdr_t));
- uint64_t rem = ((efpDataSize_kib_ + (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB)) * 1024) - sizeof(::file_hdr_t);
- while (rem--)
- ofs.put('\0');
- ofs.close();
- return true;
-//std::cout << "*** WARNING: EFP " << efpDirectory_ << " is empty - created new journal file " << fqFileName.substr(fqFileName.rfind('/') + 1) << " on the fly" << std::endl; // DEBUG
- } else {
- std::ostringstream oss;
- oss << "std::ofstream ofs: file=\"" << fqFileName.c_str() << "\"" << " failed to be open";
- throw jexception(jerrno::JERR_EFP_FOPEN, oss.str(), "EmptyFilePool", "overwriteFileContents");
+ checkIosState(ofs, jerrno::JERR_EFP_FOPEN, fqFileName, "constructor", "Failed to create file", "EmptyFilePool", "overwriteFileContents");
+ ofs.write((char*)&fh, sizeof(::file_hdr_t));
+ checkIosState(ofs, jerrno::JERR_EFP_FWRITE, fqFileName, "write()", "Failed to write header", "EmptyFilePool", "overwriteFileContents");
+ uint64_t rem = ((efpDataSize_kib_ + (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB)) * 1024) - sizeof(::file_hdr_t);
+ while (rem--) {
+ ofs.put('\0');
+ checkIosState(ofs, jerrno::JERR_EFP_FWRITE, fqFileName, "put()", "Failed to put \0", "EmptyFilePool", "overwriteFileContents");
}
- return false;
+ ofs.close();
+//std::cout << "*** WARNING: EFP " << efpDirectory_ << " is empty - created new journal file " << fqFileName.substr(fqFileName.rfind('/') + 1) << " on the fly" << std::endl; // DEBUG
}
std::string EmptyFilePool::popEmptyFile() {
@@ -271,7 +288,7 @@ void EmptyFilePool::pushEmptyFile(const
void EmptyFilePool::returnEmptyFile(const std::string& emptyFileName) {
std::string returnedFileName = efpDirectory_ + "/" + s_returnedFileDirectory_ + emptyFileName.substr(emptyFileName.rfind('/')); // NOTE: substr() includes leading '/'
- if (moveFile(emptyFileName, returnedFileName)) {
+ if (!moveFile(emptyFileName, returnedFileName)) {
::unlink(emptyFileName.c_str());
//std::cerr << "*** WARNING: Unable to move file " << emptyFileName << " to " << returnedFileName << "; deleted." << std::endl; // DEBUG
}
@@ -283,7 +300,7 @@ void EmptyFilePool::returnEmptyFile(cons
overwriteFileContents(returnedFileName);
}
std::string sanitizedEmptyFileName = efpDirectory_ + returnedFileName.substr(returnedFileName.rfind('/')); // NOTE: substr() includes leading '/'
- if (moveFile(returnedFileName, sanitizedEmptyFileName)) {
+ if (!moveFile(returnedFileName, sanitizedEmptyFileName)) {
::unlink(returnedFileName.c_str());
//std::cerr << "*** WARNING: Unable to move file " << returnedFileName << " to " << sanitizedEmptyFileName << "; deleted." << std::endl; // DEBUG
} else {
@@ -395,18 +412,6 @@ bool EmptyFilePool::validateEmptyFile(co
return true;
}
-// static
-int EmptyFilePool::moveFile(const std::string& from,
- const std::string& to) {
- if (::rename(from.c_str(), to.c_str())) {
- if (errno == EEXIST) return errno; // File name exists
- std::ostringstream oss;
- oss << "file=\"" << from << "\" dest=\"" << to << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_JDIR_FMOVE, oss.str(), "EmptyFilePool", "returnEmptyFile");
- }
- return 0;
-}
-
//static
int EmptyFilePool::createSymLink(const std::string& fqFileName,
const std::string& fqLinkName) {
@@ -414,7 +419,7 @@ int EmptyFilePool::createSymLink(const s
if (errno == EEXIST) return errno; // File name exists
std::ostringstream oss;
oss << "file=\"" << fqFileName << "\" symlink=\"" << fqLinkName << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_EFP_SYMLINK, oss.str(), "EmptyFilePool", "createSymLink");
+ throw jexception(jerrno::JERR__SYMLINK, oss.str(), "EmptyFilePool", "createSymLink");
}
return 0;
}
@@ -426,7 +431,7 @@ std::string EmptyFilePool::deleteSymlink
if (len < 0) {
std::ostringstream oss;
oss << "symlink=\"" << fqLinkName << "\"" << FORMAT_SYSERR(errno);
- throw jexception(jerrno::JERR_EFP_SYMLINK, oss.str(), "EmptyFilePool", "deleteSymlink");
+ throw jexception(jerrno::JERR__SYMLINK, oss.str(), "EmptyFilePool", "deleteSymlink");
}
::unlink(fqLinkName.c_str());
return std::string(buff, len);
@@ -455,4 +460,18 @@ bool EmptyFilePool::isSymlink(const std:
}
+// static
+bool EmptyFilePool::moveFile(const std::string& from,
+ const std::string& to) {
+ if (::rename(from.c_str(), to.c_str())) {
+ if (errno == EEXIST) {
+ return false; // File name exists
+ }
+ std::ostringstream oss;
+ oss << "file=\"" << from << "\" dest=\"" << to << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_FMOVE, oss.str(), "EmptyFilePool", "returnEmptyFile");
+ }
+ return true;
+}
+
}}}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h Tue Jan 27 15:00:13 2015
@@ -87,23 +87,30 @@ public:
const efpPartitionNumber_t partitionNumber);
protected:
+ void checkIosState(std::ofstream& ofs,
+ const uint32_t jerrno,
+ const std::string& fqFileName,
+ const std::string& operation,
+ const std::string& errorMessage,
+ const std::string& className,
+ const std::string& fnName);
std::string createEmptyFile();
std::string getEfpFileName();
void initializeSubDirectory(const std::string& fqDirName);
- bool overwriteFileContents(const std::string& fqFileName);
+ void overwriteFileContents(const std::string& fqFileName);
std::string popEmptyFile();
void pushEmptyFile(const std::string fqFileName);
void returnEmptyFile(const std::string& emptyFileName);
void resetEmptyFileHeader(const std::string& fqFileName);
bool validateEmptyFile(const std::string& emptyFileName) const;
- static int moveFile(const std::string& fromFqPath,
- const std::string& toFqPath);
static int createSymLink(const std::string& fqFileName,
const std::string& fqLinkName);
static std::string deleteSymlink(const std::string& fqLinkName);
static bool isFile(const std::string& fqName);
static bool isSymlink(const std::string& fqName);
+ static bool moveFile(const std::string& fromFqPath,
+ const std::string& toFqPath);
};
}}}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolManager.cpp Tue Jan 27 15:00:13 2015
@@ -74,6 +74,7 @@ void EmptyFilePoolManager::findEfpPartit
if (!foundPartition) {
std::ostringstream oss1;
oss1 << qlsStorePath_ << "/" << EmptyFilePoolPartition::getPartionDirectoryName(defaultPartitionNumber_)
+ << "/" << EmptyFilePoolPartition::s_efpTopLevelDir_
<< "/" << EmptyFilePool::dirNameFromDataSize(defaultEfpDataSize_kib_);
jdir::create_dir(oss1.str());
insertPartition(defaultPartitionNumber_, oss1.str());
@@ -165,9 +166,10 @@ EmptyFilePool* EmptyFilePoolManager::get
EmptyFilePool* EmptyFilePoolManager::getEmptyFilePool(const efpPartitionNumber_t partitionNumber,
const efpDataSize_kib_t efpDataSize_kib) {
EmptyFilePoolPartition* efppp = getEfpPartition(partitionNumber > 0 ? partitionNumber : defaultPartitionNumber_);
- if (efppp != 0)
- return efppp->getEmptyFilePool(efpDataSize_kib > 0 ? efpDataSize_kib : defaultEfpDataSize_kib_);
- return 0;
+ if (efppp == 0) {
+ return 0;
+ }
+ return efppp->getEmptyFilePool(efpDataSize_kib > 0 ? efpDataSize_kib : defaultEfpDataSize_kib_, true);
}
void EmptyFilePoolManager::getEmptyFilePools(std::vector<EmptyFilePool*>& emptyFilePoolList,
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp Tue Jan 27 15:00:13 2015
@@ -32,6 +32,9 @@ namespace qpid {
namespace linearstore {
namespace journal {
+// static
+const std::string EmptyFilePoolPartition::s_efpTopLevelDir_("efp"); // Sets the top-level efp dir within a partition
+
EmptyFilePoolPartition::EmptyFilePoolPartition(const efpPartitionNumber_t partitionNum,
const std::string& partitionDir,
const bool overwriteBeforeReturnFlag,
@@ -57,72 +60,31 @@ EmptyFilePoolPartition::~EmptyFilePoolPa
void
EmptyFilePoolPartition::findEmptyFilePools() {
//std::cout << "*** EmptyFilePoolPartition::findEmptyFilePools(): Reading " << partitionDir_ << std::endl; // DEBUG
- std::vector<std::string> dirList;
- bool upgradeDirStructureFlag = false;
- std::string oldPartitionDir;
- jdir::read_dir(partitionDir_, dirList, true, false, false, false);
-//std::cout << "*** dirList.size()=" << dirList.size() << "; dirList.front()=" << dirList.front() << std::endl; // DEBUG
- if (dirList.size() == 1 && dirList.front().compare("efp") == 0) {
- upgradeDirStructureFlag = true;
- oldPartitionDir = partitionDir_ + "/efp";
-//std::cout << "*** oldPartitionDir=" << oldPartitionDir << std::endl; // DEBUG
- dirList.clear();
- jdir::read_dir(oldPartitionDir, dirList, true, false, false, false);
- }
+ std::string efpDir(partitionDir_ + "/" + s_efpTopLevelDir_);
+ if (jdir::is_dir(efpDir)) {
+ std::vector<std::string> dirList;
+ jdir::read_dir(efpDir, dirList, true, false, false, true);
for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
- std::string fqFileName(partitionDir_ + "/" + *i);
- if (upgradeDirStructureFlag) {
- std::string fqOldFileName(partitionDir_ + "/efp/" + *i);
- if (::rename(fqOldFileName.c_str(), fqFileName.c_str())) {
- // File move failed
- std::ostringstream oss;
- oss << "File \'" << fqOldFileName << "\' could not be renamed to \'" << fqFileName << "\' (" << FORMAT_SYSERR(errno) << "); file deleted";
- journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
- if (::unlink(fqOldFileName.c_str())) {
- std::ostringstream oss;
- oss << "File \'" << fqOldFileName << "\' could not be deleted (" << FORMAT_SYSERR(errno) << "\'; file orphaned";
- journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
- }
- }
- }
- EmptyFilePool* efpp = 0;
- try {
- efpp = new EmptyFilePool(fqFileName, this, overwriteBeforeReturnFlag_, truncateFlag_, journalLogRef_);
- {
- slock l(efpMapMutex_);
- efpMap_[efpp->dataSize_kib()] = efpp;
- }
- }
- catch (const std::exception& e) {
- if (efpp != 0) {
- delete efpp;
- efpp = 0;
- }
- std::ostringstream oss;
- oss << "EmptyFilePool create failed: " << e.what();
- journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
- }
- if (efpp != 0) {
- efpp->initialize();
- }
- }
- if (upgradeDirStructureFlag) {
- std::string oldEfpDir(partitionDir_ + "/efp");
- if (::rmdir(oldEfpDir.c_str())) {
- // Unable to delete old "efp" dir
- std::ostringstream oss;
- oss << "Unable to delete old EFP directory \'" << oldEfpDir << "\' (" << FORMAT_SYSERR(errno) << "\'; directory orphaned";
- journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
- }
+ createEmptyFilePool(*i);
}
+ } else {
+ std::ostringstream oss;
+ oss << "Partition \"" << partitionDir_ << "\" does not contain top level EFP dir \"" << s_efpTopLevelDir_ << "\"";
+ journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
+ }
}
-EmptyFilePool* EmptyFilePoolPartition::getEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib) {
- slock l(efpMapMutex_);
- efpMapItr_t i = efpMap_.find(efpDataSize_kib);
- if (i == efpMap_.end())
- return 0;
- return i->second;
+EmptyFilePool* EmptyFilePoolPartition::getEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib, const bool createIfNonExistent) {
+ {
+ slock l(efpMapMutex_);
+ efpMapItr_t i = efpMap_.find(efpDataSize_kib);
+ if (i != efpMap_.end())
+ return i->second;
+ }
+ if (createIfNonExistent) {
+ return createEmptyFilePool(efpDataSize_kib);
+ }
+ return 0;
}
void EmptyFilePoolPartition::getEmptyFilePools(std::vector<EmptyFilePool*>& efpList) {
@@ -183,7 +145,7 @@ std::string EmptyFilePoolPartition::getP
//static
efpPartitionNumber_t EmptyFilePoolPartition::getPartitionNumber(const std::string& name) {
if (name.length() == 4 && name[0] == 'p' && ::isdigit(name[1]) && ::isdigit(name[2]) && ::isdigit(name[3])) {
- long pn = ::strtol(name.c_str() + 1, 0, 0);
+ long pn = ::strtol(name.c_str() + 1, 0, 10);
if (pn == 0 && errno) {
return 0;
} else {
@@ -195,12 +157,42 @@ efpPartitionNumber_t EmptyFilePoolPartit
// --- protected functions ---
+EmptyFilePool* EmptyFilePoolPartition::createEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib) {
+ std::string fqEfpDirectoryName(partitionDir_ + "/" + EmptyFilePoolPartition::s_efpTopLevelDir_ + "/" + EmptyFilePool::dirNameFromDataSize(efpDataSize_kib));
+ return createEmptyFilePool(fqEfpDirectoryName);
+}
+
+EmptyFilePool* EmptyFilePoolPartition::createEmptyFilePool(const std::string fqEfpDirectoryName) {
+ EmptyFilePool* efpp = 0;
+ try {
+ efpp = new EmptyFilePool(fqEfpDirectoryName, this, overwriteBeforeReturnFlag_, truncateFlag_, journalLogRef_);
+ {
+ slock l(efpMapMutex_);
+ efpMap_[efpp->dataSize_kib()] = efpp;
+ }
+ }
+ catch (const std::exception& e) {
+ if (efpp != 0) {
+ delete efpp;
+ efpp = 0;
+ }
+ std::ostringstream oss;
+ oss << "EmptyFilePool create failed: " << e.what();
+ journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
+ }
+ if (efpp != 0) {
+ efpp->initialize();
+ }
+ return efpp;
+}
+
void EmptyFilePoolPartition::validatePartitionDir() {
+ std::ostringstream ss;
if (!jdir::is_dir(partitionDir_)) {
- std::ostringstream ss;
ss << "Invalid partition directory: \'" << partitionDir_ << "\' is not a directory";
throw jexception(jerrno::JERR_EFP_BADPARTITIONDIR, ss.str(), "EmptyFilePoolPartition", "validatePartitionDir");
}
+
// TODO: other validity checks here
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.h Tue Jan 27 15:00:13 2015
@@ -37,6 +37,8 @@ class JournalLog;
class EmptyFilePoolPartition
{
+public:
+ static const std::string s_efpTopLevelDir_;
protected:
typedef std::map<efpDataSize_kib_t, EmptyFilePool*> efpMap_t;
typedef efpMap_t::iterator efpMapItr_t;
@@ -59,7 +61,7 @@ public:
virtual ~EmptyFilePoolPartition();
void findEmptyFilePools();
- EmptyFilePool* getEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib);
+ EmptyFilePool* getEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib, const bool createIfNonExistent);
void getEmptyFilePools(std::vector<EmptyFilePool*>& efpList);
void getEmptyFilePoolSizes_kib(std::vector<efpDataSize_kib_t>& efpDataSizesList) const;
std::string getPartitionDirectory() const;
@@ -70,6 +72,8 @@ public:
static efpPartitionNumber_t getPartitionNumber(const std::string& name);
protected:
+ EmptyFilePool* createEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib);
+ EmptyFilePool* createEmptyFilePool(const std::string fqEfpDirectoryName);
void validatePartitionDir();
};
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp Tue Jan 27 15:00:13 2015
@@ -43,6 +43,7 @@
#include "qpid/linearstore/journal/utils/file_hdr.h"
#include <sstream>
#include <string>
+#include <unistd.h>
#include <vector>
namespace qpid {
@@ -101,7 +102,11 @@ void RecoveryManager::analyzeJournals(co
analyzeJournalFileHeaders(efpIdentity);
if (journalEmptyFlag_) {
- *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(0, 0); // Use default EFP
+ if (uninitFileList_.empty()) {
+ *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(0, 0); // Use default EFP
+ } else {
+ *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity);
+ }
} else {
*emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity);
if (! *emptyFilePoolPtrPtr) {
@@ -294,6 +299,7 @@ void RecoveryManager::setLinearFileContr
LinearFileController* lfcPtr) {
if (journalEmptyFlag_) {
if (uninitFileList_.size() > 0) {
+ // TODO: Handle case if uninitFileList_.size() > 1, but this should not happen in normal operation. Here we assume only one item in the list.
std::string uninitFile = uninitFileList_.back();
uninitFileList_.pop_back();
lfcPtr->restoreEmptyFile(uninitFile);
@@ -377,11 +383,28 @@ void RecoveryManager::analyzeJournalFile
jdir::read_dir(journalDirectory_, directoryList, false, true, false, true);
for (stringListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) {
bool hdrOk = readJournalFileHeader(*i, fileHeader, headerQueueName);
- if (!hdrOk || headerQueueName.empty()) {
+ bool hdrEmpty = ::is_file_hdr_reset(&fileHeader);
+ if (!hdrOk) {
std::ostringstream oss;
- oss << "Journal file " << (*i) << " is uninitialized or corrupted";
+ oss << "Journal file " << (*i) << " is corrupted or invalid";
journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str());
+ } else if (hdrEmpty) {
+ // Read symlink, find efp directory name which is efp size in KiB
+ // TODO: place this bit into a common function as it is also used in EmptyFilePool.cpp::deleteSymlink()
+ char buff[1024];
+ ssize_t len = ::readlink((*i).c_str(), buff, 1024);
+ if (len < 0) {
+ std::ostringstream oss;
+ oss << "symlink=\"" << (*i) << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR__SYMLINK, oss.str(), "RecoveryManager", "analyzeJournalFileHeaders");
+ }
+ // Find second and third '/' from back of string, which contains the EFP directory name
+ *(::strrchr(buff, '/')) = '\0';
+ *(::strrchr(buff, '/')) = '\0';
+ int efpDataSize_kib = atoi(::strrchr(buff, '/') + 1);
uninitFileList_.push_back(*i);
+ efpIdentity.pn_ = fileHeader._efp_partition;
+ efpIdentity.ds_ = efpDataSize_kib;
} else if (headerQueueName.compare(queueName_) != 0) {
std::ostringstream oss;
oss << "Journal file " << (*i) << " belongs to queue \"" << headerQueueName << "\": ignoring";
@@ -406,6 +429,7 @@ void RecoveryManager::analyzeJournalFile
}
}
+//std::cerr << "*** RecoveryManager::analyzeJournalFileHeaders() fileNumberMap_.size()=" << fileNumberMap_.size() << std::endl; // DEBUG
if (fileNumberMap_.empty()) {
journalEmptyFlag_ = true;
} else {
@@ -905,7 +929,9 @@ bool RecoveryManager::readJournalFileHea
}
ifs.close();
::memcpy(&fileHeaderRef, buffer, sizeof(::file_hdr_t));
- if (::file_hdr_check(&fileHeaderRef, QLS_FILE_MAGIC, QLS_JRNL_VERSION, 0, QLS_MAX_QUEUE_NAME_LEN)) return false;
+ if (::file_hdr_check(&fileHeaderRef, QLS_FILE_MAGIC, QLS_JRNL_VERSION, 0, QLS_MAX_QUEUE_NAME_LEN)) {
+ return false;
+ }
queueName.assign(buffer + sizeof(::file_hdr_t), fileHeaderRef._queue_name_len);
return true;
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp Tue Jan 27 15:00:13 2015
@@ -42,6 +42,7 @@ const uint32_t jerrno::JERR__UNEXPRESPON
const uint32_t jerrno::JERR__RECNFOUND = 0x0109;
const uint32_t jerrno::JERR__NOTIMPL = 0x010a;
const uint32_t jerrno::JERR__NULL = 0x010b;
+const uint32_t jerrno::JERR__SYMLINK = 0x010c;
// class jcntl
const uint32_t jerrno::JERR_JCNTL_STOPPED = 0x0200;
@@ -112,10 +113,11 @@ const uint32_t jerrno::JERR_EFP_BADPARTI
const uint32_t jerrno::JERR_EFP_BADEFPDIRNAME = 0x0d03;
const uint32_t jerrno::JERR_EFP_NOEFP = 0x0d04;
const uint32_t jerrno::JERR_EFP_EMPTY = 0x0d05;
-const uint32_t jerrno::JERR_EFP_SYMLINK = 0x0d06;
-const uint32_t jerrno::JERR_EFP_LSTAT = 0x0d07;
-const uint32_t jerrno::JERR_EFP_BADFILETYPE = 0x0d08;
-const uint32_t jerrno::JERR_EFP_FOPEN = 0x0d09;
+const uint32_t jerrno::JERR_EFP_LSTAT = 0x0d06;
+const uint32_t jerrno::JERR_EFP_BADFILETYPE = 0x0d07;
+const uint32_t jerrno::JERR_EFP_FOPEN = 0x0d08;
+const uint32_t jerrno::JERR_EFP_FWRITE = 0x0d09;
+const uint32_t jerrno::JERR_EFP_MKDIR = 0x0d0a;
// Negative returns for some functions
const int32_t jerrno::AIO_TIMEOUT = -1;
@@ -140,6 +142,7 @@ jerrno::__init()
_err_map[JERR__RECNFOUND] = "JERR__RECNFOUND: Record not found.";
_err_map[JERR__NOTIMPL] = "JERR__NOTIMPL: Not implemented";
_err_map[JERR__NULL] = "JERR__NULL: Operation on null pointer";
+ _err_map[JERR__SYMLINK] = "JERR__SYMLINK: Symbolic link operation failed";
// class jcntl
_err_map[JERR_JCNTL_STOPPED] = "JERR_JCNTL_STOPPED: Operation on stopped journal.";
@@ -210,10 +213,11 @@ jerrno::__init()
_err_map[JERR_EFP_BADPARTITIONDIR] = "JERR_EFP_BADPARTITIONDIR: Invalid partition directory";
_err_map[JERR_EFP_NOEFP] = "JERR_EFP_NOEFP: No Empty File Pool found for given partition and empty file size";
_err_map[JERR_EFP_EMPTY] = "JERR_EFP_EMPTY: Empty File Pool is empty";
- _err_map[JERR_EFP_SYMLINK] = "JERR_EFP_SYMLINK: Symbolic link operation failed";
_err_map[JERR_EFP_LSTAT] = "JERR_EFP_LSTAT: lstat() operation failed";
_err_map[JERR_EFP_BADFILETYPE] = "JERR_EFP_BADFILETYPE: File type incorrect for operation";
_err_map[JERR_EFP_FOPEN] = "JERR_EFP_FOPEN: Unable to fopen file for write";
+ _err_map[JERR_EFP_FWRITE] = "JERR_EFP_FWRITE: Write failed";
+ _err_map[JERR_EFP_MKDIR] = "JERR_EFP_MKDIR: Directory creation failed";
//_err_map[] = "";
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/journal/jerrno.h Tue Jan 27 15:00:13 2015
@@ -60,6 +60,7 @@ namespace journal {
static const uint32_t JERR__RECNFOUND; ///< Record not found
static const uint32_t JERR__NOTIMPL; ///< Not implemented
static const uint32_t JERR__NULL; ///< Operation on null pointer
+ static const uint32_t JERR__SYMLINK; ///< Symbolic Link operation failed
// class jcntl
static const uint32_t JERR_JCNTL_STOPPED; ///< Operation on stopped journal
@@ -130,10 +131,11 @@ namespace journal {
static const uint32_t JERR_EFP_BADPARTITIONDIR; ///< Invalid partition directory
static const uint32_t JERR_EFP_NOEFP; ///< No EFP found for given partition and file size
static const uint32_t JERR_EFP_EMPTY; ///< EFP empty
- static const uint32_t JERR_EFP_SYMLINK; ///< Symbolic Link operation failed
static const uint32_t JERR_EFP_LSTAT; ///< lstat operation failed
static const uint32_t JERR_EFP_BADFILETYPE; ///< Bad file type
static const uint32_t JERR_EFP_FOPEN; ///< Unable to fopen file for write
+ static const uint32_t JERR_EFP_FWRITE; ///< Write failed
+ static const uint32_t JERR_EFP_MKDIR; ///< Directory creation failed
// Negative returns for some functions
static const int32_t AIO_TIMEOUT; ///< Timeout waiting for AIO return
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/management-schema.xml
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/management-schema.xml?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/management-schema.xml (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/linearstore/management-schema.xml Tue Jan 27 15:00:13 2015
@@ -21,38 +21,24 @@
<class name="Store">
<property name="brokerRef" type="objId" access="RO" references="qpid.Broker" index="y" parentRef="y"/>
- <property name="location" type="sstr" access="RO" desc="Logical directory on disk"/>
- <!--property name="defaultInitialFileCount" type="uint16" access="RO" unit="file" desc="Default number of files initially allocated to each journal"/-->
- <!--property name="defaultDataFileSize" type="uint32" access="RO" unit="RdPg" desc="Default size of each journal data file"/-->
+ <property name="storeDir" type="sstr" access="RO" desc="Logical directory on disk"/>
<property name="tplIsInitialized" type="bool" access="RO" desc="Transaction prepared list has been initialized by a transactional prepare"/>
<property name="tplDirectory" type="sstr" access="RO" desc="Transaction prepared list directory"/>
<property name="tplWritePageSize" type="uint32" access="RO" unit="byte" desc="Page size in transaction prepared list write-page-cache"/>
<property name="tplWritePages" type="uint32" access="RO" unit="wpage" desc="Number of pages in transaction prepared list write-page-cache"/>
- <!--property name="tplInitialFileCount" type="uint16" access="RO" unit="file" desc="Number of files initially allocated to transaction prepared list journal"/-->
- <!--property name="tplDataFileSize" type="uint32" access="RO" unit="byte" desc="Size of each journal data file in transaction prepared list journal"/-->
- <!--property name="tplCurrentFileCount" type="uint32" access="RO" unit="file" desc="Number of files currently allocated to transaction prepared list journal"/-->
<statistic name="tplTransactionDepth" type="hilo32" unit="txn" desc="Number of currently enqueued prepared transactions"/>
<statistic name="tplTxnPrepares" type="count64" unit="record" desc="Total transaction prepares on transaction prepared list"/>
<statistic name="tplTxnCommits" type="count64" unit="record" desc="Total transaction commits on transaction prepared list"/>
<statistic name="tplTxnAborts" type="count64" unit="record" desc="Total transaction aborts on transaction prepared list"/>
- <statistic name="tplOutstandingAIOs" type="hilo32" unit="aio_op" desc="Deprecated"/>
</class>
<class name="Journal">
<property name="queueRef" type="objId" access="RO" references="qpid.Queue" isGeneralReference="y"/>
- <property name="name" type="sstr" access="RC" index="y"/>
+ <property name="queueName" type="sstr" access="RC" index="y"/>
<property name="directory" type="sstr" access="RO" desc="Directory containing journal files"/>
- <property name="baseFileName" type="sstr" access="RO" desc="Deprecated"/>
<property name="writePageSize" type="uint32" access="RO" unit="byte" desc="Deprecated"/>
<property name="writePages" type="uint32" access="RO" unit="wpage" desc="Deprecated"/>
- <property name="readPageSize" type="uint32" access="RO" unit="byte" desc="Deprecated"/>
- <property name="readPages" type="uint32" access="RO" unit="rpage" desc="Deprecated"/>
- <!--property name="initialFileCount" type="uint16" access="RO" unit="file" desc="Number of files initially allocated to this journal"/-->
- <!--property name="autoExpand" type="bool" access="RO" desc="Auto-expand enabled"/-->
- <!--property name="currentFileCount" type="uint16" access="RO" unit="file" desc="Number of files currently allocated to this journal"/-->
- <!--property name="maxFileCount" type="uint16" access="RO" unit="file" desc="Max number of files allowed for this journal"/-->
- <!--property name="dataFileSize" type="uint32" access="RO" unit="byte" desc="Size of each journal data file"/-->
<statistic name="recordDepth" type="hilo32" unit="record" desc="Number of currently enqueued records (durable messages)"/>
<statistic name="enqueues" type="count64" unit="record" desc="Total enqueued records on journal"/>
@@ -64,36 +50,5 @@
<statistic name="txnAborts" type="count64" unit="record" desc="Total transactional abort records on journal"/>
<statistic name="outstandingAIOs" type="hilo32" unit="aio_op" desc="Number of currently outstanding AIO requests in Async IO system"/>
-<!--
- The following are not yet "wired up" in JournalImpl.cpp
--->
- <statistic name="freeFileCount" type="hilo32" unit="file" desc="Deprecated"/>
- <statistic name="availableFileCount" type="hilo32" unit="file" desc="Deprecated"/>
- <statistic name="writeWaitFailures" type="count64" unit="record" desc="Deprecated"/>
- <statistic name="writeBusyFailures" type="count64" unit="record" desc="Deprecated"/>
- <statistic name="readRecordCount" type="count64" unit="record" desc="Deprecated"/>
- <statistic name="readBusyFailures" type="count64" unit="record" desc="Deprecated"/>
- <statistic name="writePageCacheDepth" type="hilo32" unit="wpage" desc="Deprecated"/>
- <statistic name="readPageCacheDepth" type="hilo32" unit="rpage" desc="Deprecated"/>
-
- <!--method name="expand" desc="Increase number of files allocated for this journal">
- <arg name="by" type="uint32" dir="I" desc="Number of files to increase journal size by"/>
- </method-->
</class>
-
- <eventArguments>
- <!--arg name="autoExpand" type="bool" desc="Journal auto-expand enabled"/-->
- <arg name="fileSize" type="uint32" desc="Journal file size in bytes"/>
- <arg name="jrnlId" type="sstr" desc="Journal Id"/>
- <arg name="numEnq" type="uint32" desc="Number of recovered enqueues"/>
- <arg name="numFiles" type="uint16" desc="Number of journal files"/>
- <arg name="numTxn" type="uint32" desc="Number of recovered transactions"/>
- <arg name="numTxnDeq" type="uint32" desc="Number of recovered transactional dequeues"/>
- <arg name="numTxnEnq" type="uint32" desc="Number of recovered transactional enqueues"/>
- <arg name="what" type="sstr" desc="Description of event"/>
- </eventArguments>
- <event name="enqThresholdExceeded" sev="warn" args="jrnlId, what"/>
- <event name="created" sev="notice" args="jrnlId, fileSize, numFiles"/>
- <event name="full" sev="error" args="jrnlId, what"/>
- <event name="recovered" sev="notice" args="jrnlId, fileSize, numFiles, numEnq, numTxn, numTxnEnq, numTxnDeq"/>
</schema>
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.cpp Tue Jan 27 15:00:13 2015
@@ -242,7 +242,7 @@ bool replace(Variant::Map& map, const st
}
}
-const uint32_t DEFAULT_DURABLE_TIMEOUT(15*60);//15 minutes
+const uint32_t DEFAULT_DURABLE_TIMEOUT(2*60);//2 minutes
const uint32_t DEFAULT_TIMEOUT(0);
}
@@ -267,7 +267,7 @@ AddressHelper::AddressHelper(const Addre
bind(link, RELIABILITY, reliability);
durableNode = test(node, DURABLE);
durableLink = test(link, DURABLE);
- timeout = get(link, TIMEOUT, durableLink ? DEFAULT_DURABLE_TIMEOUT : DEFAULT_TIMEOUT);
+ timeout = get(link, TIMEOUT, durableLink && reliability != AT_LEAST_ONCE ? DEFAULT_DURABLE_TIMEOUT : DEFAULT_TIMEOUT);
std::string mode;
if (bind(address, MODE, mode)) {
if (mode == BROWSE) {
@@ -571,7 +571,8 @@ bool AddressHelper::enabled(const std::s
bool AddressHelper::isUnreliable() const
{
- return reliability == AT_MOST_ONCE || reliability == UNRELIABLE;
+ return reliability == AT_MOST_ONCE || reliability == UNRELIABLE ||
+ (reliability.empty() && browse); // A browser defaults to unreliable.
}
const qpid::types::Variant::Map& AddressHelper::getNodeProperties() const
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/AddressHelper.h Tue Jan 27 15:00:13 2015
@@ -44,7 +44,6 @@ class AddressHelper
const qpid::types::Variant::Map& getNodeProperties() const;
bool getLinkSource(std::string& out) const;
bool getLinkTarget(std::string& out) const;
- bool getBrowse() const { return browse; }
const qpid::types::Variant::Map& getLinkProperties() const;
static std::string getLinkName(const Address& address);
private:
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Tue Jan 27 15:00:13 2015
@@ -292,7 +292,7 @@ bool ConnectionContext::get(boost::share
QPID_LOG(debug, "Received message of " << encoded->getSize() << " bytes: ");
encoded->init(impl);
impl.setEncoded(encoded);
- impl.setInternalId(ssn->record(current, lnk->getBrowse()));
+ impl.setInternalId(ssn->record(current));
pn_link_advance(lnk->receiver);
if (lnk->capacity) {
pn_link_flow(lnk->receiver, 1);
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp Tue Jan 27 15:00:13 2015
@@ -36,12 +36,10 @@ ReceiverContext::ReceiverContext(pn_sess
address(a),
helper(address),
receiver(pn_receiver(session, name.c_str())),
- capacity(0), used(0)
-{}
-
+ capacity(0), used(0) {}
ReceiverContext::~ReceiverContext()
{
- //pn_link_free(receiver);
+ pn_link_free(receiver);
}
void ReceiverContext::setCapacity(uint32_t c)
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h Tue Jan 27 15:00:13 2015
@@ -60,8 +60,6 @@ class ReceiverContext
void verify();
Address getAddress() const;
bool hasCurrent();
- bool getBrowse() const { return helper.getBrowse(); }
-
private:
friend class ConnectionContext;
const std::string name;
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp Tue Jan 27 15:00:13 2015
@@ -30,6 +30,7 @@
#include "qpid/messaging/Message.h"
#include "qpid/messaging/MessageImpl.h"
#include "qpid/log/Statement.h"
+#include "config.h"
extern "C" {
#include <proton/engine.h>
}
@@ -49,7 +50,7 @@ SenderContext::SenderContext(pn_session_
SenderContext::~SenderContext()
{
- //pn_link_free(sender);
+ pn_link_free(sender);
}
void SenderContext::close()
@@ -510,7 +511,11 @@ void SenderContext::Delivery::send(pn_li
{
pn_delivery_tag_t tag;
tag.size = sizeof(id);
+#ifdef NO_PROTON_DELIVERY_TAG_T
+ tag.start = reinterpret_cast<const char*>(&id);
+#else
tag.bytes = reinterpret_cast<const char*>(&id);
+#endif
token = pn_delivery(sender, tag);
pn_link_send(sender, encoded.getData(), encoded.getSize());
if (unreliable) {
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Tue Jan 27 15:00:13 2015
@@ -110,10 +110,11 @@ uint32_t SessionContext::getUnsettledAck
return 0;//TODO
}
-qpid::framing::SequenceNumber SessionContext::record(pn_delivery_t* delivery, bool browse)
+qpid::framing::SequenceNumber SessionContext::record(pn_delivery_t* delivery)
{
qpid::framing::SequenceNumber id = next++;
- if (!browse) unacked[id] = delivery;
+ if (!pn_delivery_settled(delivery))
+ unacked[id] = delivery;
QPID_LOG(debug, "Recorded delivery " << id << " -> " << delivery);
return id;
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h Tue Jan 27 15:00:13 2015
@@ -75,7 +75,7 @@ class SessionContext
qpid::framing::SequenceNumber next;
std::string name;
- qpid::framing::SequenceNumber record(pn_delivery_t*, bool browse);
+ qpid::framing::SequenceNumber record(pn_delivery_t*);
void acknowledge();
void acknowledge(const qpid::framing::SequenceNumber& id, bool cummulative);
void acknowledge(DeliveryMap::iterator begin, DeliveryMap::iterator end);
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp Tue Jan 27 15:00:13 2015
@@ -150,7 +150,7 @@ void AsynchIOHandler::readbuff(AsynchIO&
if (!codec) {
//TODO: may still want to revise this...
//send valid version header & close connection.
- write(framing::ProtocolInitiation(framing::highestProtocolVersion));
+ write(framing::ProtocolInitiation(factory->supportedVersion()));
readError = true;
aio->queueWriteClose();
} else {
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/ConnectionCodec.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/ConnectionCodec.h?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/ConnectionCodec.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/sys/ConnectionCodec.h Tue Jan 27 15:00:13 2015
@@ -60,6 +60,8 @@ class ConnectionCodec : public Codec {
virtual ConnectionCodec* create(
OutputControl&, const std::string& id, const SecuritySettings&
) = 0;
+
+ virtual framing::ProtocolVersion supportedVersion() const = 0;
};
};
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Jan 27 15:00:13 2015
@@ -7,3 +7,4 @@
/qpid/branches/java-network-refactor/qpid/cpp/src/tests:805429-825319
/qpid/branches/qpid-2935/qpid/cpp/src/tests:1061302-1072333
/qpid/branches/qpid-3346/qpid/cpp/src/tests:1144319-1179855
+/qpid/trunk/qpid/cpp/src/tests:1643238-1655056
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/assertions.py
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/assertions.py?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/assertions.py (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/assertions.py Tue Jan 27 15:00:13 2015
@@ -177,3 +177,18 @@ class AssertionTests (VersionTest):
assert False, "Expected assertion to fail on unspecified option"
except AssertionFailed: None
except MessagingError: None
+
+ def test_queue_autodelete_timeout(self):
+ name = str(uuid4())
+ # create subscription queue with 0-10 to be sure of name
+ ssn_0_10 = self.create_connection("amqp0-10", True).session()
+ ssn_0_10.receiver("amq.direct; {link:{name:%s,timeout:30}}" % name)
+ self.ssn.sender("%s; {assert:always, node:{x-declare:{arguments: {qpid.auto_delete_timeout: 30}}}}" % name)
+ ssn_0_10_other = self.create_connection("amqp0-10", True).session()
+ ssn_0_10_other.sender("%s; {assert:always, node:{x-declare:{arguments: {qpid.auto_delete_timeout: 30}}}}" % name)
+ try:
+ self.ssn.sender("%s; {assert:always, node:{x-declare:{arguments: {qpid.auto_delete_timeout: 60}}}}" % name)
+ ssn_0_10_other.sender("%s; {assert:always, node:{x-declare:{arguments: {qpid.auto_delete_timeout: 60}}}}" % name)
+ assert False, "Expected assertion to fail for auto_delete_timeout"
+ except AssertionFailed: None
+ except MessagingError: None
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/txshift.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/txshift.cpp?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/txshift.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/tests/txshift.cpp Tue Jan 27 15:00:13 2015
@@ -40,7 +40,7 @@ namespace tests {
struct Args : public qpid::TestOptions
{
std::string workQueue;
- size_t workers;
+ uint workers;
Args() : workQueue("txshift-control"), workers(1)
{
@@ -178,7 +178,7 @@ int main(int argc, char** argv)
worker.run();
} else {
boost::ptr_vector<Worker> workers;
- for (size_t i = 0; i < opts.workers; i++) {
+ for (uint i = 0; i < opts.workers; i++) {
workers.push_back(new Worker(connection, opts.workQueue));
}
std::for_each(workers.begin(), workers.end(), boost::bind(&Worker::start, _1));
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/Java-Broker-Appendix-Operational-Logging-Messages.xml
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/Java-Broker-Appendix-Operational-Logging-Messages.xml?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/Java-Broker-Appendix-Operational-Logging-Messages.xml (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/Java-Broker-Appendix-Operational-Logging-Messages.xml Tue Jan 27 15:00:13 2015
@@ -378,6 +378,15 @@
<para>Indicates that broker was shut down due to fatal error.</para>
</entry>
</row>
+ <row id="Java-Broker-Appendix-Operation-Logging-Message-BRK-1017">
+ <entry morerows="1">BRK-1017</entry>
+ <entry>Process : PID <replaceable>process identifier</replaceable></entry>
+ </row>
+ <row>
+ <entry>
+ <para>Process identifier (PID) of the Broker process.</para>
+ </entry>
+ </row>
</tbody>
</tgroup>
</table>
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/commonEntities.xml
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/commonEntities.xml?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/commonEntities.xml (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/java-broker/commonEntities.xml Tue Jan 27 15:00:13 2015
@@ -43,7 +43,7 @@
<!ENTITY oracleBdbProductOverviewUrl "http://www.oracle.com/technetwork/products/berkeleydb/overview/index-093405.html">
<!ENTITY oracleBdbRepGuideUrl "http://oracle.com/cd/E17277_02/html/ReplicationGuide/">
<!ENTITY oracleBdbJavaDocUrl "http://docs.oracle.com/cd/E17277_02/html/java/">
-<!ENTITY oracleBdbProductVersion "5.0.97">
+<!ENTITY oracleBdbProductVersion "5.0.104">
<!ENTITY oracleJmxTutorial "http://docs.oracle.com/javase/tutorial/jmx/">
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Jan 27 15:00:13 2015
@@ -9,3 +9,4 @@
/qpid/branches/java-network-refactor/qpid/java:805429-821809
/qpid/branches/qpid-2935/qpid/java:1061302-1072333
/qpid/trunk/qpid:796646-796653
+/qpid/trunk/qpid/java:1643238-1655056
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java Tue Jan 27 15:00:13 2015
@@ -25,6 +25,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
@@ -32,19 +33,20 @@ import javax.jms.InvalidDestinationExcep
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.jms.MessageListener;
+
import org.apache.qpid.amqp_1_0.client.AcknowledgeMode;
import org.apache.qpid.amqp_1_0.client.ConnectionErrorException;
import org.apache.qpid.amqp_1_0.client.Message;
import org.apache.qpid.amqp_1_0.client.Receiver;
import org.apache.qpid.amqp_1_0.client.Transaction;
import org.apache.qpid.amqp_1_0.jms.MessageConsumer;
+import org.apache.qpid.amqp_1_0.jms.MessageConsumerException;
import org.apache.qpid.amqp_1_0.jms.Queue;
import org.apache.qpid.amqp_1_0.jms.QueueReceiver;
import org.apache.qpid.amqp_1_0.jms.Session;
import org.apache.qpid.amqp_1_0.jms.TemporaryDestination;
import org.apache.qpid.amqp_1_0.jms.Topic;
import org.apache.qpid.amqp_1_0.jms.TopicSubscriber;
-import org.apache.qpid.amqp_1_0.jms.MessageConsumerException;
import org.apache.qpid.amqp_1_0.type.Binary;
import org.apache.qpid.amqp_1_0.type.Symbol;
import org.apache.qpid.amqp_1_0.type.UnsignedInteger;
@@ -333,10 +335,23 @@ public class MessageConsumerImpl impleme
message.setFromTopic(_isTopicSubscriber);
if(redelivery)
{
+ UnsignedInteger failures = message.getDeliveryFailures();
+
if(!message.getJMSRedelivered())
{
message.setJMSRedelivered(true);
}
+
+ if(failures == null)
+ {
+ message.setDeliveryFailures(UnsignedInteger.ONE);
+ }
+ else
+ {
+ message.setDeliveryFailures(failures.add(UnsignedInteger.ONE));
+ }
+
+
}
return message;
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java?rev=1655057&r1=1655056&r2=1655057&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/amqp-1-0-client/src/main/java/org/apache/qpid/amqp_1_0/client/Receiver.java Tue Jan 27 15:00:13 2015
@@ -530,6 +530,7 @@ public class Receiver implements Deliver
{
release(msg);
}
+ _session.removeReceiver(this);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org