You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2015/02/27 17:37:07 UTC
svn commit: r1662743 [2/2] - in /qpid/trunk/qpid: cpp/include/qpid/types/
cpp/src/ cpp/src/qpid/ cpp/src/qpid/amqp/ cpp/src/qpid/broker/
cpp/src/qpid/broker/amqp/ cpp/src/qpid/client/ cpp/src/qpid/messaging/amqp/
cpp/src/qpid/types/ cpp/src/tests/ test...
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h Fri Feb 27 16:37:06 2015
@@ -24,6 +24,7 @@
#include <deque>
#include <string>
#include <vector>
+#include <boost/shared_ptr.hpp>
#include "qpid/sys/IntegerTypes.h"
#include "qpid/messaging/Address.h"
#include "qpid/messaging/amqp/AddressHelper.h"
@@ -41,9 +42,10 @@ class Message;
class MessageImpl;
namespace amqp {
-/**
- *
- */
+
+class Transaction;
+
+
class SenderContext
{
public:
@@ -52,13 +54,15 @@ class SenderContext
public:
Delivery(int32_t id);
void encode(const qpid::messaging::MessageImpl& message, const qpid::messaging::Address&, bool setToField);
- void send(pn_link_t*, bool unreliable);
+ void send(pn_link_t*, bool unreliable, const types::Variant& state=types::Variant());
bool delivered();
bool accepted();
bool rejected();
void settle();
void reset();
bool sent() const;
+ pn_delivery_t* getToken() const { return token; }
+ std::string error();
private:
int32_t id;
pn_delivery_t* token;
@@ -66,22 +70,32 @@ class SenderContext
bool presettled;
};
- SenderContext(pn_session_t* session, const std::string& name, const qpid::messaging::Address& target, bool setToOnSend);
+ typedef boost::shared_ptr<Transaction> CoordinatorPtr;
+
+ SenderContext(pn_session_t* session, const std::string& name,
+ const qpid::messaging::Address& target,
+ bool setToOnSend,
+ const CoordinatorPtr& transaction = CoordinatorPtr());
~SenderContext();
- void reset(pn_session_t* session);
- void close();
- void setCapacity(uint32_t);
- uint32_t getCapacity();
- uint32_t getUnsettled();
- const std::string& getName() const;
- const std::string& getTarget() const;
- bool send(const qpid::messaging::Message& message, Delivery**);
- void configure();
- void verify();
- void check();
- bool settled();
- bool closed();
- Address getAddress() const;
+
+ virtual void reset(pn_session_t* session);
+ virtual void close();
+ virtual void setCapacity(uint32_t);
+ virtual uint32_t getCapacity();
+ virtual uint32_t getUnsettled();
+ virtual const std::string& getName() const;
+ virtual const std::string& getTarget() const;
+ virtual bool send(const qpid::messaging::Message& message, Delivery**);
+ virtual void configure();
+ virtual void verify();
+ virtual void check();
+ virtual bool settled();
+ virtual bool closed();
+ virtual Address getAddress() const;
+
+ protected:
+ pn_link_t* sender;
+
private:
friend class ConnectionContext;
typedef std::deque<Delivery> Deliveries;
@@ -89,12 +103,12 @@ class SenderContext
const std::string name;
qpid::messaging::Address address;
AddressHelper helper;
- pn_link_t* sender;
int32_t nextId;
Deliveries deliveries;
uint32_t capacity;
bool unreliable;
bool setToOnSend;
+ boost::shared_ptr<Transaction> transaction;
uint32_t processUnsettled(bool silent);
void configure(pn_terminus_t*);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp Fri Feb 27 16:37:06 2015
@@ -39,7 +39,8 @@ SenderHandle::SenderHandle(boost::shared
void SenderHandle::send(const Message& message, bool sync)
{
- connection->send(session, sender, message, sync);
+ SenderContext::Delivery* d = 0;
+ connection->send(session, sender, message, sync, &d);
}
void SenderHandle::close()
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Fri Feb 27 16:37:06 2015
@@ -21,11 +21,15 @@
#include "SessionContext.h"
#include "SenderContext.h"
#include "ReceiverContext.h"
+#include "Transaction.h"
+#include "PnData.h"
#include <boost/format.hpp>
#include "qpid/messaging/Address.h"
#include "qpid/messaging/Duration.h"
#include "qpid/messaging/exceptions.h"
#include "qpid/log/Statement.h"
+#include "qpid/amqp/descriptors.h"
+
extern "C" {
#include <proton/engine.h>
}
@@ -35,23 +39,32 @@ namespace messaging {
namespace amqp {
SessionContext::SessionContext(pn_connection_t* connection) : session(pn_session(connection)) {}
+
SessionContext::~SessionContext()
{
- senders.clear(); receivers.clear();
- pn_session_free(session);
+ // Clear all pointers to senders and receivers before we free the session.
+ senders.clear();
+ receivers.clear();
+ transaction.reset(); // Transaction is a sender.
+ if (!error && session)
+ pn_session_free(session);
}
boost::shared_ptr<SenderContext> SessionContext::createSender(const qpid::messaging::Address& address, bool setToOnSend)
{
+ error.raise();
std::string name = AddressHelper::getLinkName(address);
- if (senders.find(name) != senders.end()) throw LinkError("Link name must be unique within the scope of the connection");
- boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address, setToOnSend));
+ if (senders.find(name) != senders.end())
+ throw LinkError("Link name must be unique within the scope of the connection");
+ boost::shared_ptr<SenderContext> s(
+ new SenderContext(session, name, address, setToOnSend, transaction));
senders[name] = s;
return s;
}
boost::shared_ptr<ReceiverContext> SessionContext::createReceiver(const qpid::messaging::Address& address)
{
+ error.raise();
std::string name = AddressHelper::getLinkName(address);
if (receivers.find(name) != receivers.end()) throw LinkError("Link name must be unique within the scope of the connection");
boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address));
@@ -61,6 +74,7 @@ boost::shared_ptr<ReceiverContext> Sessi
boost::shared_ptr<SenderContext> SessionContext::getSender(const std::string& name) const
{
+ error.raise();
SenderMap::const_iterator i = senders.find(name);
if (i == senders.end()) {
throw qpid::messaging::KeyError(std::string("No such sender") + name);
@@ -71,6 +85,7 @@ boost::shared_ptr<SenderContext> Session
boost::shared_ptr<ReceiverContext> SessionContext::getReceiver(const std::string& name) const
{
+ error.raise();
ReceiverMap::const_iterator i = receivers.find(name);
if (i == receivers.end()) {
throw qpid::messaging::KeyError(std::string("No such receiver") + name);
@@ -81,16 +96,19 @@ boost::shared_ptr<ReceiverContext> Sessi
void SessionContext::removeReceiver(const std::string& n)
{
+ error.raise();
receivers.erase(n);
}
void SessionContext::removeSender(const std::string& n)
{
+ error.raise();
senders.erase(n);
}
boost::shared_ptr<ReceiverContext> SessionContext::nextReceiver()
{
+ error.raise();
for (SessionContext::ReceiverMap::iterator i = receivers.begin(); i != receivers.end(); ++i) {
if (i->second->hasCurrent()) {
return i->second;
@@ -102,16 +120,19 @@ boost::shared_ptr<ReceiverContext> Sessi
uint32_t SessionContext::getReceivable()
{
+ error.raise();
return 0;//TODO
}
uint32_t SessionContext::getUnsettledAcks()
{
+ error.raise();
return 0;//TODO
}
qpid::framing::SequenceNumber SessionContext::record(pn_delivery_t* delivery)
{
+ error.raise();
qpid::framing::SequenceNumber id = next++;
if (!pn_delivery_settled(delivery))
unacked[id] = delivery;
@@ -121,22 +142,32 @@ qpid::framing::SequenceNumber SessionCon
void SessionContext::acknowledge(DeliveryMap::iterator begin, DeliveryMap::iterator end)
{
+ error.raise();
for (DeliveryMap::iterator i = begin; i != end; ++i) {
- QPID_LOG(debug, "Setting disposition for delivery " << i->first << " -> " << i->second);
- pn_delivery_update(i->second, PN_ACCEPTED);
- pn_delivery_settle(i->second);//TODO: different settlement modes?
+ types::Variant txState;
+ if (transaction) {
+ QPID_LOG(trace, "Setting disposition for transactional delivery "
+ << i->first << " -> " << i->second);
+ transaction->acknowledge(i->second);
+ } else {
+ QPID_LOG(trace, "Setting disposition for delivery " << i->first << " -> " << i->second);
+ pn_delivery_update(i->second, PN_ACCEPTED);
+ pn_delivery_settle(i->second); //TODO: different settlement modes?
+ }
}
unacked.erase(begin, end);
}
void SessionContext::acknowledge()
{
+ error.raise();
QPID_LOG(debug, "acknowledging all " << unacked.size() << " messages");
acknowledge(unacked.begin(), unacked.end());
}
void SessionContext::acknowledge(const qpid::framing::SequenceNumber& id, bool cumulative)
{
+ error.raise();
QPID_LOG(debug, "acknowledging selected messages, id=" << id << ", cumulative=" << cumulative);
DeliveryMap::iterator i = unacked.find(id);
if (i != unacked.end()) {
@@ -149,6 +180,7 @@ void SessionContext::acknowledge(const q
void SessionContext::nack(const qpid::framing::SequenceNumber& id, bool reject)
{
+ error.raise();
DeliveryMap::iterator i = unacked.find(id);
if (i != unacked.end()) {
if (reject) {
@@ -166,7 +198,9 @@ void SessionContext::nack(const qpid::fr
bool SessionContext::settled()
{
+ error.raise();
bool result = true;
+
for (SenderMap::iterator i = senders.begin(); i != senders.end(); ++i) {
try {
if (!i->second->closed() && !i->second->settled()) result = false;
@@ -189,8 +223,25 @@ std::string SessionContext::getName() co
void SessionContext::reset(pn_connection_t* connection)
{
- session = pn_session(connection);
unacked.clear();
+ if (transaction) {
+ if (transaction->isCommitting())
+ error = new TransactionUnknown("Transaction outcome unknown: transport failure");
+ else
+ error = new TransactionAborted("Transaction aborted: transport failure");
+ resetSession(0);
+ senders.clear();
+ receivers.clear();
+ transaction.reset();
+ return;
+ }
+ resetSession(pn_session(connection));
+
+}
+
+void SessionContext::resetSession(pn_session_t* session_) {
+ session = session_;
+ if (transaction) transaction->reset(session);
for (SessionContext::SenderMap::iterator i = senders.begin(); i != senders.end(); ++i) {
i->second->reset(session);
}
@@ -198,4 +249,6 @@ void SessionContext::reset(pn_connection
i->second->reset(session);
}
}
+
+
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h Fri Feb 27 16:37:06 2015
@@ -26,6 +26,7 @@
#include <boost/shared_ptr.hpp>
#include "qpid/sys/IntegerTypes.h"
#include "qpid/framing/SequenceNumber.h"
+#include "qpid/sys/ExceptionHolder.h"
struct pn_connection_t;
struct pn_session_t;
@@ -42,6 +43,8 @@ namespace amqp {
class ConnectionContext;
class SenderContext;
class ReceiverContext;
+class Transaction;
+
/**
*
*/
@@ -63,23 +66,29 @@ class SessionContext
bool settled();
void setName(const std::string&);
std::string getName() const;
+
+ void nack(const qpid::framing::SequenceNumber& id, bool reject);
+
private:
friend class ConnectionContext;
typedef std::map<std::string, boost::shared_ptr<SenderContext> > SenderMap;
typedef std::map<std::string, boost::shared_ptr<ReceiverContext> > ReceiverMap;
typedef std::map<qpid::framing::SequenceNumber, pn_delivery_t*> DeliveryMap;
+
pn_session_t* session;
SenderMap senders;
+ boost::shared_ptr<Transaction> transaction;
ReceiverMap receivers;
DeliveryMap unacked;
qpid::framing::SequenceNumber next;
std::string name;
+ sys::ExceptionHolder error;
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);
- void nack(const qpid::framing::SequenceNumber& id, bool reject);
+ void resetSession(pn_session_t*);
};
}}} // namespace qpid::messaging::amqp
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp Fri Feb 27 16:37:06 2015
@@ -42,12 +42,12 @@ SessionHandle::SessionHandle(boost::shar
void SessionHandle::commit()
{
-
+ connection->commit(session);
}
void SessionHandle::rollback()
{
-
+ connection->rollback(session);
}
void SessionHandle::acknowledge(bool /*sync*/)
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp?rev=1662743&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp Fri Feb 27 16:37:06 2015
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "Transaction.h"
+#include "SessionContext.h"
+#include "ConnectionContext.h"
+#include "PnData.h"
+#include <proton/engine.h>
+#include <qpid/Exception.h>
+#include <qpid/amqp/descriptors.h>
+#include <qpid/messaging/exceptions.h>
+#include <qpid/log/Statement.h>
+#include "qpid/messaging/Message.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+using namespace types;
+using types::Exception;
+
+namespace {
+const std::string LOCAL_TRANSACTIONS("amqp:local-transactions");
+const std::string TX_COORDINATOR("tx-transaction");
+const std::string ADDRESS("tx-transaction;{link:{reliability:at-least-once}}");
+}
+
+Transaction::Transaction(pn_session_t* session) :
+ SenderContext(session, TX_COORDINATOR, Address(ADDRESS), false), committing(false)
+{}
+
+void Transaction::clear() {
+ id.clear();
+ sendState.reset();
+ acceptState.reset();
+}
+
+void Transaction::configure() {
+ SenderContext::configure();
+ pn_terminus_t* target = pn_link_target(sender);
+ pn_terminus_set_type(target, PN_COORDINATOR);
+ PnData(pn_terminus_capabilities(target)).putSymbol(LOCAL_TRANSACTIONS);
+}
+
+void Transaction::verify() {}
+
+const std::string& Transaction::getTarget() const { return getName(); }
+
+void Transaction::declare(SendFunction send, const SessionPtr& session) {
+ committing = false;
+ error.raise();
+ clear();
+ Variant declare = Variant::described(qpid::amqp::transaction::DECLARE_CODE, Variant::List());
+ SenderContext::Delivery* delivery = 0;
+ send(session, shared_from_this(), Message(declare), true, &delivery);
+ setId(*delivery);
+}
+
+void Transaction::discharge(SendFunction send, const SessionPtr& session, bool fail) {
+ error.raise();
+ committing = !fail;
+ try {
+ // Send a discharge message to the remote coordinator.
+ Variant::List dischargeList;
+ dischargeList.push_back(Variant(id));
+ dischargeList.push_back(Variant(fail));
+ Variant discharge(dischargeList);
+ discharge.setDescriptor(qpid::amqp::transaction::DISCHARGE_CODE);
+ SenderContext::Delivery* delivery = 0;
+ send(session, shared_from_this(), Message(discharge), true, &delivery);
+ if (!delivery->accepted())
+ throw TransactionAborted(delivery->error());
+ committing = false;
+ }
+ catch(const TransactionError&) {
+ throw;
+ }
+ catch(const Exception& e) {
+ committing = false;
+ throw TransactionAborted(e.what());
+ }
+}
+
+// Set the transaction ID from the delivery returned by the remote coordinator.
+void Transaction::setId(const SenderContext::Delivery& delivery)
+{
+ if (delivery.getToken() &&
+ pn_delivery_remote_state(delivery.getToken()) == qpid::amqp::transaction::DECLARED_CODE)
+ {
+ pn_data_t* data = pn_disposition_data(pn_delivery_remote(delivery.getToken()));
+ if (data && pn_data_next(data)) {
+ size_t count = pn_data_get_list(data);
+ if (count > 0) {
+ pn_data_enter(data);
+ pn_data_next(data);
+ setId(PnData::string(pn_data_get_binary(data)));
+ pn_data_exit(data);
+ return;
+ }
+ }
+ }
+ throw TransactionError("No transaction ID returned by remote coordinator.");
+}
+
+void Transaction::setId(const std::string& id_) {
+ id = id_;
+ if (id.empty()) {
+ clear();
+ }
+ else {
+ // NOTE: The send and accept states are NOT described, the descriptor
+ // is added in pn_delivery_update.
+ Variant::List list;
+ list.push_back(Variant(id, "binary"));
+ sendState = Variant(list);
+
+ Variant accepted = Variant::described(qpid::amqp::message::ACCEPTED_CODE, Variant::List());
+ list.push_back(accepted);
+ acceptState = Variant(list);
+ }
+}
+
+types::Variant Transaction::getSendState() const {
+ error.raise();
+ return sendState;
+}
+
+void Transaction::acknowledge(pn_delivery_t* delivery)
+{
+ error.raise();
+ PnData data(pn_disposition_data(pn_delivery_local(delivery)));
+ data.put(acceptState);
+ pn_delivery_update(delivery, qpid::amqp::transaction::TRANSACTIONAL_STATE_CODE);
+ pn_delivery_settle(delivery);
+}
+
+
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.h?rev=1662743&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transaction.h Fri Feb 27 16:37:06 2015
@@ -0,0 +1,95 @@
+#ifndef COORDINATORCONTEXT_H
+#define COORDINATORCONTEXT_H
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+#include "SenderContext.h"
+#include <qpid/types/Variant.h>
+#include "qpid/sys/ExceptionHolder.h"
+#include <boost/enable_shared_from_this.hpp>
+#include <boost/function.hpp>
+
+struct pn_session_t;
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+class SessionContext;
+class ConnectionContext;
+
+/**
+ * Track the current transaction for a session.
+ *
+ * Implements SenderContext, to send transaction command messages to remote coordinator.
+ */
+class Transaction : public SenderContext, public boost::enable_shared_from_this<Transaction> {
+ public:
+ typedef boost::shared_ptr<SessionContext> SessionPtr;
+
+ typedef boost::function<void (boost::shared_ptr<SessionContext> ssn,
+ boost::shared_ptr<SenderContext> snd,
+ const qpid::messaging::Message& message,
+ bool sync,
+ SenderContext::Delivery** delivery)> SendFunction;
+
+ Transaction(pn_session_t*);
+
+ sys::ExceptionHolder error;
+
+ /** Declare a transaction using connection and session to send to remote co-ordinator. */
+ void declare(SendFunction, const SessionPtr& session);
+
+ /** Discharge a transaction using connection and session to send to remote co-ordinator.
+ *@param fail: true means rollback, false means commit.
+ */
+ void discharge(SendFunction, const SessionPtr& session, bool fail);
+
+ /** Update a delivery with a transactional accept state. */
+ void acknowledge(pn_delivery_t* delivery);
+
+ /** Get delivery state to attach to transfers sent in a transaction. */
+ types::Variant getSendState() const;
+
+ /** Override SenderContext::getTarget with a more readable value */
+ const std::string& getTarget() const;
+
+ bool isCommitting() const { return committing; }
+
+ protected:
+ // SenderContext overrides
+ void configure();
+ void verify();
+
+ private:
+ std::string id;
+ types::Variant sendState;
+ types::Variant acceptState;
+ bool committing;
+
+
+ void clear();
+ void setId(const SenderContext::Delivery& delivery);
+ void setId(const std::string& id);
+};
+
+}}}
+
+#endif
Modified: qpid/trunk/qpid/cpp/src/qpid/types/Variant.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/types/Variant.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/types/Variant.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/types/Variant.cpp Fri Feb 27 16:37:06 2015
@@ -43,21 +43,23 @@ class VariantImpl
{
public:
VariantImpl();
- VariantImpl(bool);
- VariantImpl(uint8_t);
- VariantImpl(uint16_t);
- VariantImpl(uint32_t);
- VariantImpl(uint64_t);
- VariantImpl(int8_t);
- VariantImpl(int16_t);
- VariantImpl(int32_t);
- VariantImpl(int64_t);
- VariantImpl(float);
- VariantImpl(double);
- VariantImpl(const std::string&, const std::string& encoding=std::string());
- VariantImpl(const Variant::Map&);
- VariantImpl(const Variant::List&);
- VariantImpl(const Uuid&);
+ void reset();
+ void set(bool);
+ void set(uint8_t);
+ void set(uint16_t);
+ void set(uint32_t);
+ void set(uint64_t);
+ void set(int8_t);
+ void set(int16_t);
+ void set(int32_t);
+ void set(int64_t);
+ void set(float);
+ void set(double);
+ void set(const std::string&, const std::string& encoding=std::string());
+ void set(const Variant::Map&);
+ void set(const Variant::List&);
+ void set(const Uuid&);
+ void set(const Variant&);
~VariantImpl();
VariantType getType() const;
@@ -90,9 +92,10 @@ class VariantImpl
bool isEqualTo(VariantImpl&) const;
bool isEquivalentTo(VariantImpl&) const;
- static VariantImpl* create(const Variant&);
+ Variant::List descriptors; // Optional descriptors for described value.
+
private:
- const VariantType type;
+ VariantType type;
union {
bool b;
uint8_t ui8;
@@ -110,7 +113,7 @@ class VariantImpl
Variant::List* list;
std::string* string;
} value;
- std::string encoding;//optional encoding for variable length data
+ std::string encoding; // Optional encoding for variable length data.
template<class T> T convertFromString() const
{
@@ -136,26 +139,34 @@ class VariantImpl
};
+VariantImpl::VariantImpl() : type(VAR_VOID) {}
+
+void VariantImpl::set(bool b) { reset(); type = VAR_BOOL; value.b = b; }
+void VariantImpl::set(uint8_t i) { reset(); type = VAR_UINT8; value.ui8 = i; }
+void VariantImpl::set(uint16_t i) { reset(); type = VAR_UINT16; value.ui16 = i; }
+void VariantImpl::set(uint32_t i) { reset(); type = VAR_UINT32; value.ui32 = i; }
+void VariantImpl::set(uint64_t i) { reset(); type = VAR_UINT64; value.ui64 = i; }
+void VariantImpl::set(int8_t i) { reset(); type = VAR_INT8; value.i8 = i; }
+void VariantImpl::set(int16_t i) { reset(); type = VAR_INT16; value.i16 = i; }
+void VariantImpl::set(int32_t i) { reset(); type = VAR_INT32; value.i32 = i; }
+void VariantImpl::set(int64_t i) { reset(); type = VAR_INT64; value.i64 = i; }
+void VariantImpl::set(float f) { reset(); type = VAR_FLOAT; value.f = f; }
+void VariantImpl::set(double d) { reset(); type = VAR_DOUBLE; value.d = d; }
+void VariantImpl::set(const std::string& s, const std::string& e) { reset(); type = VAR_STRING; encoding = e; value.string = new std::string(s); }
+
+void VariantImpl::set(const Variant::Map& m) {
+ reset();
+ type = VAR_MAP;
+ value.map = new Variant::Map(m);
+}
+
+void VariantImpl::set(const Variant::List& l) { reset(); type = VAR_LIST; value.list = new Variant::List(l); }
+
+void VariantImpl::set(const Uuid& u) { reset(); type = VAR_UUID; value.uuid = new Uuid(u); }
-VariantImpl::VariantImpl() : type(VAR_VOID) { value.i64 = 0; }
-VariantImpl::VariantImpl(bool b) : type(VAR_BOOL) { value.b = b; }
-VariantImpl::VariantImpl(uint8_t i) : type(VAR_UINT8) { value.ui8 = i; }
-VariantImpl::VariantImpl(uint16_t i) : type(VAR_UINT16) { value.ui16 = i; }
-VariantImpl::VariantImpl(uint32_t i) : type(VAR_UINT32) { value.ui32 = i; }
-VariantImpl::VariantImpl(uint64_t i) : type(VAR_UINT64) { value.ui64 = i; }
-VariantImpl::VariantImpl(int8_t i) : type(VAR_INT8) { value.i8 = i; }
-VariantImpl::VariantImpl(int16_t i) : type(VAR_INT16) { value.i16 = i; }
-VariantImpl::VariantImpl(int32_t i) : type(VAR_INT32) { value.i32 = i; }
-VariantImpl::VariantImpl(int64_t i) : type(VAR_INT64) { value.i64 = i; }
-VariantImpl::VariantImpl(float f) : type(VAR_FLOAT) { value.f = f; }
-VariantImpl::VariantImpl(double d) : type(VAR_DOUBLE) { value.d = d; }
-VariantImpl::VariantImpl(const std::string& s, const std::string& e)
- : type(VAR_STRING), encoding(e) { value.string = new std::string(s); }
-VariantImpl::VariantImpl(const Variant::Map& m) : type(VAR_MAP) { value.map = new Variant::Map(m); }
-VariantImpl::VariantImpl(const Variant::List& l) : type(VAR_LIST) { value.list = new Variant::List(l); }
-VariantImpl::VariantImpl(const Uuid& u) : type(VAR_UUID) { value.uuid = new Uuid(u); }
+VariantImpl::~VariantImpl() { reset(); }
-VariantImpl::~VariantImpl() {
+void VariantImpl::reset() {
switch (type) {
case VAR_STRING:
delete value.string;
@@ -172,6 +183,7 @@ VariantImpl::~VariantImpl() {
default:
break;
}
+ type = VAR_VOID;
}
VariantType VariantImpl::getType() const { return type; }
@@ -637,46 +649,50 @@ bool isIntegerType(VariantType type)
}
}
-VariantImpl* VariantImpl::create(const Variant& v)
+void VariantImpl::set(const Variant& v)
{
switch (v.getType()) {
- case VAR_BOOL: return new VariantImpl(v.asBool());
- case VAR_UINT8: return new VariantImpl(v.asUint8());
- case VAR_UINT16: return new VariantImpl(v.asUint16());
- case VAR_UINT32: return new VariantImpl(v.asUint32());
- case VAR_UINT64: return new VariantImpl(v.asUint64());
- case VAR_INT8: return new VariantImpl(v.asInt8());
- case VAR_INT16: return new VariantImpl(v.asInt16());
- case VAR_INT32: return new VariantImpl(v.asInt32());
- case VAR_INT64: return new VariantImpl(v.asInt64());
- case VAR_FLOAT: return new VariantImpl(v.asFloat());
- case VAR_DOUBLE: return new VariantImpl(v.asDouble());
- case VAR_STRING: return new VariantImpl(v.asString(), v.getEncoding());
- case VAR_MAP: return new VariantImpl(v.asMap());
- case VAR_LIST: return new VariantImpl(v.asList());
- case VAR_UUID: return new VariantImpl(v.asUuid());
- default: return new VariantImpl();
- }
-}
-
-Variant::Variant() : impl(new VariantImpl()) {}
-Variant::Variant(bool b) : impl(new VariantImpl(b)) {}
-Variant::Variant(uint8_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(uint16_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(uint32_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(uint64_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(int8_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(int16_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(int32_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(int64_t i) : impl(new VariantImpl(i)) {}
-Variant::Variant(float f) : impl(new VariantImpl(f)) {}
-Variant::Variant(double d) : impl(new VariantImpl(d)) {}
-Variant::Variant(const std::string& s) : impl(new VariantImpl(s)) {}
-Variant::Variant(const char* s) : impl(new VariantImpl(std::string(s))) {}
-Variant::Variant(const Map& m) : impl(new VariantImpl(m)) {}
-Variant::Variant(const List& l) : impl(new VariantImpl(l)) {}
-Variant::Variant(const Variant& v) : impl(VariantImpl::create(v)) {}
-Variant::Variant(const Uuid& u) : impl(new VariantImpl(u)) {}
+ case VAR_BOOL: set(v.asBool()); break;
+ case VAR_UINT8: set(v.asUint8()); break;
+ case VAR_UINT16: set(v.asUint16()); break;
+ case VAR_UINT32: set(v.asUint32()); break;
+ case VAR_UINT64: set(v.asUint64()); break;
+ case VAR_INT8: set(v.asInt8()); break;
+ case VAR_INT16: set(v.asInt16()); break;
+ case VAR_INT32: set(v.asInt32()); break;
+ case VAR_INT64: set(v.asInt64()); break;
+ case VAR_FLOAT: set(v.asFloat()); break;
+ case VAR_DOUBLE: set(v.asDouble()); break;
+ case VAR_STRING: set(v.asString(), v.getEncoding()); break;
+ case VAR_MAP: set(v.asMap()); break;
+ case VAR_LIST: set(v.asList()); break;
+ case VAR_UUID: set(v.asUuid()); break;
+ default: reset();
+ }
+ encoding = v.getEncoding();
+ descriptors = v.getDescriptors();
+}
+
+Variant::Variant() : impl(0) {}
+Variant::Variant(bool b) : impl(new VariantImpl()) { impl->set(b); }
+Variant::Variant(uint8_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(uint16_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(uint32_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(uint64_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(int8_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(int16_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(int32_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(int64_t i) : impl(new VariantImpl()) { impl->set(i); }
+Variant::Variant(float f) : impl(new VariantImpl()) { impl->set(f); }
+Variant::Variant(double d) : impl(new VariantImpl()) { impl->set(d); }
+Variant::Variant(const std::string& s) : impl(new VariantImpl()) { impl->set(s); }
+Variant::Variant(const std::string& s, const std::string& encoding) : impl(new VariantImpl()) { impl->set(s, encoding); }
+Variant::Variant(const char* s) : impl(new VariantImpl()) { impl->set(std::string(s)); }
+Variant::Variant(const char* s, const char* encoding) : impl(new VariantImpl()) { impl->set(std::string(s), std::string(encoding)); }
+Variant::Variant(const Map& m) : impl(new VariantImpl()) { impl->set(m); }
+Variant::Variant(const List& l) : impl(new VariantImpl()) { impl->set(l); }
+Variant::Variant(const Variant& v) : impl(new VariantImpl()) { impl->set(v); }
+Variant::Variant(const Uuid& u) : impl(new VariantImpl()) { impl->set(u); }
Variant::~Variant() { if (impl) delete impl; }
@@ -686,116 +702,105 @@ void Variant::reset()
impl = 0;
}
+namespace {
+VariantImpl* assure(VariantImpl*& ptr) {
+ if (!ptr) ptr = new VariantImpl();
+ return ptr;
+}
+}
Variant& Variant::operator=(bool b)
{
- if (impl) delete impl;
- impl = new VariantImpl(b);
+ assure(impl)->set(b);
return *this;
}
Variant& Variant::operator=(uint8_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(uint16_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(uint32_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(uint64_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(int8_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(int16_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(int32_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(int64_t i)
{
- if (impl) delete impl;
- impl = new VariantImpl(i);
+ assure(impl)->set(i);
return *this;
}
Variant& Variant::operator=(float f)
{
- if (impl) delete impl;
- impl = new VariantImpl(f);
+ assure(impl)->set(f);
return *this;
}
Variant& Variant::operator=(double d)
{
- if (impl) delete impl;
- impl = new VariantImpl(d);
+ assure(impl)->set(d);
return *this;
}
Variant& Variant::operator=(const std::string& s)
{
- if (impl) delete impl;
- impl = new VariantImpl(s);
+ assure(impl)->set(s);
return *this;
}
Variant& Variant::operator=(const char* s)
{
- if (impl) delete impl;
- impl = new VariantImpl(std::string(s));
+ assure(impl)->set(std::string(s));
return *this;
}
Variant& Variant::operator=(const Uuid& u)
{
- if (impl) delete impl;
- impl = new VariantImpl(u);
+ assure(impl)->set(u);
return *this;
}
Variant& Variant::operator=(const Map& m)
{
- if (impl) delete impl;
- impl = new VariantImpl(m);
+ assure(impl)->set(m);
return *this;
}
Variant& Variant::operator=(const List& l)
{
- if (impl) delete impl;
- impl = new VariantImpl(l);
+ assure(impl)->set(l);
return *this;
}
Variant& Variant::operator=(const Variant& v)
{
- if (impl) delete impl;
- impl = VariantImpl::create(v);
+ assure(impl)->set(v);
return *this;
}
@@ -841,8 +846,7 @@ Variant::List& Variant::asList() { if (!
const std::string& Variant::getString() const { if (!impl) throw InvalidConversion("Can't convert VOID to STRING"); return impl->getString(); }
std::string& Variant::getString() { if (!impl) throw InvalidConversion("Can't convert VOID to STRING"); return impl->getString(); }
void Variant::setEncoding(const std::string& s) {
- if (!impl) impl = new VariantImpl();
- impl->setEncoding(s);
+ assure(impl)->setEncoding(s);
}
const std::string& Variant::getEncoding() const { return impl ? impl->getEncoding() : EMPTY; }
@@ -884,6 +888,12 @@ std::ostream& operator<<(std::ostream& o
std::ostream& operator<<(std::ostream& out, const Variant& value)
{
+ // Print the descriptors
+ const Variant::List& descriptors = value.getDescriptors();
+ for (Variant::List::const_iterator i = descriptors.begin(); i != descriptors.end(); ++i)
+ out << "@" << *i << " ";
+
+ // Print the value
switch (value.getType()) {
case VAR_MAP:
out << value.asMap();
@@ -910,7 +920,43 @@ bool operator!=(const Variant& a, const
bool Variant::isEqualTo(const Variant& other) const
{
+ if (isVoid() && other.isVoid()) return true;
+ if (isVoid() || other.isVoid()) return false;
return impl && impl->isEqualTo(*other.impl);
}
+bool Variant::isDescribed() const {
+ return impl && !impl->descriptors.empty();
+}
+
+Variant::List& Variant::getDescriptors() {
+ return assure(impl)->descriptors;
+}
+
+const Variant::List& Variant::getDescriptors() const {
+ return assure(impl)->descriptors;
+}
+
+Variant Variant::getDescriptor() const {
+ if (getDescriptors().size() > 0) return getDescriptors().front();
+ else return Variant();
+}
+
+void Variant::setDescriptor(const Variant& descriptor) {
+ getDescriptors().clear();
+ getDescriptors().push_back(descriptor);
+}
+
+Variant Variant::described(const Variant& descriptor, const Variant& value) {
+ Variant described(value);
+ described.setDescriptor(descriptor);
+ return described;
+}
+
+Variant Variant::described(const Variant& descriptor, const List& value) {
+ Variant described(value);
+ described.setDescriptor(descriptor);
+ return described;
+}
+
}} // namespace qpid::types
Modified: qpid/trunk/qpid/cpp/src/qpid/types/encodings.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/types/encodings.h?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/types/encodings.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/types/encodings.h Fri Feb 27 16:37:06 2015
@@ -23,11 +23,13 @@
*/
namespace qpid {
namespace types {
+
namespace encodings {
const std::string BINARY("binary");
const std::string UTF8("utf8");
const std::string ASCII("ascii");
}
+
}} // namespace qpid::types
#endif /*!QPID_TYPES_ENCODINGS_H*/
Modified: qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt Fri Feb 27 16:37:06 2015
@@ -360,6 +360,11 @@ if (NOT CMAKE_SYSTEM_NAME STREQUAL Windo
# paged queue not yet implemented for windows
add_test (paged_queue_tests ${shell} ${CMAKE_CURRENT_SOURCE_DIR}/run_paged_queue_tests${test_script_suffix})
endif (NOT CMAKE_SYSTEM_NAME STREQUAL Windows)
+
+if (BUILD_AMQP)
+ add_test (interop_tests ${python_wrap} -- ${CMAKE_CURRENT_SOURCE_DIR}/interop_tests.py)
+endif (BUILD_AMQP)
+
add_test (ha_tests ${python_wrap} -- ${CMAKE_CURRENT_SOURCE_DIR}/ha_tests.py)
add_test (qpidd_qmfv2_tests ${python_wrap} -- ${CMAKE_CURRENT_SOURCE_DIR}/qpidd_qmfv2_tests.py)
if (BUILD_AMQP)
Modified: qpid/trunk/qpid/cpp/src/tests/Variant.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/Variant.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/Variant.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/Variant.cpp Fri Feb 27 16:37:06 2015
@@ -18,14 +18,16 @@
* under the License.
*
*/
-#include <iostream>
-#include "qpid/types/Variant.h"
-#include "qpid/amqp_0_10/Codecs.h"
#include "unit_test.h"
+#include "qpid/types/Variant.h"
+#include "qpid/amqp_0_10/Codecs.h"
+#include <boost/assign.hpp>
+#include <iostream>
using namespace qpid::types;
using namespace qpid::amqp_0_10;
+using boost::assign::list_of;
namespace qpid {
namespace tests {
@@ -807,6 +809,22 @@ QPID_AUTO_TEST_CASE(parse)
BOOST_CHECK(a.getType()==types::VAR_DOUBLE);
}
+QPID_AUTO_TEST_CASE(described)
+{
+ Variant a;
+ BOOST_CHECK(!a.isDescribed());
+ a.getDescriptors().push_back("foo");
+ BOOST_CHECK(a.isDescribed());
+ BOOST_CHECK_EQUAL(a.getDescriptors(), list_of<Variant>("foo"));
+ a = 42;
+ BOOST_CHECK(a.isDescribed());
+ BOOST_CHECK_EQUAL(a.getDescriptors(), list_of<Variant>("foo"));
+ a.getDescriptors().push_back(33);
+ BOOST_CHECK_EQUAL(a.getDescriptors(), list_of<Variant>("foo")(33));
+ a.getDescriptors().clear();
+ BOOST_CHECK(!a.isDescribed());
+}
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
Modified: qpid/trunk/qpid/cpp/src/tests/brokertest.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/brokertest.py?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/brokertest.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/brokertest.py Fri Feb 27 16:37:06 2015
@@ -21,6 +21,7 @@
import os, signal, string, tempfile, subprocess, socket, threading, time, imp, re
import qpid, traceback, signal
+import proton
from qpid import connection, util
from qpid.compat import format_exc
from unittest import TestCase
@@ -48,13 +49,18 @@ from qpidtoollibs import BrokerAgent
import qpid.messaging
qm = qpid.messaging
qpid_messaging = None
+
+def env_has_log_config():
+ """True if there are qpid log configuratoin settings in the environment."""
+ return "QPID_LOG_ENABLE" in os.environ or "QPID_TRACE" in os.environ
+
if not os.environ.get("QPID_PY_NO_SWIG"):
try:
import qpid_messaging
from qpid.datatypes import uuid4
qm = qpid_messaging
# Silence warnings from swigged messaging library unless enabled in environment.
- if "QPID_LOG_ENABLE" not in os.environ and "QPID_TRACE" not in os.environ:
+ if not env_has_log_config():
qm.Logger.configure(["--log-enable=error"])
except ImportError:
print "Cannot load python SWIG bindings, falling back to native qpid.messaging."
@@ -135,7 +141,7 @@ _popen_id = AtomicCounter() # Popen iden
# Constants for file descriptor arguments to Popen
FILE = "FILE" # Write to file named after process
-PIPE = subprocess.PIPE
+from subprocess import PIPE, STDOUT
class Popen(subprocess.Popen):
"""
@@ -201,7 +207,7 @@ class Popen(subprocess.Popen):
def communicate(self, input=None):
ret = subprocess.Popen.communicate(self, input)
- self.cleanup()
+ self._cleanup()
return ret
def is_running(self): return self.poll() is None
@@ -253,6 +259,7 @@ class Popen(subprocess.Popen):
def cmd_str(self): return " ".join([str(s) for s in self.cmd])
+
def checkenv(name):
value = os.getenv(name)
if not value: raise Exception("Environment variable %s is not set" % name)
@@ -307,7 +314,7 @@ class Broker(Popen):
cmd += ["--log-to-stderr=no"]
# Add default --log-enable arguments unless args already has --log arguments.
- if not [l for l in args if l.startswith("--log")]:
+ if not env_has_log_config() and not [l for l in args if l.startswith("--log")]:
args += ["--log-enable=info+"]
if test_store: cmd += ["--load-module", BrokerTest.test_store_lib,
@@ -443,10 +450,11 @@ def browse(session, queue, timeout=0, tr
finally:
r.close()
-def assert_browse(session, queue, expect_contents, timeout=0, transform=lambda m: m.content, msg="browse failed"):
+def assert_browse(session, queue, expect_contents, timeout=0, transform=lambda m: m.content, msg=None):
"""Assert that the contents of messages on queue (as retrieved
using session and timeout) exactly match the strings in
expect_contents"""
+ if msg is None: msg = "browse '%s' failed" % queue
actual_contents = browse(session, queue, timeout, transform=transform)
if msg: msg = "%s: %r != %r"%(msg, expect_contents, actual_contents)
assert expect_contents == actual_contents, msg
@@ -485,6 +493,18 @@ class BrokerTest(TestCase):
test_store_lib = os.getenv("TEST_STORE_LIB")
rootdir = os.getcwd()
+ PN_VERSION = (proton.VERSION_MAJOR, proton.VERSION_MINOR)
+ PN_TX_VERSION = (0, 9)
+
+ amqp_tx_supported = PN_VERSION >= PN_TX_VERSION
+
+ @classmethod
+ def amqp_tx_warning(cls):
+ if not cls.amqp_tx_supported:
+ print "WARNING: Cannot test transactions over AMQP 1.0, proton version %s.%s < %s.%s" % (cls.PN_VERSION + cls.PN_TX_VERSION)
+ return False
+ return True
+
def configure(self, config): self.config=config
def setUp(self):
@@ -497,8 +517,8 @@ class BrokerTest(TestCase):
if qpid_messaging and self.amqp_lib: default_protocol="amqp1.0"
else: default_protocol="amqp0-10"
self.protocol = defs.get("PROTOCOL") or default_protocol
- self.tx_protocol = "amqp0-10" # Transactions not yet supported over 1.0
-
+ self.tx_protocol = self.protocol
+ if not self.amqp_tx_supported: self.tx_protocol = "amqp0-10"
def tearDown(self):
err = []
@@ -529,15 +549,22 @@ class BrokerTest(TestCase):
self.teardown_add(p)
return p
- def broker(self, args=[], name=None, expect=EXPECT_RUNNING, wait=True, port=0, show_cmd=False):
+ def broker(self, args=[], name=None, expect=EXPECT_RUNNING, wait=True, port=0, show_cmd=False, **kw):
"""Create and return a broker ready for use"""
- b = Broker(self, args=args, name=name, expect=expect, port=port, show_cmd=show_cmd)
+ b = Broker(self, args=args, name=name, expect=expect, port=port, show_cmd=show_cmd, **kw)
if (wait):
try: b.ready()
except Exception, e:
raise RethrownException("Failed to start broker %s(%s): %s" % (b.name, b.log, e))
return b
+ def check_output(self, args, stdin=None):
+ p = self.popen(args, stdout=PIPE, stderr=STDOUT)
+ out = p.communicate(stdin)
+ if p.returncode != 0:
+ raise Exception("%s exit code %s, output:\n%s" % (args, p.returncode, out[0]))
+ return out[0]
+
def browse(self, *args, **kwargs): browse(*args, **kwargs)
def assert_browse(self, *args, **kwargs): assert_browse(*args, **kwargs)
def assert_browse_retry(self, *args, **kwargs): assert_browse_retry(*args, **kwargs)
Modified: qpid/trunk/qpid/cpp/src/tests/ha_test.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_test.py?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_test.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_test.py Fri Feb 27 16:37:06 2015
@@ -132,7 +132,7 @@ class HaBroker(Broker):
"--link-maintenance-interval=0.1",
"--ha-cluster=%s"%ha_cluster]
# Add default --log-enable arguments unless args already has --log arguments.
- if not [l for l in args if l.startswith("--log")]:
+ if not env_has_log_config() and not [l for l in args if l.startswith("--log")]:
args += ["--log-enable=info+", "--log-enable=debug+:ha::"]
if not [h for h in args if h.startswith("--link-heartbeat-interval")]:
args += ["--link-heartbeat-interval=%s"%(HaBroker.heartbeat)]
Modified: qpid/trunk/qpid/cpp/src/tests/ha_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_tests.py?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_tests.py Fri Feb 27 16:37:06 2015
@@ -1025,8 +1025,8 @@ class LongTests(HaBrokerTest):
"--broker", brokers[0].host_port(),
"--address", "q;{create:always}",
"--messages=1000",
- "--tx=10"
- # TODO aconway 2014-02-21: can't use amqp1.0 for transactions yet
+ "--tx=10",
+ "--connection-options={protocol:%s}" % self.tx_protocol
])
receiver = self.popen(
["qpid-receive",
@@ -1034,8 +1034,8 @@ class LongTests(HaBrokerTest):
"--address", "q;{create:always}",
"--messages=990",
"--timeout=10",
- "--tx=10"
- # TODO aconway 2014-02-21: can't use amqp1.0 for transactions yet
+ "--tx=10",
+ "--connection-options={protocol:%s}" % self.tx_protocol
])
self.assertEqual(sender.wait(), 0)
self.assertEqual(receiver.wait(), 0)
@@ -1268,7 +1268,7 @@ class StoreTests(HaBrokerTest):
"""Verify that a backup erases queue data from store recovery before
doing catch-up from the primary."""
if self.check_skip(): return
- cluster = HaCluster(self, 2, args=['--log-enable=trace+:ha', '--log-enable=trace+:Store'])
+ cluster = HaCluster(self, 2)
sn = cluster[0].connect(heartbeat=HaBroker.heartbeat).session()
s1 = sn.sender("q1;{create:always,node:{durable:true}}")
for m in ["foo","bar"]: s1.send(qm.Message(m, durable=True))
@@ -1532,7 +1532,7 @@ class TransactionTests(HaBrokerTest):
except qm.TransactionUnknown: pass
for b in cluster: self.assert_tx_clean(b)
try: tx.connection.close()
- except TransactionUnknown: pass # Occasionally get exception on close.
+ except qm.TransactionUnknown: pass # Occasionally get exception on close.
finally: l.restore()
def test_tx_no_backups(self):
@@ -1622,17 +1622,20 @@ class TransactionTests(HaBrokerTest):
import qpid_tests.broker_0_10
except ImportError:
raise Skipped("Tests not found")
-
cluster = HaCluster(self, 3)
- self.popen(["qpid-txtest", "-p%s"%cluster[0].port()]).assert_exit_ok()
+ if "QPID_PORT" in os.environ: del os.environ["QPID_PORT"]
+ self.popen(["qpid-txtest2", "--broker", cluster[0].host_port()]).assert_exit_ok()
+ print
self.popen(["qpid-python-test",
"-m", "qpid_tests.broker_0_10",
+ "-m", "qpid_tests.broker_1_0",
"-b", "localhost:%s"%(cluster[0].port()),
- "*.tx.*"]).assert_exit_ok()
+ "*.tx.*"], stdout=None, stderr=None).assert_exit_ok()
if __name__ == "__main__":
qpid_ha_exec = os.getenv("QPID_HA_EXEC")
if qpid_ha_exec and os.path.isfile(qpid_ha_exec):
+ BrokerTest.amqp_tx_warning()
outdir = "ha_tests.tmp"
shutil.rmtree(outdir, True)
os.execvp("qpid-python-test",
Added: qpid/trunk/qpid/cpp/src/tests/interop_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/interop_tests.py?rev=1662743&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/interop_tests.py (added)
+++ qpid/trunk/qpid/cpp/src/tests/interop_tests.py Fri Feb 27 16:37:06 2015
@@ -0,0 +1,220 @@
+#!/usr/bin/env python
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+"""
+A set of tests that can be run against a foreign AMQP 1.0 broker.
+
+RUNNING WITH A FOREIGN BROKER:
+
+1. Start the broker
+2. Create persistent queues named: interop-a interop-b interop-q tx-1 tx-2
+3. Export the environment variable QPID_INTEROP_URL with the URL to connect to your broker
+ in the form [user[:password]@]host[:port]
+4. From the build directory run this test:
+ ctest -VV -R interop_tests
+
+If QPID_INTEROP_URL is not set, a qpidd broker will be started for the test.
+"""
+
+import os, sys, shutil, subprocess
+import qpid_messaging as qm
+from brokertest import *
+
+URL='QPID_INTEROP_URL'
+
+class InteropTest(BrokerTest):
+
+ def setUp(self):
+ super(InteropTest, self).setUp()
+ self.url = os.environ[URL]
+ self.connect_opts = ['--broker', self.url, '--connection-options', '{protocol:amqp1.0}']
+
+ def connect(self, **kwargs):
+ """Python connection to interop URL"""
+ c = qm.Connection.establish(self.url, protocol='amqp1.0', **kwargs)
+ self.teardown_add(c)
+ return c
+
+ def drain(self, queue, connection=None):
+ """
+ Drain a queue to make sure it is empty. Throw away the messages.
+ """
+ c = connection or self.connect()
+ r = c.session().receiver(queue)
+ try:
+ while True:
+ r.fetch(timeout=0)
+ r.session.acknowledge()
+ except qm.Empty:
+ pass
+ r.close()
+
+ def clear_queue(self, queue, connection=None, properties=None, durable=False):
+ """
+ Make empty queue, prefix with self.id(). Create if needed, drain if needed
+ @return queue name.
+ """
+ queue = "interop-%s" % queue
+ c = connection or self.connect()
+ props = {'create':'always'}
+ if durable: props['node'] = {'durable':True}
+ if properties: props.update(properties)
+ self.drain("%s;%s" % (queue, props), c)
+ return queue
+
+
+class SimpleTest(InteropTest):
+ """Simple test to check the broker is responding."""
+
+ def test_send_receive_python(self):
+ c = self.connect()
+ q = self.clear_queue('q', c)
+ s = c.session()
+ s.sender(q).send('foo')
+ self.assertEqual('foo', s.receiver(q).fetch().content)
+
+ def test_send_receive_cpp(self):
+ q = self.clear_queue('q')
+ args = ['-b', self.url, '-a', q]
+ self.check_output(['qpid-send', '--content-string=cpp_foo'] + args)
+ self.assertEqual('cpp_foo', self.check_output(['qpid-receive'] + args).strip())
+
+
+class PythonTxTest(InteropTest):
+
+ def tx_simple_setup(self):
+ """Start a transaction, remove messages from queue a, add messages to queue b"""
+ c = self.connect()
+ qa, qb = self.clear_queue('a', c, durable=True), self.clear_queue('b', c, durable=True)
+
+ # Send messages to a, no transaction.
+ sa = c.session().sender(qa+";{create:always,node:{durable:true}}")
+ tx_msgs = ['x', 'y', 'z']
+ for m in tx_msgs: sa.send(qm.Message(content=m, durable=True))
+
+ # Receive messages from a, in transaction.
+ tx = c.session(transactional=True)
+ txr = tx.receiver(qa)
+ self.assertEqual(tx_msgs, [txr.fetch(1).content for i in xrange(3)])
+ tx.acknowledge()
+
+ # Send messages to b, transactional, mixed with non-transactional.
+ sb = c.session().sender(qb+";{create:always,node:{durable:true}}")
+ txs = tx.sender(qb)
+ msgs = [str(i) for i in xrange(3)]
+ for tx_m, m in zip(tx_msgs, msgs):
+ txs.send(tx_m);
+ sb.send(m)
+ tx.sync()
+ return tx, qa, qb
+
+ def test_tx_simple_commit(self):
+ tx, qa, qb = self.tx_simple_setup()
+ s = self.connect().session()
+ assert_browse(s, qa, [])
+ assert_browse(s, qb, ['0', '1', '2'])
+ tx.commit()
+ assert_browse(s, qa, [])
+ assert_browse(s, qb, ['0', '1', '2', 'x', 'y', 'z'])
+
+ def test_tx_simple_rollback(self):
+ tx, qa, qb = self.tx_simple_setup()
+ s = self.connect().session()
+ assert_browse(s, qa, [])
+ assert_browse(s, qb, ['0', '1', '2'])
+ tx.rollback()
+ assert_browse(s, qa, ['x', 'y', 'z'])
+ assert_browse(s, qb, ['0', '1', '2'])
+
+ def test_tx_sequence(self):
+ tx = self.connect().session(transactional=True)
+ notx = self.connect().session()
+ q = self.clear_queue('q', tx.connection, durable=True)
+ s = tx.sender(q)
+ r = tx.receiver(q)
+ s.send('a')
+ tx.commit()
+ assert_browse(notx, q, ['a'])
+ s.send('b')
+ tx.commit()
+ assert_browse(notx, q, ['a', 'b'])
+ self.assertEqual('a', r.fetch().content)
+ tx.acknowledge();
+ tx.commit()
+ assert_browse(notx, q, ['b'])
+ s.send('z')
+ tx.rollback()
+ assert_browse(notx, q, ['b'])
+ self.assertEqual('b', r.fetch().content)
+ tx.acknowledge();
+ tx.rollback()
+ assert_browse(notx, q, ['b'])
+
+
+class CppTxTest(InteropTest):
+
+ def test_txtest2(self):
+ self.popen(["qpid-txtest2"] + self.connect_opts).assert_exit_ok()
+
+ def test_send_receive(self):
+ q = self.clear_queue('q', durable=True)
+ sender = self.popen(["qpid-send",
+ "--address", q,
+ "--messages=100",
+ "--tx=10",
+ "--durable=yes"] + self.connect_opts)
+ receiver = self.popen(["qpid-receive",
+ "--address", q,
+ "--messages=90",
+ "--timeout=10",
+ "--tx=10"] + self.connect_opts)
+ sender.assert_exit_ok()
+ receiver.assert_exit_ok()
+ expect = [long(i) for i in range(91, 101)]
+ sn = lambda m: m.properties["sn"]
+ assert_browse(self.connect().session(), q, expect, transform=sn)
+
+
+if __name__ == "__main__":
+ if not BrokerTest.amqp_tx_supported:
+ BrokerTest.amqp_tx_warning()
+ print "Skipping interop_tests"
+ exit(0)
+ outdir = "interop_tests.tmp"
+ shutil.rmtree(outdir, True)
+ cmd = ["qpid-python-test", "-m", "interop_tests", "-DOUTDIR=%s"%outdir] + sys.argv[1:]
+ if "QPID_PORT" in os.environ: del os.environ["QPID_PORT"]
+ if os.environ.get(URL):
+ os.execvp(cmd[0], cmd)
+ else:
+ dir = os.getcwd()
+ class StartBroker(BrokerTest):
+ def start_qpidd(self): pass
+ test = StartBroker('start_qpidd')
+ class Config:
+ def __init__(self):
+ self.defines = { 'OUTDIR': outdir }
+ test.configure(Config())
+ test.setUp()
+ os.environ[URL] = test.broker().host_port()
+ os.chdir(dir)
+ p = subprocess.Popen(cmd)
+ status = p.wait()
+ test.tearDown()
+ sys.exit(status)
Propchange: qpid/trunk/qpid/cpp/src/tests/interop_tests.py
------------------------------------------------------------------------------
svn:executable = *
Modified: qpid/trunk/qpid/cpp/src/tests/qpid-receive.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/qpid-receive.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/qpid-receive.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/qpid-receive.cpp Fri Feb 27 16:37:06 2015
@@ -197,7 +197,7 @@ int main(int argc, char ** argv)
std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
Session session = opts.tx ? connection.createTransactionalSession() : connection.createSession();
Receiver receiver = session.createReceiver(opts.address);
- receiver.setCapacity(opts.capacity);
+ receiver.setCapacity(std::min(opts.capacity, opts.messages));
Message msg;
uint count = 0;
uint txCount = 0;
@@ -207,9 +207,9 @@ int main(int argc, char ** argv)
Reporter<ThroughputAndLatency> reporter(std::cout, opts.reportEvery, opts.reportHeader);
if (!opts.readyAddress.empty()) {
session.createSender(opts.readyAddress).send(msg);
- if (opts.tx)
- session.commit();
- }
+ if (opts.tx)
+ session.commit();
+ }
// For receive rate calculation
qpid::sys::AbsTime start = qpid::sys::now();
int64_t interval = 0;
@@ -290,6 +290,7 @@ int main(int argc, char ** argv)
connection.close();
return 0;
}
+ return 1;
} catch(const std::exception& error) {
std::cerr << "qpid-receive: " << error.what() << std::endl;
connection.close();
Modified: qpid/trunk/qpid/cpp/src/tests/qpid-send.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/qpid-send.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/qpid-send.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/qpid-send.cpp Fri Feb 27 16:37:06 2015
@@ -112,14 +112,14 @@ struct Options : public qpid::Options
log(argv0),
reportTotal(false),
reportEvery(0),
- reportHeader(true),
- sendRate(0),
- sequence(true),
- timestamp(true),
- groupPrefix("GROUP-"),
- groupSize(10),
- groupRandSize(false),
- groupInterleave(1)
+ reportHeader(true),
+ sendRate(0),
+ sequence(true),
+ timestamp(true),
+ groupPrefix("GROUP-"),
+ groupSize(10),
+ groupRandSize(false),
+ groupInterleave(1)
{
addOptions()
("broker,b", qpid::optValue(url, "URL"), "url of broker to connect to")
@@ -272,7 +272,7 @@ class MapContentGenerator : public Con
// tag each generated message with a group identifer
//
class GroupGenerator {
-public:
+ public:
GroupGenerator(const std::string& key,
const std::string& prefix,
const uint size,
@@ -351,7 +351,7 @@ int main(int argc, char ** argv)
try {
Options opts;
if (opts.parse(argc, argv)) {
- connection = Connection(opts.url, opts.connectionOptions);
+ connection = Connection(opts.url, opts.connectionOptions);
connection.open();
std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
Session session = opts.tx ? connection.createTransactionalSession() : connection.createSession();
@@ -447,6 +447,7 @@ int main(int argc, char ** argv)
connection.close();
return 0;
}
+ return 1;
} catch(const std::exception& error) {
std::cerr << "qpid-send: " << error.what() << std::endl;
connection.close();
Modified: qpid/trunk/qpid/cpp/src/tests/qpid-txtest2.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/qpid-txtest2.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/qpid-txtest2.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/qpid-txtest2.cpp Fri Feb 27 16:37:06 2015
@@ -353,10 +353,11 @@ int main(int argc, char** argv)
if (opts.init) controller.init();
if (opts.transfer) controller.transfer();
if (opts.check) return controller.check();
+ return 0;
}
- return 0;
+ return 1;
} catch(const std::exception& e) {
- std::cout << argv[0] << ": " << e.what() << std::endl;
+ std::cerr << argv[0] << ": " << e.what() << std::endl;
}
return 2;
}
Modified: qpid/trunk/qpid/cpp/src/tests/swig_python_tests
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/swig_python_tests?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/swig_python_tests (original)
+++ qpid/trunk/qpid/cpp/src/tests/swig_python_tests Fri Feb 27 16:37:06 2015
@@ -39,7 +39,8 @@ skip() {
}
start_broker() {
- QPID_PORT=$($QPIDD_EXEC --daemon --port 0 --interface 127.0.0.1 --no-data-dir $MODULES --auth no) || fail "Could not start broker"
+ rm -f swig_python_tests.log
+ QPID_PORT=$($QPIDD_EXEC --daemon --port 0 --interface 127.0.0.1 --no-data-dir $MODULES --auth no --log-to-file swig_python_tests.log) || fail "Could not start broker"
}
stop_broker() {
@@ -54,9 +55,9 @@ echo "Running swigged python tests using
export PYTHONPATH=$PYTHONPATH:$PYTHONPATH_SWIG
export QPID_USE_SWIG_CLIENT=1
-$QPID_PYTHON_TEST -m qpid.tests.messaging.message -m qpid_tests.broker_0_10.priority -m qpid_tests.broker_0_10.lvq -m qpid_tests.broker_0_10.new_api -b localhost:$QPID_PORT -I $srcdir/failing-amqp0-10-python-tests || FAILED=1
+$QPID_PYTHON_TEST -m qpid.tests.messaging.message -m qpid_tests.broker_0_10.priority -m qpid_tests.broker_0_10.lvq -m qpid_tests.broker_0_10.new_api -b localhost:$QPID_PORT -I $srcdir/failing-amqp0-10-python-tests $* || FAILED=1
if [[ -a $AMQP_LIB ]] ; then
- $QPID_PYTHON_TEST --define="protocol_version=amqp1.0" -m qpid_tests.broker_1_0 -m qpid_tests.broker_0_10.new_api -m assertions -m reject_release -m misc -m policies -b localhost:$QPID_PORT -I $srcdir/failing-amqp1.0-python-tests || FAILED=1
+ $QPID_PYTHON_TEST --define="protocol_version=amqp1.0" -m qpid_tests.broker_1_0 -m qpid_tests.broker_0_10.new_api -m assertions -m reject_release -m misc -m policies -b localhost:$QPID_PORT -I $srcdir/failing-amqp1.0-python-tests $* || FAILED=1
fi
stop_broker
if [[ $FAILED -eq 1 ]]; then
Modified: qpid/trunk/qpid/cpp/src/tests/test_env.sh.in
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/test_env.sh.in?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/test_env.sh.in (original)
+++ qpid/trunk/qpid/cpp/src/tests/test_env.sh.in Fri Feb 27 16:37:06 2015
@@ -20,14 +20,14 @@
absdir() { echo `cd $1 && pwd`; }
# Environment variables substituted by cmake.
-srcdir=`absdir @abs_srcdir@`
-builddir=`absdir @abs_builddir@`
-top_srcdir=`absdir @abs_top_srcdir@`
-top_builddir=`absdir @abs_top_builddir@`
-moduledir=$top_builddir/src@builddir_lib_suffix@
-pythonswigdir=$top_builddir/bindings/qpid/python/
-pythonswiglibdir=$top_builddir/bindings/qpid/python@builddir_lib_suffix@
-testmoduledir=$builddir@builddir_lib_suffix@
+export srcdir=`absdir @abs_srcdir@`
+export builddir=`absdir @abs_builddir@`
+export top_srcdir=`absdir @abs_top_srcdir@`
+export top_builddir=`absdir @abs_top_builddir@`
+export moduledir=$top_builddir/src@builddir_lib_suffix@
+export pythonswigdir=$top_builddir/bindings/qpid/python/
+export pythonswiglibdir=$top_builddir/bindings/qpid/python@builddir_lib_suffix@
+export testmoduledir=$builddir@builddir_lib_suffix@
export QPID_INSTALL_PREFIX=@prefix@
# Tools substituted by cmake
Modified: qpid/trunk/qpid/cpp/src/tests/test_store.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/test_store.cpp?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/test_store.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/test_store.cpp Fri Feb 27 16:37:06 2015
@@ -223,27 +223,18 @@ class TestStore : public NullMessageStor
const boost::intrusive_ptr<PersistableMessage>& pmsg,
const PersistableQueue& queue)
{
- qpid::broker::amqp_0_10::MessageTransfer* msg =
- dynamic_cast<qpid::broker::amqp_0_10::MessageTransfer*>(pmsg.get());
- assert(msg);
-
ostringstream o;
- o << "<enqueue " << queue.getName() << " " << getContent(msg);
+ string data = getContent(pmsg);
+ o << "<enqueue " << queue.getName() << " " << data;
if (tx) o << " tx=" << getId(*tx);
o << ">";
log(o.str());
// Dump the message if there is a dump file.
if (dump.get()) {
- msg->getFrames().getMethod()->print(*dump);
- *dump << endl << " ";
- msg->getFrames().getHeaders()->print(*dump);
- *dump << endl << " ";
- *dump << msg->getFrames().getContentSize() << endl;
+ *dump << "Message(" << data.size() << "): " << data << endl;
}
string logPrefix = "TestStore "+name+": ";
- // Check the message for special instructions for this store.
- string data = msg->getFrames().getContent();
Action action(data);
bool doComplete = true;
if (action.index && action.executeIn(name)) {
@@ -258,7 +249,7 @@ class TestStore : public NullMessageStor
QPID_LOG(error, logPrefix << "async-id needs argument: " << data);
break;
}
- asyncIds[action.args[0]] = msg;
+ asyncIds[action.args[0]] = pmsg;
QPID_LOG(debug, logPrefix << "delayed completion " << action.args[0]);
doComplete = false;
break;
@@ -284,7 +275,7 @@ class TestStore : public NullMessageStor
QPID_LOG(error, logPrefix << "unknown action: " << data);
}
}
- if (doComplete) msg->enqueueComplete();
+ if (doComplete) pmsg->enqueueComplete();
}
void dequeue(TransactionContext* tx,
Modified: qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/__init__.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/__init__.py?rev=1662743&r1=1662742&r2=1662743&view=diff
==============================================================================
--- qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/__init__.py (original)
+++ qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/__init__.py Fri Feb 27 16:37:06 2015
@@ -23,3 +23,4 @@ from general import *
from legacy_exchanges import *
from selector import *
from translation import *
+from tx import *
Added: qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/tx.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/tx.py?rev=1662743&view=auto
==============================================================================
--- qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/tx.py (added)
+++ qpid/trunk/qpid/tests/src/py/qpid_tests/broker_1_0/tx.py Fri Feb 27 16:37:06 2015
@@ -0,0 +1,264 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+from qpid.client import Client, Closed
+from qpid.queue import Empty
+from qpid.datatypes import Message, RangedSet
+from qpid.testlib import TestBase010
+
+class TxTests(TestBase010):
+ """
+ Tests for 'methods' on the amqp tx 'class'
+ """
+
+ def test_commit(self):
+ """
+ Test that commited publishes are delivered and commited acks are not re-delivered
+ """
+ session = self.session
+
+ #declare queues and create subscribers in the checking session
+ #to ensure that the queues are not auto-deleted too early:
+ self.declare_queues(["tx-commit-a", "tx-commit-b", "tx-commit-c"])
+ session.message_subscribe(queue="tx-commit-a", destination="qa")
+ session.message_subscribe(queue="tx-commit-b", destination="qb")
+ session.message_subscribe(queue="tx-commit-c", destination="qc")
+
+ #use a separate session for actual work
+ session2 = self.conn.session("worker", 2)
+ self.perform_txn_work(session2, "tx-commit-a", "tx-commit-b", "tx-commit-c")
+ session2.tx_commit()
+ session2.close()
+
+ session.tx_select()
+
+ self.enable_flow("qa")
+ queue_a = session.incoming("qa")
+
+ self.enable_flow("qb")
+ queue_b = session.incoming("qb")
+
+ self.enable_flow("qc")
+ queue_c = session.incoming("qc")
+
+ #check results
+ for i in range(1, 5):
+ msg = queue_c.get(timeout=1)
+ self.assertEqual("TxMessage %d" % i, msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_b.get(timeout=1)
+ self.assertEqual("TxMessage 6", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_a.get(timeout=1)
+ self.assertEqual("TxMessage 7", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ for q in [queue_a, queue_b, queue_c]:
+ try:
+ extra = q.get(timeout=1)
+ self.fail("Got unexpected message: " + extra.body)
+ except Empty: None
+
+ #cleanup
+ session.tx_commit()
+
+ def test_auto_rollback(self):
+ """
+ Test that a session closed with an open transaction is effectively rolled back
+ """
+ session = self.session
+ self.declare_queues(["tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c"])
+ session.message_subscribe(queue="tx-autorollback-a", destination="qa")
+ session.message_subscribe(queue="tx-autorollback-b", destination="qb")
+ session.message_subscribe(queue="tx-autorollback-c", destination="qc")
+
+ session2 = self.conn.session("worker", 2)
+ queue_a, queue_b, queue_c, ignore = self.perform_txn_work(session2, "tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c")
+
+ for q in [queue_a, queue_b, queue_c]:
+ try:
+ extra = q.get(timeout=1)
+ self.fail("Got unexpected message: " + extra.body)
+ except Empty: None
+
+ session2.close()
+
+ session.tx_select()
+
+ self.enable_flow("qa")
+ queue_a = session.incoming("qa")
+
+ self.enable_flow("qb")
+ queue_b = session.incoming("qb")
+
+ self.enable_flow("qc")
+ queue_c = session.incoming("qc")
+
+ #check results
+ for i in range(1, 5):
+ msg = queue_a.get(timeout=1)
+ self.assertEqual("Message %d" % i, msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_b.get(timeout=1)
+ self.assertEqual("Message 6", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_c.get(timeout=1)
+ self.assertEqual("Message 7", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ for q in [queue_a, queue_b, queue_c]:
+ try:
+ extra = q.get(timeout=1)
+ self.fail("Got unexpected message: " + extra.body)
+ except Empty: None
+
+ #cleanup
+ session.tx_commit()
+
+ def test_rollback(self):
+ """
+ Test that rolled back publishes are not delivered and rolled back acks are re-delivered
+ """
+ session = self.session
+ queue_a, queue_b, queue_c, consumed = self.perform_txn_work(session, "tx-rollback-a", "tx-rollback-b", "tx-rollback-c")
+
+ for q in [queue_a, queue_b, queue_c]:
+ try:
+ extra = q.get(timeout=1)
+ self.fail("Got unexpected message: " + extra.body)
+ except Empty: None
+
+ session.tx_rollback()
+
+ #need to release messages to get them redelivered now:
+ session.message_release(consumed)
+
+ #check results
+ for i in range(1, 5):
+ msg = queue_a.get(timeout=1)
+ self.assertEqual("Message %d" % i, msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_b.get(timeout=1)
+ self.assertEqual("Message 6", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ msg = queue_c.get(timeout=1)
+ self.assertEqual("Message 7", msg.body)
+ session.message_accept(RangedSet(msg.id))
+
+ for q in [queue_a, queue_b, queue_c]:
+ try:
+ extra = q.get(timeout=1)
+ self.fail("Got unexpected message: " + extra.body)
+ except Empty: None
+
+ #cleanup
+ session.tx_commit()
+
+ def perform_txn_work(self, session, name_a, name_b, name_c):
+ """
+ Utility method that does some setup and some work under a transaction. Used for testing both
+ commit and rollback
+ """
+ #setup:
+ self.declare_queues([name_a, name_b, name_c])
+
+ key = "my_key_" + name_b
+ topic = "my_topic_" + name_c
+
+ session.exchange_bind(queue=name_b, exchange="amq.direct", binding_key=key)
+ session.exchange_bind(queue=name_c, exchange="amq.topic", binding_key=topic)
+
+ dp = session.delivery_properties(routing_key=name_a)
+ for i in range(1, 5):
+ mp = session.message_properties(message_id="msg%d" % i)
+ session.message_transfer(message=Message(dp, mp, "Message %d" % i))
+
+ dp = session.delivery_properties(routing_key=key)
+ mp = session.message_properties(message_id="msg6")
+ session.message_transfer(destination="amq.direct", message=Message(dp, mp, "Message 6"))
+
+ dp = session.delivery_properties(routing_key=topic)
+ mp = session.message_properties(message_id="msg7")
+ session.message_transfer(destination="amq.topic", message=Message(dp, mp, "Message 7"))
+
+ session.tx_select()
+
+ #consume and ack messages
+ acked = RangedSet()
+ self.subscribe(session, queue=name_a, destination="sub_a")
+ queue_a = session.incoming("sub_a")
+ for i in range(1, 5):
+ msg = queue_a.get(timeout=1)
+ acked.add(msg.id)
+ self.assertEqual("Message %d" % i, msg.body)
+
+ self.subscribe(session, queue=name_b, destination="sub_b")
+ queue_b = session.incoming("sub_b")
+ msg = queue_b.get(timeout=1)
+ self.assertEqual("Message 6", msg.body)
+ acked.add(msg.id)
+
+ sub_c = self.subscribe(session, queue=name_c, destination="sub_c")
+ queue_c = session.incoming("sub_c")
+ msg = queue_c.get(timeout=1)
+ self.assertEqual("Message 7", msg.body)
+ acked.add(msg.id)
+
+ session.message_accept(acked)
+
+ dp = session.delivery_properties(routing_key=topic)
+ #publish messages
+ for i in range(1, 5):
+ mp = session.message_properties(message_id="tx-msg%d" % i)
+ session.message_transfer(destination="amq.topic", message=Message(dp, mp, "TxMessage %d" % i))
+
+ dp = session.delivery_properties(routing_key=key)
+ mp = session.message_properties(message_id="tx-msg6")
+ session.message_transfer(destination="amq.direct", message=Message(dp, mp, "TxMessage 6"))
+
+ dp = session.delivery_properties(routing_key=name_a)
+ mp = session.message_properties(message_id="tx-msg7")
+ session.message_transfer(message=Message(dp, mp, "TxMessage 7"))
+ return queue_a, queue_b, queue_c, acked
+
+ def declare_queues(self, names, session=None):
+ session = session or self.session
+ for n in names:
+ session.queue_declare(queue=n, auto_delete=True)
+
+ def subscribe(self, session=None, **keys):
+ session = session or self.session
+ consumer_tag = keys["destination"]
+ session.message_subscribe(**keys)
+ session.message_flow(destination=consumer_tag, unit=session.credit_unit.message, value=0xFFFFFFFFL)
+ session.message_flow(destination=consumer_tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL)
+
+ def enable_flow(self, tag, session=None):
+ session = session or self.session
+ session.message_flow(destination=tag, unit=session.credit_unit.message, value=0xFFFFFFFFL)
+ session.message_flow(destination=tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL)
+
+ def complete(self, session, msg):
+ session.receiver._completed.add(msg.id)#TODO: this may be done automatically
+ session.channel.session_completed(session.receiver._completed)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org