| /* |
| * |
| * 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 "qpid/broker/SessionState.h" |
| |
| #include "qpid/broker/AsyncCommandCallback.h" |
| #include "qpid/broker/Broker.h" |
| #include "qpid/broker/amqp_0_10/Connection.h" |
| #include "qpid/broker/DeliverableMessage.h" |
| #include "qpid/broker/DtxAck.h" |
| #include "qpid/broker/DtxTimeout.h" |
| #include "qpid/broker/Message.h" |
| #include "qpid/broker/Queue.h" |
| #include "qpid/broker/Selector.h" |
| #include "qpid/broker/SessionContext.h" |
| #include "qpid/broker/SessionOutputException.h" |
| #include "qpid/broker/TransactionObserver.h" |
| #include "qpid/broker/TxAccept.h" |
| #include "qpid/broker/amqp_0_10/MessageTransfer.h" |
| #include "qpid/framing/reply_exceptions.h" |
| #include "qpid/framing/MessageTransferBody.h" |
| #include "qpid/framing/SequenceSet.h" |
| #include "qpid/framing/IsInSequenceSet.h" |
| #include "qpid/log/Statement.h" |
| #include "qpid/management/ManagementAgent.h" |
| #include "qpid/ptr_map.h" |
| #include "qpid/broker/AclModule.h" |
| #include "qpid/broker/FedOps.h" |
| #include "qpid/sys/ConnectionOutputHandler.h" |
| |
| #include <boost/bind.hpp> |
| #include <boost/format.hpp> |
| #include <boost/tuple/tuple_comparison.hpp> |
| |
| #include <iostream> |
| #include <sstream> |
| #include <algorithm> |
| #include <functional> |
| |
| #include <assert.h> |
| |
| namespace { |
| const std::string X_SCOPE("x-scope"); |
| const std::string SESSION("session"); |
| } |
| |
| namespace qpid { |
| namespace broker { |
| |
| using namespace std; |
| using boost::intrusive_ptr; |
| using boost::shared_ptr; |
| using boost::bind; |
| using namespace qpid::broker; |
| using namespace qpid::framing; |
| using namespace qpid::sys; |
| using qpid::ptr_map_ptr; |
| using qpid::management::ManagementAgent; |
| using qpid::management::ManagementObject; |
| using qpid::management::Manageable; |
| using qpid::management::Args; |
| namespace _qmf = qmf::org::apache::qpid::broker; |
| |
| SemanticState::SemanticState(SessionState& ss) |
| : session(ss), |
| tagGenerator("sgen"), |
| dtxSelected(false), |
| authMsg(getSession().getBroker().isAuthenticating() && !getSession().getConnection().isFederationLink()), |
| userID(getSession().getConnection().getUserId()), |
| closeComplete(false), |
| connectionId(getSession().getConnection().getMgmtId()) |
| {} |
| |
| SemanticState::~SemanticState() { |
| closed(); |
| } |
| |
| void SemanticState::closed() { |
| if (!closeComplete) { |
| //prevent requeued messages being redelivered to consumers |
| for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) { |
| disable(i->second); |
| } |
| if (dtxBuffer.get()) { |
| dtxBuffer->fail(); |
| } |
| unbindSessionBindings(); |
| requeue(); |
| |
| //now unsubscribe, which may trigger queue deletion and thus |
| //needs to occur after the requeueing of unacked messages |
| for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) { |
| cancel(i->second); |
| } |
| closeComplete = true; |
| if (txBuffer) txBuffer->rollback(); |
| } |
| } |
| |
| bool SemanticState::exists(const string& consumerTag){ |
| return consumers.find(consumerTag) != consumers.end(); |
| } |
| |
| namespace { |
| const std::string SEPARATOR("::"); |
| } |
| |
| void SemanticState::consume(const string& tag, |
| Queue::shared_ptr queue, bool ackRequired, bool acquire, |
| bool exclusive, const string& resumeId, uint64_t resumeTtl, |
| const FieldTable& arguments) |
| { |
| // "tag" is only guaranteed to be unique to this session (see AMQP 0-10 Message.subscribe, destination). |
| // Create a globally unique name so the broker can identify individual consumers |
| std::string name = session.getSessionId().str() + SEPARATOR + tag; |
| const ConsumerFactories::Factories& cf( |
| session.getBroker().getConsumerFactories().get()); |
| ConsumerImpl::shared_ptr c; |
| for (ConsumerFactories::Factories::const_iterator i = cf.begin(); i != cf.end() && !c; ++i) |
| c = (*i)->create(this, name, queue, ackRequired, acquire, exclusive, tag, |
| resumeId, resumeTtl, arguments); |
| if (!c) // Create plain consumer |
| c = ConsumerImpl::shared_ptr( |
| new ConsumerImpl(this, name, queue, ackRequired, acquire ? CONSUMER : BROWSER, exclusive, tag, |
| resumeId, resumeTtl, arguments)); |
| queue->consume(c, exclusive, arguments, connectionId, userID);//may throw exception |
| consumers[tag] = c; |
| } |
| |
| bool SemanticState::cancel(const string& tag) |
| { |
| ConsumerImplMap::iterator i = consumers.find(tag); |
| if (i != consumers.end()) { |
| cancel(i->second); |
| consumers.erase(i); |
| //should cancel all unacked messages for this consumer so that |
| //they are not redelivered on recovery |
| for_each(unacked.begin(), unacked.end(), boost::bind(&DeliveryRecord::cancel, _1, tag)); |
| //can also remove any records that are now redundant |
| DeliveryRecords::iterator removed = |
| remove_if(unacked.begin(), unacked.end(), bind(&DeliveryRecord::isRedundant, _1)); |
| unacked.erase(removed, unacked.end()); |
| getSession().setUnackedCount(unacked.size()); |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| |
| void SemanticState::startTx() |
| { |
| accumulatedAck.clear(); |
| txBuffer = boost::intrusive_ptr<TxBuffer>(new TxBuffer()); |
| session.getBroker().getBrokerObservers().startTx(txBuffer); |
| session.startTx(); //just to update statistics |
| } |
| |
| namespace { |
| struct StartTxOnExit { |
| SemanticState& session; |
| StartTxOnExit(SemanticState& ss) : session(ss) {} |
| ~StartTxOnExit() { session.startTx(); } |
| }; |
| } // namespace |
| |
| void SemanticState::commit(MessageStore* const store) |
| { |
| if (!txBuffer) throw |
| CommandInvalidException( |
| QPID_MSG("Session has not been selected for use with transactions")); |
| // Start a new TX regardless of outcome of this one. |
| StartTxOnExit e(*this); |
| session.getCurrentCommand().setCompleteSync(false); // Async completion |
| txBuffer->begin(); // Begin async completion. |
| session.commitTx(); //just to update statistics |
| TxOp::shared_ptr txAck(static_cast<TxOp*>(new TxAccept(accumulatedAck, unacked))); |
| txBuffer->enlist(txAck); |
| // In a HA cluster, tx.commit may complete asynchronously. |
| txBuffer->startCommit(store); |
| AsyncCommandCallback callback( |
| session, |
| boost::bind(&TxBuffer::endCommit, txBuffer, store), |
| true // This is a sync point |
| ); |
| txBuffer->end(callback); |
| } |
| |
| void SemanticState::rollback() |
| { |
| if (!txBuffer) |
| throw CommandInvalidException(QPID_MSG("Session has not been selected for use with transactions")); |
| StartTxOnExit e(*this); |
| session.rollbackTx(); // Just to update statistics |
| txBuffer->rollback(); |
| } |
| |
| void SemanticState::selectDtx() |
| { |
| dtxSelected = true; |
| } |
| |
| void SemanticState::startDtx(const std::string& xid, DtxManager& mgr, bool join) |
| { |
| if (!dtxSelected) { |
| throw CommandInvalidException(QPID_MSG("Session has not been selected for use with dtx")); |
| } |
| dtxBuffer = new DtxBuffer(xid); |
| txBuffer = dtxBuffer; |
| session.getBroker().getBrokerObservers().startDtx(dtxBuffer); |
| if (join) { |
| mgr.join(xid, dtxBuffer); |
| } else { |
| mgr.start(xid, dtxBuffer); |
| } |
| } |
| |
| void SemanticState::endDtx(const std::string& xid, bool fail) |
| { |
| if (!dtxBuffer) { |
| throw IllegalStateException(QPID_MSG("xid " << xid << " not associated with this session")); |
| } |
| if (dtxBuffer->getXid() != xid) { |
| throw CommandInvalidException( |
| QPID_MSG("xid specified on start was " << dtxBuffer->getXid() << ", but " << xid << " specified on end")); |
| |
| } |
| |
| txBuffer = 0;//ops on this session no longer transactional |
| |
| checkDtxTimeout(); |
| if (fail) { |
| dtxBuffer->fail(); |
| } else { |
| dtxBuffer->markEnded(); |
| } |
| dtxBuffer = 0; |
| } |
| |
| void SemanticState::suspendDtx(const std::string& xid) |
| { |
| if (dtxBuffer->getXid() != xid) { |
| throw CommandInvalidException( |
| QPID_MSG("xid specified on start was " << dtxBuffer->getXid() << ", but " << xid << " specified on suspend")); |
| } |
| txBuffer = 0;//ops on this session no longer transactional |
| |
| checkDtxTimeout(); |
| dtxBuffer->setSuspended(true); |
| suspendedXids[xid] = dtxBuffer; |
| dtxBuffer = 0; |
| } |
| |
| void SemanticState::resumeDtx(const std::string& xid) |
| { |
| if (!dtxSelected) { |
| throw CommandInvalidException(QPID_MSG("Session has not been selected for use with dtx")); |
| } |
| |
| dtxBuffer = suspendedXids[xid]; |
| if (!dtxBuffer) { |
| throw CommandInvalidException(QPID_MSG("xid " << xid << " not attached")); |
| } else { |
| suspendedXids.erase(xid); |
| } |
| |
| if (dtxBuffer->getXid() != xid) { |
| throw CommandInvalidException( |
| QPID_MSG("xid specified on start was " << dtxBuffer->getXid() << ", but " << xid << " specified on resume")); |
| |
| } |
| if (!dtxBuffer->isSuspended()) { |
| throw CommandInvalidException(QPID_MSG("xid " << xid << " not suspended")); |
| } |
| |
| checkDtxTimeout(); |
| dtxBuffer->setSuspended(false); |
| txBuffer = dtxBuffer; |
| } |
| |
| void SemanticState::checkDtxTimeout() |
| { |
| if (dtxBuffer->isExpired()) { |
| dtxBuffer = 0; |
| throw DtxTimeoutException(); |
| } |
| } |
| |
| void SemanticState::record(const DeliveryRecord& delivery) |
| { |
| unacked.push_back(delivery); |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| const std::string QPID_SYNC_FREQUENCY("qpid.sync_frequency"); |
| const std::string APACHE_SELECTOR("x-apache-selector"); |
| |
| SemanticStateConsumerImpl::SemanticStateConsumerImpl(SemanticState* _parent, |
| const string& _name, |
| Queue::shared_ptr _queue, |
| bool ack, |
| SubscriptionType type, |
| bool _exclusive, |
| const string& _tag, |
| const string& _resumeId, |
| uint64_t _resumeTtl, |
| const framing::FieldTable& _arguments |
| |
| ) : |
| Consumer(_name, type, _tag), |
| parent(_parent), |
| queue(_queue), |
| ackExpected(ack), |
| acquire(type == CONSUMER), |
| blocked(true), |
| exclusive(_exclusive), |
| resumeId(_resumeId), |
| selector(returnSelector(_arguments.getAsString(APACHE_SELECTOR))), |
| resumeTtl(_resumeTtl), |
| arguments(_arguments), |
| notifyEnabled(true), |
| syncFrequency(_arguments.getAsInt(QPID_SYNC_FREQUENCY)), |
| deliveryCount(0), |
| protocols(parent->getSession().getBroker().getProtocolRegistry()) |
| { |
| if (parent != 0 && queue.get() != 0 && queue->GetManagementObject() !=0) |
| { |
| ManagementAgent* agent = parent->session.getBroker().getManagementAgent(); |
| qpid::management::Manageable* ms = dynamic_cast<qpid::management::Manageable*> (&(parent->session)); |
| |
| if (agent != 0) |
| { |
| mgmtObject = _qmf::Subscription::shared_ptr(new _qmf::Subscription(agent, this, ms , queue->GetManagementObject()->getObjectId(), getTag(), |
| !acquire, ackExpected, exclusive, ManagementAgent::toMap(arguments))); |
| agent->addObject (mgmtObject); |
| mgmtObject->set_creditMode("WINDOW"); |
| } |
| } |
| } |
| |
| ManagementObject::shared_ptr SemanticStateConsumerImpl::GetManagementObject (void) const |
| { |
| return mgmtObject; |
| } |
| |
| Manageable::status_t SemanticStateConsumerImpl::ManagementMethod (uint32_t methodId, Args&, string&) |
| { |
| Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD; |
| |
| QPID_LOG (debug, "Queue::ManagementMethod [id=" << methodId << "]"); |
| |
| return status; |
| } |
| |
| |
| OwnershipToken* SemanticStateConsumerImpl::getSession() |
| { |
| return &(parent->session); |
| } |
| |
| bool SemanticStateConsumerImpl::deliver(const QueueCursor& cursor, const Message& msg) |
| { |
| return deliver(cursor, msg, shared_from_this()); |
| } |
| bool SemanticStateConsumerImpl::deliver(const QueueCursor& cursor, const Message& msg, boost::shared_ptr<Consumer> consumer) |
| { |
| allocateCredit(msg); |
| boost::intrusive_ptr<const amqp_0_10::MessageTransfer> transfer = protocols.translate(msg); |
| DeliveryRecord record(cursor, msg.getSequence(), msg.getReplicationId(), queue, getTag(), |
| consumer, acquire, !ackExpected, credit.isWindowMode(), transfer->getRequiredCredit()); |
| bool sync = syncFrequency && ++deliveryCount >= syncFrequency; |
| if (sync) deliveryCount = 0;//reset |
| |
| record.setId(parent->session.deliver(*transfer, getTag(), msg.isRedelivered(), msg.getTtl(), |
| ackExpected ? message::ACCEPT_MODE_EXPLICIT : message::ACCEPT_MODE_NONE, |
| acquire ? message::ACQUIRE_MODE_PRE_ACQUIRED : message::ACQUIRE_MODE_NOT_ACQUIRED, |
| msg.getAnnotations(), |
| sync)); |
| if (credit.isWindowMode() || ackExpected || !acquire) { |
| parent->record(record); |
| } |
| if (acquire && !ackExpected) { // auto acquire && auto accept |
| queue->dequeue(0 /*ctxt*/, cursor); |
| } |
| if (mgmtObject) { mgmtObject->inc_delivered(); } |
| return true; |
| } |
| |
| bool SemanticStateConsumerImpl::filter(const Message& msg) |
| { |
| return !selector || selector->filter(msg); |
| } |
| |
| bool SemanticStateConsumerImpl::accept(const Message& msg) |
| { |
| // TODO aconway 2009-06-08: if we have byte & message credit but |
| // checkCredit fails because the message is to big, we should |
| // remain on queue's listener list for possible smaller messages |
| // in future. |
| // |
| blocked = !checkCredit(msg); |
| return !blocked; |
| } |
| |
| namespace { |
| struct ConsumerName { |
| const SemanticStateConsumerImpl& consumer; |
| ConsumerName(const SemanticStateConsumerImpl& ci) : consumer(ci) {} |
| }; |
| |
| ostream& operator<<(ostream& o, const ConsumerName& pc) { |
| return o << pc.consumer.getTag() << " on " |
| << pc.consumer.getParent().getSession().getSessionId(); |
| } |
| } |
| |
| void SemanticStateConsumerImpl::allocateCredit(const Message& msg) |
| { |
| Credit original = credit; |
| boost::intrusive_ptr<const amqp_0_10::MessageTransfer> transfer = protocols.translate(msg); |
| credit.consume(1, transfer->getRequiredCredit()); |
| QPID_LOG(debug, "Credit allocated for " << ConsumerName(*this) |
| << ", was " << original << " now " << credit); |
| |
| } |
| |
| bool SemanticStateConsumerImpl::checkCredit(const Message& msg) |
| { |
| boost::intrusive_ptr<const amqp_0_10::MessageTransfer> transfer = protocols.translate(msg); |
| bool enoughCredit = credit.check(1, transfer->getRequiredCredit()); |
| QPID_LOG(debug, "Subscription " << ConsumerName(*this) << " has " << (enoughCredit ? "sufficient " : "insufficient") |
| << " credit for message of " << transfer->getRequiredCredit() << " bytes: " |
| << credit); |
| return enoughCredit; |
| } |
| |
| SemanticStateConsumerImpl::~SemanticStateConsumerImpl() |
| { |
| if (mgmtObject != 0) { |
| mgmtObject->debugStats("destroying"); |
| mgmtObject->resourceDestroy (); |
| } |
| } |
| |
| void SemanticState::disable(ConsumerImpl::shared_ptr c) |
| { |
| c->disableNotify(); |
| if (session.isAttached()) |
| session.getConnection().removeOutputTask(c.get()); |
| } |
| |
| void SemanticState::cancel(ConsumerImpl::shared_ptr c) |
| { |
| disable(c); |
| Queue::shared_ptr queue = c->getQueue(); |
| if(queue) { |
| queue->cancel(c, connectionId, userID); |
| } |
| c->cancel(); |
| } |
| |
| TxBuffer* SemanticState::getTxBuffer() |
| { |
| return txBuffer.get(); |
| } |
| |
| void SemanticState::route(Message& msg, Deliverable& strategy) { |
| std::string exchangeName = qpid::broker::amqp_0_10::MessageTransfer::get(msg).getExchangeName(); |
| if (!cacheExchange || cacheExchange->getName() != exchangeName || cacheExchange->isDestroyed()) |
| cacheExchange = session.getBroker().getExchanges().get(exchangeName); |
| |
| /* verify the userid if specified: */ |
| std::string id = msg.getUserId(); |
| if (authMsg && !id.empty() && !session.getConnection().isAuthenticatedUser(id)) |
| { |
| QPID_LOG(debug, "authorised user id : " << userID << " but user id in message declared as " << id); |
| throw UnauthorizedAccessException(QPID_MSG("authorised user id : " << userID << " but user id in message declared as " << id)); |
| } |
| |
| AclModule* acl = getSession().getBroker().getAcl(); |
| if (acl && acl->doTransferAcl()) |
| { |
| if (!acl->authorise(getSession().getConnection().getUserId(),acl::ACT_PUBLISH,acl::OBJ_EXCHANGE,exchangeName, msg.getRoutingKey() )) |
| throw UnauthorizedAccessException(QPID_MSG(userID << " cannot publish to " << |
| exchangeName << " with routing-key " << msg.getRoutingKey())); |
| } |
| |
| cacheExchange->route(strategy); |
| |
| if (!strategy.delivered) { |
| //TODO:if discard-unroutable, just drop it |
| //TODO:else if accept-mode is explicit, reject it |
| //else route it to alternate exchange |
| if (cacheExchange->getAlternate()) { |
| cacheExchange->getAlternate()->route(strategy); |
| } |
| } |
| |
| } |
| |
| void SemanticState::requestDispatch() |
| { |
| for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) |
| i->second->requestDispatch(); |
| } |
| |
| void SemanticStateConsumerImpl::requestDispatch() |
| { |
| if (blocked) { |
| parent->session.getConnection().addOutputTask(this); |
| parent->session.getConnection().activateOutput(); |
| blocked = false; |
| } |
| } |
| |
| bool SemanticState::complete(DeliveryRecord& delivery) |
| { |
| ConsumerImplMap::iterator i = consumers.find(delivery.getTag()); |
| if (i != consumers.end()) { |
| i->second->complete(delivery); |
| } |
| return delivery.isRedundant(); |
| } |
| |
| void SemanticStateConsumerImpl::complete(DeliveryRecord& delivery) |
| { |
| if (!delivery.isComplete()) { |
| delivery.complete(); |
| if (credit.isWindowMode()) { |
| credit.moveWindow(1, delivery.getCredit()); |
| } |
| } |
| } |
| |
| void SemanticState::requeue() |
| { |
| //take copy and clear unacked as requeue may result in redelivery to this session |
| //which will in turn result in additions to unacked |
| DeliveryRecords copy = unacked; |
| unacked.clear(); |
| for_each(copy.rbegin(), copy.rend(), mem_fun_ref(&DeliveryRecord::requeue)); |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| |
| SessionContext& SemanticState::getSession() { return session; } |
| const SessionContext& SemanticState::getSession() const { return session; } |
| |
| |
| const SemanticStateConsumerImpl::shared_ptr SemanticState::find(const std::string& destination) const |
| { |
| ConsumerImpl::shared_ptr consumer; |
| if (!find(destination, consumer)) { |
| throw NotFoundException(QPID_MSG("Unknown destination " << destination << " session=" << session.getSessionId())); |
| } else { |
| return consumer; |
| } |
| } |
| |
| bool SemanticState::find(const std::string& destination, ConsumerImpl::shared_ptr& consumer) const |
| { |
| // @todo KAG gsim: shouldn't the consumers map be locked???? |
| ConsumerImplMap::const_iterator i = consumers.find(destination); |
| if (i == consumers.end()) { |
| return false; |
| } |
| consumer = i->second; |
| return true; |
| } |
| |
| void SemanticState::setWindowMode(const std::string& destination) |
| { |
| find(destination)->setWindowMode(); |
| } |
| |
| void SemanticState::setCreditMode(const std::string& destination) |
| { |
| find(destination)->setCreditMode(); |
| } |
| |
| void SemanticState::addByteCredit(const std::string& destination, uint32_t value) |
| { |
| ConsumerImpl::shared_ptr c = find(destination); |
| c->addByteCredit(value); |
| c->requestDispatch(); |
| } |
| |
| |
| void SemanticState::addMessageCredit(const std::string& destination, uint32_t value) |
| { |
| ConsumerImpl::shared_ptr c = find(destination); |
| c->addMessageCredit(value); |
| c->requestDispatch(); |
| } |
| |
| void SemanticState::flush(const std::string& destination) |
| { |
| find(destination)->flush(); |
| } |
| |
| |
| void SemanticState::stop(const std::string& destination) |
| { |
| find(destination)->stop(); |
| } |
| |
| void SemanticStateConsumerImpl::setWindowMode() |
| { |
| credit.setWindowMode(true); |
| if (mgmtObject){ |
| mgmtObject->set_creditMode("WINDOW"); |
| } |
| } |
| |
| void SemanticStateConsumerImpl::setCreditMode() |
| { |
| credit.setWindowMode(false); |
| if (mgmtObject){ |
| mgmtObject->set_creditMode("CREDIT"); |
| } |
| } |
| |
| void SemanticStateConsumerImpl::addByteCredit(uint32_t value) |
| { |
| credit.addByteCredit(value); |
| } |
| |
| void SemanticStateConsumerImpl::addMessageCredit(uint32_t value) |
| { |
| credit.addMessageCredit(value); |
| } |
| |
| bool SemanticStateConsumerImpl::haveCredit() |
| { |
| if (credit) { |
| return true; |
| } else { |
| blocked = true; |
| return false; |
| } |
| } |
| |
| bool SemanticStateConsumerImpl::doDispatch() |
| { |
| return queue->dispatch(shared_from_this()); |
| } |
| |
| void SemanticStateConsumerImpl::flush() |
| { |
| while(haveCredit() && doDispatch()) |
| ; |
| credit.cancel(); |
| } |
| |
| void SemanticStateConsumerImpl::stop() |
| { |
| credit.cancel(); |
| } |
| |
| Queue::shared_ptr SemanticState::getQueue(const string& name) const { |
| Queue::shared_ptr queue; |
| if (name.empty()) { |
| throw NotAllowedException(QPID_MSG("No queue name specified.")); |
| } else { |
| queue = session.getBroker().getQueues().get(name); |
| } |
| return queue; |
| } |
| |
| AckRange SemanticState::findRange(DeliveryId first, DeliveryId last) |
| { |
| return DeliveryRecord::findRange(unacked, first, last); |
| } |
| |
| void SemanticState::acquire(DeliveryId first, DeliveryId last, DeliveryIds& acquired) |
| { |
| AckRange range = findRange(first, last); |
| for_each(range.start, range.end, AcquireFunctor(acquired)); |
| } |
| |
| void SemanticState::release(DeliveryId first, DeliveryId last, bool setRedelivered) |
| { |
| AckRange range = findRange(first, last); |
| //release results in the message being added to the head so want |
| //to release in reverse order to keep the original transfer order |
| DeliveryRecords::reverse_iterator start(range.end); |
| DeliveryRecords::reverse_iterator end(range.start); |
| for_each(start, end, boost::bind(&DeliveryRecord::release, _1, setRedelivered)); |
| |
| DeliveryRecords::iterator removed = |
| remove_if(range.start, range.end, bind(&DeliveryRecord::isRedundant, _1)); |
| unacked.erase(removed, range.end); |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| void SemanticState::reject(DeliveryId first, DeliveryId last) |
| { |
| AckRange range = findRange(first, last); |
| for_each(range.start, range.end, mem_fun_ref(&DeliveryRecord::reject)); |
| //may need to remove the delivery records as well |
| for (DeliveryRecords::iterator i = range.start; i != unacked.end() && i->getId() <= last; ) { |
| if (i->isRedundant()) i = unacked.erase(i); |
| else i++; |
| } |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| bool SemanticStateConsumerImpl::doOutput() |
| { |
| try { |
| return haveCredit() && doDispatch(); |
| } catch (const SessionException& e) { |
| throw SessionOutputException(e, parent->session.getChannel()); |
| } |
| } |
| |
| void SemanticStateConsumerImpl::enableNotify() |
| { |
| Mutex::ScopedLock l(lock); |
| notifyEnabled = true; |
| } |
| |
| void SemanticStateConsumerImpl::disableNotify() |
| { |
| Mutex::ScopedLock l(lock); |
| notifyEnabled = false; |
| } |
| |
| bool SemanticStateConsumerImpl::isNotifyEnabled() const { |
| Mutex::ScopedLock l(lock); |
| return notifyEnabled; |
| } |
| |
| void SemanticStateConsumerImpl::notify() |
| { |
| Mutex::ScopedLock l(lock); |
| if (notifyEnabled) { |
| parent->session.getConnection().addOutputTask(this); |
| parent->session.getConnection().activateOutput(); |
| } |
| } |
| |
| |
| // Test that a DeliveryRecord's ID is in a sequence set and some other |
| // predicate on DeliveryRecord holds. |
| template <class Predicate> struct IsInSequenceSetAnd { |
| IsInSequenceSet isInSet; |
| Predicate predicate; |
| IsInSequenceSetAnd(const SequenceSet& s, Predicate p) : isInSet(s), predicate(p) {} |
| bool operator()(DeliveryRecord& dr) { |
| return isInSet(dr.getId()) && predicate(dr); |
| } |
| }; |
| |
| template<class Predicate> IsInSequenceSetAnd<Predicate> |
| isInSequenceSetAnd(const SequenceSet& s, Predicate p) { |
| return IsInSequenceSetAnd<Predicate>(s,p); |
| } |
| |
| void SemanticState::accepted(const SequenceSet& commands) { |
| if (txBuffer.get()) { |
| //in transactional mode, don't dequeue or remove, just |
| //maintain set of acknowledged messages: |
| accumulatedAck.add(commands); |
| |
| if (dtxBuffer.get()) { |
| //if enlisted in a dtx, copy the relevant slice from |
| //unacked and record it against that transaction |
| TxOp::shared_ptr txAck(new DtxAck(accumulatedAck, unacked)); |
| accumulatedAck.clear(); |
| dtxBuffer->enlist(txAck); |
| //mark the relevant messages as 'ended' in unacked |
| //if the messages are already completed, they can be |
| //removed from the record |
| DeliveryRecords::iterator removed = |
| remove_if(unacked.begin(), unacked.end(), |
| isInSequenceSetAnd(commands, |
| bind(&DeliveryRecord::setEnded, _1))); |
| unacked.erase(removed, unacked.end()); |
| } |
| } else { |
| DeliveryRecords::iterator removed = |
| remove_if(unacked.begin(), unacked.end(), |
| isInSequenceSetAnd(commands, |
| bind(&DeliveryRecord::accept, _1, |
| (TransactionContext*) 0))); |
| unacked.erase(removed, unacked.end()); |
| } |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| void SemanticState::completed(const SequenceSet& commands) { |
| DeliveryRecords::iterator removed = |
| remove_if(unacked.begin(), unacked.end(), |
| isInSequenceSetAnd(commands, |
| bind(&SemanticState::complete, this, _1))); |
| unacked.erase(removed, unacked.end()); |
| requestDispatch(); |
| getSession().setUnackedCount(unacked.size()); |
| } |
| |
| void SemanticState::attached() |
| { |
| for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) { |
| i->second->enableNotify(); |
| session.getConnection().addOutputTask(i->second.get()); |
| } |
| session.getConnection().activateOutput(); |
| } |
| |
| void SemanticState::detached() |
| { |
| for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) { |
| i->second->disableNotify(); |
| session.getConnection().removeOutputTask(i->second.get()); |
| } |
| } |
| |
| void SemanticState::addBinding(const string& queueName, const string& exchangeName, |
| const string& routingKey, const framing::FieldTable& arguments) |
| { |
| QPID_LOG (debug, "SemanticState::addBinding [" |
| << "queue=" << queueName << ", " |
| << "exchange=" << exchangeName << ", " |
| << "key=" << routingKey << ", " |
| << "args=" << arguments << "]"); |
| std::string fedOp = arguments.getAsString(qpidFedOp); |
| if ((arguments.isSet(qpidFedOp)) && (fedOp.empty())) { |
| fedOp = fedOpBind; |
| } |
| std::string fedOrigin = arguments.getAsString(qpidFedOrigin); |
| if ((arguments.getAsString(X_SCOPE) == SESSION) || (fedOp == fedOpBind)) { |
| bindings.insert(boost::make_tuple(queueName, exchangeName, routingKey, fedOrigin)); |
| } |
| else if (fedOp == fedOpUnbind) { |
| bindings.erase(boost::make_tuple(queueName, exchangeName, routingKey, fedOrigin)); |
| } |
| } |
| |
| void SemanticState::removeBinding(const string& queueName, const string& exchangeName, |
| const string& routingKey) |
| { |
| QPID_LOG (debug, "SemanticState::removeBinding [" |
| << "queue=" << queueName << ", " |
| << "exchange=" << exchangeName << ", " |
| << "key=" << routingKey) |
| bindings.erase(boost::make_tuple(queueName, exchangeName, routingKey, "")); |
| } |
| |
| void SemanticState::unbindSessionBindings() |
| { |
| //unbind session-scoped bindings |
| for (Bindings::iterator i = bindings.begin(); i != bindings.end(); i++) { |
| QPID_LOG (debug, "SemanticState::unbindSessionBindings [" |
| << "queue=" << i->get<0>() << ", " |
| << "exchange=" << i->get<1>()<< ", " |
| << "key=" << i->get<2>() << ", " |
| << "fedOrigin=" << i->get<3>() << "]"); |
| try { |
| std::string fedOrigin = i->get<3>(); |
| if (!fedOrigin.empty()) { |
| framing::FieldTable fedArguments; |
| fedArguments.setString(qpidFedOp, fedOpUnbind); |
| fedArguments.setString(qpidFedOrigin, fedOrigin); |
| session.getBroker().bind(i->get<0>(), i->get<1>(), i->get<2>(), fedArguments, |
| &session, userID, connectionId); |
| } else { |
| session.getBroker().unbind(i->get<0>(), i->get<1>(), i->get<2>(), |
| &session, userID, connectionId); |
| } |
| } |
| catch (...) { |
| } |
| } |
| bindings.clear(); |
| } |
| |
| }} // namespace qpid::broker |
