| /* |
| * |
| * 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 "Session.h" |
| #include "Outgoing.h" |
| #include "Message.h" |
| #include "ManagedConnection.h" |
| #include "qpid/broker/AsyncCompletion.h" |
| #include "qpid/broker/Broker.h" |
| #include "qpid/broker/DeliverableMessage.h" |
| #include "qpid/broker/Exchange.h" |
| #include "qpid/broker/DirectExchange.h" |
| #include "qpid/broker/TopicExchange.h" |
| #include "qpid/broker/FanOutExchange.h" |
| #include "qpid/broker/Message.h" |
| #include "qpid/broker/Queue.h" |
| #include "qpid/broker/TopicExchange.h" |
| #include "qpid/broker/amqp/Filter.h" |
| #include "qpid/broker/amqp/NodeProperties.h" |
| #include "qpid/framing/AMQFrame.h" |
| #include "qpid/framing/FieldTable.h" |
| #include "qpid/framing/MessageTransferBody.h" |
| #include "qpid/log/Statement.h" |
| #include <boost/intrusive_ptr.hpp> |
| #include <boost/format.hpp> |
| #include <map> |
| #include <sstream> |
| extern "C" { |
| #include <proton/engine.h> |
| } |
| |
| namespace qpid { |
| namespace broker { |
| namespace amqp { |
| |
| class Target |
| { |
| public: |
| Target(pn_link_t* l) : credit(100), window(0), link(l) {} |
| virtual ~Target() {} |
| bool flow(); |
| bool needFlow(); |
| virtual void handle(qpid::broker::Message& m) = 0;//TODO: revise this for proper message |
| protected: |
| const uint32_t credit; |
| uint32_t window; |
| pn_link_t* link; |
| }; |
| |
| class Queue : public Target |
| { |
| public: |
| Queue(boost::shared_ptr<qpid::broker::Queue> q, pn_link_t* l) : Target(l), queue(q) {} |
| void handle(qpid::broker::Message& m); |
| private: |
| boost::shared_ptr<qpid::broker::Queue> queue; |
| }; |
| |
| class Exchange : public Target |
| { |
| public: |
| Exchange(boost::shared_ptr<qpid::broker::Exchange> e, pn_link_t* l) : Target(l), exchange(e) {} |
| void handle(qpid::broker::Message& m); |
| private: |
| boost::shared_ptr<qpid::broker::Exchange> exchange; |
| }; |
| |
| Session::Session(pn_session_t* s, qpid::broker::Broker& b, ManagedConnection& c, qpid::sys::OutputControl& o) |
| : ManagedSession(b, c, (boost::format("%1%") % s).str()), session(s), broker(b), connection(c), out(o), deleted(false) {} |
| |
| |
| Session::ResolvedNode Session::resolve(const std::string name, pn_terminus_t* terminus) |
| { |
| ResolvedNode node; |
| node.exchange = broker.getExchanges().find(name); |
| node.queue = broker.getQueues().find(name); |
| if (!node.queue && !node.exchange && pn_terminus_is_dynamic(terminus)) { |
| //TODO: handle dynamic creation |
| //is it a queue or an exchange? |
| NodeProperties properties; |
| properties.read(pn_terminus_properties(terminus)); |
| if (properties.isQueue()) { |
| node.queue = broker.createQueue(name, properties.getQueueSettings(), this, properties.getAlternateExchange(), connection.getUserid(), connection.getId()).first; |
| } else { |
| qpid::framing::FieldTable args; |
| node.exchange = broker.createExchange(name, properties.getExchangeType(), properties.isDurable(), properties.getAlternateExchange(), |
| args, connection.getUserid(), connection.getId()).first; |
| } |
| } else if (node.queue && node.exchange) { |
| QPID_LOG_CAT(warning, protocol, "Ambiguous node name; " << name << " could be queue or exchange, assuming queue"); |
| node.exchange.reset(); |
| } |
| return node; |
| } |
| |
| void Session::attach(pn_link_t* link) |
| { |
| if (pn_link_is_sender(link)) { |
| pn_terminus_t* source = pn_link_remote_source(link); |
| //i.e a subscription |
| if (pn_terminus_get_type(source) == PN_UNSPECIFIED) { |
| throw qpid::Exception("No source specified!");/*invalid-field?*/ |
| } |
| std::string name = pn_terminus_get_address(source); |
| QPID_LOG(debug, "Received attach request for outgoing link from " << name); |
| pn_terminus_set_address(pn_link_source(link), name.c_str()); |
| |
| ResolvedNode node = resolve(name, source); |
| Filter filter; |
| filter.read(pn_terminus_filter(source)); |
| |
| if (node.queue) { |
| boost::shared_ptr<Outgoing> q(new Outgoing(broker, node.queue, link, *this, out, false)); |
| q->init(); |
| if (filter.hasSubjectFilter()) { |
| q->setSubjectFilter(filter.getSubjectFilter()); |
| } |
| senders[link] = q; |
| } else if (node.exchange) { |
| QueueSettings settings(false, true); |
| //TODO: populate settings from source details when available from engine |
| boost::shared_ptr<qpid::broker::Queue> queue |
| = broker.createQueue(name + qpid::types::Uuid(true).str(), settings, this, "", connection.getUserid(), connection.getId()).first; |
| if (filter.hasSubjectFilter()) { |
| filter.bind(node.exchange, queue); |
| filter.write(pn_terminus_filter(pn_link_source(link))); |
| } else if (node.exchange->getType() == FanOutExchange::typeName) { |
| node.exchange->bind(queue, std::string(), 0); |
| } else if (node.exchange->getType() == TopicExchange::typeName) { |
| node.exchange->bind(queue, "#", 0); |
| } else { |
| throw qpid::Exception("Exchange type requires a filter: " + node.exchange->getType());/*not-supported?*/ |
| } |
| boost::shared_ptr<Outgoing> q(new Outgoing(broker, queue, link, *this, out, true)); |
| senders[link] = q; |
| q->init(); |
| } else { |
| pn_terminus_set_type(pn_link_source(link), PN_UNSPECIFIED); |
| throw qpid::Exception("Node not found: " + name);/*not-found*/ |
| } |
| QPID_LOG(debug, "Outgoing link attached"); |
| } else { |
| pn_terminus_t* target = pn_link_remote_target(link); |
| if (pn_terminus_get_type(target) == PN_UNSPECIFIED) { |
| throw qpid::Exception("No target specified!");/*invalid field?*/ |
| } |
| std::string name = pn_terminus_get_address(target); |
| QPID_LOG(debug, "Received attach request for incoming link to " << name); |
| pn_terminus_set_address(pn_link_target(link), name.c_str()); |
| |
| ResolvedNode node = resolve(name, target); |
| |
| if (node.queue) { |
| boost::shared_ptr<Target> q(new Queue(node.queue, link)); |
| targets[link] = q; |
| } else if (node.exchange) { |
| boost::shared_ptr<Target> e(new Exchange(node.exchange, link)); |
| targets[link] = e; |
| } else { |
| pn_terminus_set_type(pn_link_target(link), PN_UNSPECIFIED); |
| throw qpid::Exception("Node not found: " + name);/*not-found*/ |
| } |
| QPID_LOG(debug, "Incoming link attached"); |
| } |
| } |
| |
| void Session::detach(pn_link_t* link) |
| { |
| if (pn_link_is_sender(link)) { |
| Senders::iterator i = senders.find(link); |
| if (i != senders.end()) { |
| i->second->detached(); |
| senders.erase(i); |
| QPID_LOG(debug, "Outgoing link detached"); |
| } |
| } else { |
| targets.erase(link); |
| QPID_LOG(debug, "Incoming link detached"); |
| } |
| } |
| namespace { |
| class Transfer : public qpid::broker::AsyncCompletion::Callback |
| { |
| public: |
| Transfer(pn_delivery_t* d, boost::shared_ptr<Session> s) : delivery(d), session(s) {} |
| void completed(bool sync) { session->accepted(delivery, sync); } |
| boost::intrusive_ptr<qpid::broker::AsyncCompletion::Callback> clone() |
| { |
| boost::intrusive_ptr<qpid::broker::AsyncCompletion::Callback> copy(new Transfer(delivery, session)); |
| return copy; |
| } |
| private: |
| pn_delivery_t* delivery; |
| boost::shared_ptr<Session> session; |
| }; |
| } |
| |
| void Session::accepted(pn_delivery_t* delivery, bool sync) |
| { |
| if (sync) { |
| //this is on IO thread |
| pn_delivery_update(delivery, PN_ACCEPTED); |
| pn_delivery_settle(delivery);//do we need to check settlement modes/orders? |
| incomingMessageAccepted(); |
| } else { |
| //this is not on IO thread, need to delay processing until on IO thread |
| qpid::sys::Mutex::ScopedLock l(lock); |
| if (!deleted) { |
| completed.push_back(delivery); |
| out.activateOutput(); |
| } |
| } |
| } |
| |
| void Session::incoming(pn_link_t* link, pn_delivery_t* delivery) |
| { |
| pn_delivery_tag_t tag = pn_delivery_tag(delivery); |
| QPID_LOG(debug, "received delivery: " << std::string(tag.bytes, tag.size)); |
| boost::intrusive_ptr<Message> received(new Message(pn_delivery_pending(delivery))); |
| /*ssize_t read = */pn_link_recv(link, received->getData(), received->getSize()); |
| received->scan(); |
| pn_link_advance(link); |
| |
| qpid::broker::Message message(received, received); |
| |
| incomingMessageReceived(); |
| Targets::iterator target = targets.find(link); |
| if (target == targets.end()) { |
| QPID_LOG(error, "Received message on unknown link"); |
| pn_delivery_update(delivery, PN_REJECTED); |
| pn_delivery_settle(delivery);//do we need to check settlement modes/orders? |
| incomingMessageRejected(); |
| } else { |
| target->second->handle(message); |
| received->begin(); |
| Transfer t(delivery, shared_from_this()); |
| received->end(t); |
| if (target->second->needFlow()) out.activateOutput(); |
| } |
| } |
| void Session::outgoing(pn_link_t* link, pn_delivery_t* delivery) |
| { |
| Senders::iterator sender = senders.find(link); |
| if (sender == senders.end()) { |
| QPID_LOG(error, "Delivery returned for unknown link"); |
| } else { |
| sender->second->handle(delivery); |
| } |
| } |
| |
| bool Session::dispatch() |
| { |
| bool output(false); |
| for (Senders::iterator s = senders.begin(); s != senders.end(); ++s) { |
| if (s->second->dispatch()) output = true; |
| } |
| if (completed.size()) { |
| output = true; |
| std::deque<pn_delivery_t*> copy; |
| { |
| qpid::sys::Mutex::ScopedLock l(lock); |
| completed.swap(copy); |
| } |
| for (std::deque<pn_delivery_t*>::iterator i = copy.begin(); i != copy.end(); ++i) { |
| accepted(*i, true); |
| } |
| } |
| for (Targets::iterator t = targets.begin(); t != targets.end(); ++t) { |
| if (t->second->flow()) output = true; |
| } |
| |
| return output; |
| } |
| |
| void Session::close() |
| { |
| for (Senders::iterator i = senders.begin(); i != senders.end(); ++i) { |
| i->second->detached(); |
| } |
| senders.clear(); |
| targets.clear();//at present no explicit cleanup required for targets |
| QPID_LOG(debug, "Session closed, all senders cancelled."); |
| qpid::sys::Mutex::ScopedLock l(lock); |
| deleted = true; |
| } |
| |
| void Queue::handle(qpid::broker::Message& message) |
| { |
| queue->deliver(message); |
| --window; |
| } |
| |
| void Exchange::handle(qpid::broker::Message& message) |
| { |
| DeliverableMessage deliverable(message, 0); |
| exchange->route(deliverable); |
| --window; |
| } |
| |
| bool Target::flow() |
| { |
| bool issue = window < credit; |
| if (issue) { |
| pn_link_flow(link, credit - window);//TODO: proper flow control |
| window = credit; |
| } |
| return issue; |
| } |
| |
| bool Target::needFlow() |
| { |
| return window <= (credit/2); |
| } |
| |
| }}} // namespace qpid::broker::amqp |
