diff options
Diffstat (limited to 'qpid/cpp/src/qpid/client/SubscriptionImpl.cpp')
| -rw-r--r-- | qpid/cpp/src/qpid/client/SubscriptionImpl.cpp | 169 |
1 files changed, 169 insertions, 0 deletions
diff --git a/qpid/cpp/src/qpid/client/SubscriptionImpl.cpp b/qpid/cpp/src/qpid/client/SubscriptionImpl.cpp new file mode 100644 index 0000000000..a8a0b47d94 --- /dev/null +++ b/qpid/cpp/src/qpid/client/SubscriptionImpl.cpp @@ -0,0 +1,169 @@ +/* + * + * 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/client/AsyncSession.h" +#include "qpid/client/SubscriptionImpl.h" +#include "qpid/client/SessionImpl.h" +#include "qpid/client/SubscriptionManagerImpl.h" +#include "qpid/client/MessageImpl.h" +#include "qpid/client/CompletionImpl.h" +#include "qpid/client/SubscriptionManager.h" +#include "qpid/client/SubscriptionSettings.h" +#include "qpid/client/SessionBase_0_10Access.h" +#include "qpid/client/PrivateImplRef.h" + +namespace qpid { +namespace client { + +using sys::Mutex; +using framing::MessageAcquireResult; + +SubscriptionImpl::SubscriptionImpl(SubscriptionManager m, const std::string& q, const SubscriptionSettings& s, const std::string& n, MessageListener* l) + : manager(*PrivateImplRef<SubscriptionManager>::get(m)), name(n), queue(q), settings(s), listener(l) +{} + +void SubscriptionImpl::subscribe() +{ + async(manager.getSession()).messageSubscribe( + arg::queue=queue, + arg::destination=name, + arg::acceptMode=settings.acceptMode, + arg::acquireMode=settings.acquireMode, + arg::exclusive=settings.exclusive); + setFlowControl(settings.flowControl); +} + +std::string SubscriptionImpl::getName() const { return name; } + +std::string SubscriptionImpl::getQueue() const { return queue; } + +const SubscriptionSettings& SubscriptionImpl::getSettings() const { + Mutex::ScopedLock l(lock); + return settings; +} + +void SubscriptionImpl::setFlowControl(const FlowControl& f) { + Mutex::ScopedLock l(lock); + AsyncSession s=manager.getSession(); + if (&settings.flowControl != &f) settings.flowControl = f; + s.messageSetFlowMode(name, f.window); + s.messageFlow(name, CREDIT_UNIT_MESSAGE, f.messages); + s.messageFlow(name, CREDIT_UNIT_BYTE, f.bytes); + s.sync(); +} + +void SubscriptionImpl::grantCredit(framing::message::CreditUnit unit, uint32_t value) { + async(manager.getSession()).messageFlow(name, unit, value); +} + +void SubscriptionImpl::setAutoAck(size_t n) { + Mutex::ScopedLock l(lock); + settings.autoAck = n; +} + +SequenceSet SubscriptionImpl::getUnacquired() const { Mutex::ScopedLock l(lock); return unacquired; } +SequenceSet SubscriptionImpl::getUnaccepted() const { Mutex::ScopedLock l(lock); return unaccepted; } + +void SubscriptionImpl::acquire(const SequenceSet& messageIds) { + Mutex::ScopedLock l(lock); + MessageAcquireResult result = manager.getSession().messageAcquire(messageIds); + unacquired.remove(result.getTransfers()); + if (settings.acceptMode == ACCEPT_MODE_EXPLICIT) + unaccepted.add(result.getTransfers()); +} + +void SubscriptionImpl::accept(const SequenceSet& messageIds) { + Mutex::ScopedLock l(lock); + manager.getSession().messageAccept(messageIds); + unaccepted.remove(messageIds); + switch (settings.completionMode) { + case COMPLETE_ON_ACCEPT: + manager.getSession().markCompleted(messageIds, true); + break; + case COMPLETE_ON_DELIVERY: + manager.getSession().sendCompletion(); + break; + default://do nothing + break; + } +} + +void SubscriptionImpl::release(const SequenceSet& messageIds) { + Mutex::ScopedLock l(lock); + manager.getSession().messageRelease(messageIds); + if (settings.acceptMode == ACCEPT_MODE_EXPLICIT) + unaccepted.remove(messageIds); +} + +Session SubscriptionImpl::getSession() const { return manager.getSession(); } + +SubscriptionManager SubscriptionImpl::getSubscriptionManager() { return SubscriptionManager(&manager); } + +void SubscriptionImpl::cancel() { manager.cancel(name); } + +void SubscriptionImpl::received(Message& m) { + Mutex::ScopedLock l(lock); + MessageImpl& mi = *MessageImpl::get(m); + if (mi.getMethod().getAcquireMode() == ACQUIRE_MODE_NOT_ACQUIRED) + unacquired.add(m.getId()); + else if (mi.getMethod().getAcceptMode() == ACCEPT_MODE_EXPLICIT) + unaccepted.add(m.getId()); + + if (listener) { + Mutex::ScopedUnlock u(lock); + listener->received(m); + } + + if (settings.completionMode == COMPLETE_ON_DELIVERY) { + manager.getSession().markCompleted(m.getId(), false, false); + } + if (settings.autoAck) { + if (unaccepted.size() >= settings.autoAck) { + async(manager.getSession()).messageAccept(unaccepted); + switch (settings.completionMode) { + case COMPLETE_ON_ACCEPT: + manager.getSession().markCompleted(unaccepted, true); + break; + case COMPLETE_ON_DELIVERY: + manager.getSession().sendCompletion(); + break; + default://do nothing + break; + } + unaccepted.clear(); + } + } +} + +Demux::QueuePtr SubscriptionImpl::divert() +{ + Session session(manager.getSession()); + Demux& demux = SessionBase_0_10Access(session).get()->getDemux(); + demuxRule = std::auto_ptr<ScopedDivert>(new ScopedDivert(name, demux)); + return demuxRule->getQueue(); +} + +void SubscriptionImpl::cancelDiversion() { + demuxRule.reset(); +} + +}} // namespace qpid::client + |