diff options
author | Alan Conway <aconway@apache.org> | 2015-02-27 16:37:06 +0000 |
---|---|---|
committer | Alan Conway <aconway@apache.org> | 2015-02-27 16:37:06 +0000 |
commit | 3aaa53e9103b6019c9e31d15186b12a95a1993be (patch) | |
tree | f5950c063ff08f574c808023ece7745739ca7027 /qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp | |
parent | 9c9f0e2c935d11c0f8d1ebddf1bbb78c3c22c606 (diff) | |
download | qpid-python-3aaa53e9103b6019c9e31d15186b12a95a1993be.tar.gz |
QPID-4710: [AMQP 1.0] Support for transactions in qpid::messaging C++ client.
Implements the "transactional retire and settle immediately" option for
transactions as specified in AMQP 1.0 in the qpid::messaging C++ client.
NOTE: Transactions over AMQP 1.0 require proton 0.9 or greater. With older
versions, attempting a transactions over AMQP 1.0 will raise a link-detached
exception "Node not found: tx-transaction"
1. Added descriptor list to Variant with support in Encoder and PnData.
Required to support transactions, need to be able to create described lists.
Variant changes are source and binary compatible.
A Variant now has a Variant::List of descripors which can be numeric or string.
Nested descriptors are implemented by putting multiple descriptors in the list.
Other minor changes:
- Variant refactor: don't delete impl on every assignment.
- Add Variant constructors that take a string encoding.
(new constructors, not defaulted arguments, so the change is binary and source compatible.)
- Growable buffer support for Encoder.
- Printing described Variant prints descriptors in form @descriptor value
2. Added transaction support to AMQP 1.0 client code
Added messaging/amqp/Transaction.h,cpp: transaction logic
- communicate with coordinator, send declare/dischange messages.
- add tx state info to transfers and acknowledgements.
- Sync session after discharge.
- A transactional session automatically acks any message retrieved by fetch/get
to bring them into the transaction. This is consistent the 0-10 client.
Minor fixes to existing client code:
- Fix use of pn_drain API in C++ client to work with C++ and Java brokers.
- Make amqp::Exception derive from qpid::Exception
3. Fixes to existing broker code:
- Incoming.cpp fix: start async completion before processing message.
- Delay accept of dischage message till commit is complete.
- newSession - handle failover during session creation.
4. Added tests
interop_tests.py: transaction tests that can run against an external broker, see comments.
ha_tests.py: Enable transaction tests over AMQP 1.0.
Minor test fixes:
- brokertest.py don't set default logging if QPID_LOG env vars set.
- brokertest.py Pass kwargs to broker() create function.
- qpid-receive: capacity should never be larger than message count.
- Accept user:pass as well as user/pass in Url.
- brokertest.py: Always do a ready() check on all brokers.
If proton < 0.9 is used, transaction tests will be skipped or will downgrade to
the amqp0-10 protocol with a printed warning.
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1662743 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp')
-rw-r--r-- | qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp | 155 |
1 files changed, 155 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp b/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp new file mode 100644 index 0000000000..754b00d802 --- /dev/null +++ b/qpid/cpp/src/qpid/messaging/amqp/Transaction.cpp @@ -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 |