| /* |
| * |
| * 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/amqp/Outgoing.h" |
| #include "qpid/broker/amqp/Header.h" |
| #include "qpid/broker/amqp/Translation.h" |
| #include "qpid/broker/Queue.h" |
| #include "qpid/broker/TopicKeyNode.h" |
| #include "qpid/sys/OutputControl.h" |
| #include "qpid/amqp/MessageEncoder.h" |
| #include "qpid/log/Statement.h" |
| |
| namespace qpid { |
| namespace broker { |
| namespace amqp { |
| |
| Outgoing::Outgoing(Broker& broker, boost::shared_ptr<Queue> q, pn_link_t* l, ManagedSession& session, qpid::sys::OutputControl& o, bool topic) |
| : Consumer(pn_link_name(l), /*FIXME*/CONSUMER), |
| ManagedOutgoingLink(broker, *q, session, pn_link_name(l), topic), |
| exclusive(topic), |
| queue(q), deliveries(5000), link(l), out(o), |
| current(0), outstanding(0), |
| buffer(1024)/*used only for header at present*/ |
| { |
| for (size_t i = 0 ; i < deliveries.capacity(); ++i) { |
| deliveries[i].init(i); |
| } |
| } |
| |
| void Outgoing::init() |
| { |
| queue->consume(shared_from_this(), exclusive);//may throw exception |
| } |
| |
| bool Outgoing::dispatch() |
| { |
| QPID_LOG(trace, "Dispatching to " << getName() << ": " << pn_link_credit(link)); |
| if (canDeliver()) { |
| if (queue->dispatch(shared_from_this())) { |
| return true; |
| } else { |
| pn_link_drained(link); |
| QPID_LOG(debug, "No message available on " << queue->getName()); |
| } |
| } else { |
| QPID_LOG(debug, "Can't deliver to " << getName() << " from " << queue->getName() << ": " << pn_link_credit(link)); |
| } |
| return false; |
| } |
| |
| void Outgoing::write(const char* data, size_t size) |
| { |
| pn_link_send(link, data, size); |
| } |
| |
| void Outgoing::handle(pn_delivery_t* delivery) |
| { |
| pn_delivery_tag_t tag = pn_delivery_tag(delivery); |
| size_t i = *reinterpret_cast<const size_t*>(tag.bytes); |
| Record& r = deliveries[i]; |
| if (pn_delivery_writable(delivery)) { |
| assert(r.msg); |
| assert(!r.delivery); |
| r.delivery = delivery; |
| //write header |
| qpid::amqp::MessageEncoder encoder(&buffer[0], buffer.size()); |
| encoder.writeHeader(Header(r.msg)); |
| write(&buffer[0], encoder.getPosition()); |
| Translation t(r.msg); |
| t.write(*this); |
| if (pn_link_advance(link)) { |
| --outstanding; |
| outgoingMessageSent(); |
| QPID_LOG(debug, "Sent message " << r.msg.getSequence() << " from " << queue->getName() << ", index=" << r.index); |
| } else { |
| QPID_LOG(error, "Failed to send message " << r.msg.getSequence() << " from " << queue->getName() << ", index=" << r.index); |
| } |
| } |
| if (pn_delivery_updated(delivery)) { |
| assert(r.delivery == delivery); |
| r.disposition = pn_delivery_remote_state(delivery); |
| if (r.disposition) { |
| switch (r.disposition) { |
| case PN_ACCEPTED: |
| //TODO: only if consuming |
| queue->dequeue(0, r.cursor); |
| outgoingMessageAccepted(); |
| break; |
| case PN_REJECTED: |
| queue->reject(r.cursor); |
| outgoingMessageRejected(); |
| break; |
| case PN_RELEASED: |
| queue->release(r.cursor, false);//TODO: for PN_RELEASED, delivery count should not be incremented |
| outgoingMessageRejected();//TODO: not quite true... |
| break; |
| case PN_MODIFIED: |
| queue->release(r.cursor, true);//TODO: proper handling of modified |
| outgoingMessageRejected();//TODO: not quite true... |
| break; |
| default: |
| QPID_LOG(warning, "Unhandled disposition: " << r.disposition); |
| } |
| //TODO: ony settle once any dequeue on store has completed |
| pn_delivery_settle(delivery); |
| r.reset(); |
| } |
| } |
| } |
| |
| bool Outgoing::canDeliver() |
| { |
| return deliveries[current].delivery == 0 && pn_link_credit(link) > outstanding; |
| } |
| |
| void Outgoing::detached() |
| { |
| QPID_LOG(debug, "Detaching outgoing link from " << queue->getName()); |
| queue->cancel(shared_from_this()); |
| //TODO: release in a clearer order? |
| for (size_t i = 0 ; i < deliveries.capacity(); ++i) { |
| if (deliveries[i].msg) queue->release(deliveries[i].cursor, true); |
| } |
| Queue::tryAutoDelete(*queue->getBroker(), queue, "", ""); |
| } |
| |
| //Consumer interface: |
| bool Outgoing::deliver(const QueueCursor& cursor, const qpid::broker::Message& msg) |
| { |
| Record& r = deliveries[current++]; |
| if (current >= deliveries.capacity()) current = 0; |
| r.cursor = cursor; |
| r.msg = msg; |
| pn_delivery(link, r.tag); |
| QPID_LOG(debug, "Requested delivery of " << r.msg.getSequence() << " from " << queue->getName() << ", index=" << r.index); |
| ++outstanding; |
| return true; |
| } |
| |
| void Outgoing::notify() |
| { |
| QPID_LOG(trace, "Notification received for " << queue->getName()); |
| out.activateOutput(); |
| } |
| |
| bool Outgoing::accept(const qpid::broker::Message&) |
| { |
| return true; |
| } |
| |
| void Outgoing::setSubjectFilter(const std::string& f) |
| { |
| subjectFilter = f; |
| } |
| |
| namespace { |
| |
| bool match(TokenIterator& filter, TokenIterator& target) |
| { |
| bool wild = false; |
| while (!filter.finished()) |
| { |
| if (filter.match1('*')) { |
| if (target.finished()) return false; |
| //else move to next word in filter target |
| filter.next(); |
| target.next(); |
| } else if (filter.match1('#')) { |
| // i.e. filter word is '#' which can match a variable number of words in the target |
| filter.next(); |
| if (filter.finished()) return true; |
| else if (target.finished()) return false; |
| wild = true; |
| } else { |
| //filter word needs to match target exactly |
| if (target.finished()) return false; |
| std::string word; |
| target.pop(word); |
| if (filter.match(word)) { |
| wild = false; |
| filter.next(); |
| } else if (!wild) { |
| return false; |
| } |
| } |
| } |
| return target.finished(); |
| } |
| bool match(const std::string& filter, const std::string& target) |
| { |
| TokenIterator lhs(filter); |
| TokenIterator rhs(target); |
| return match(lhs, rhs); |
| } |
| } |
| |
| bool Outgoing::filter(const qpid::broker::Message& m) |
| { |
| return subjectFilter.empty() || subjectFilter == m.getRoutingKey() || match(subjectFilter, m.getRoutingKey()); |
| } |
| |
| void Outgoing::cancel() {} |
| |
| void Outgoing::acknowledged(const qpid::broker::DeliveryRecord&) {} |
| |
| qpid::broker::OwnershipToken* Outgoing::getSession() |
| { |
| return 0; |
| } |
| |
| Outgoing::Record::Record() : delivery(0), disposition(0), index(0) {} |
| void Outgoing::Record::init(size_t i) |
| { |
| index = i; |
| tag.bytes = reinterpret_cast<const char*>(&index); |
| tag.size = sizeof(index); |
| } |
| void Outgoing::Record::reset() |
| { |
| cursor = QueueCursor(); |
| msg = qpid::broker::Message(); |
| delivery = 0; |
| disposition = 0; |
| } |
| |
| |
| }}} // namespace qpid::broker::amqp |