| /* |
| * |
| * 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/Bridge.h" |
| |
| #include "qpid/broker/Broker.h" |
| #include "qpid/broker/FedOps.h" |
| #include "qpid/broker/ConnectionState.h" |
| #include "qpid/broker/Connection.h" |
| #include "qpid/broker/Link.h" |
| #include "qpid/broker/LinkRegistry.h" |
| #include "qpid/ha/BrokerReplicator.h" |
| #include "qpid/broker/SessionState.h" |
| |
| #include "qpid/management/ManagementAgent.h" |
| #include "qpid/types/Variant.h" |
| #include "qpid/amqp_0_10/Codecs.h" |
| #include "qpid/framing/Uuid.h" |
| #include "qpid/framing/MessageProperties.h" |
| #include "qpid/framing/MessageTransferBody.h" |
| #include "qpid/log/Statement.h" |
| #include <iostream> |
| |
| using qpid::framing::FieldTable; |
| using qpid::framing::Uuid; |
| using qpid::framing::Buffer; |
| using qpid::framing::AMQFrame; |
| using qpid::framing::AMQContentBody; |
| using qpid::framing::AMQHeaderBody; |
| using qpid::framing::MessageProperties; |
| using qpid::framing::MessageTransferBody; |
| using qpid::types::Variant; |
| using qpid::management::ManagementAgent; |
| using std::string; |
| namespace _qmf = qmf::org::apache::qpid::broker; |
| |
| namespace { |
| const std::string QPID_REPLICATE("qpid.replicate"); |
| const std::string NONE("none"); |
| } |
| |
| namespace qpid { |
| namespace broker { |
| |
| void Bridge::PushHandler::handle(framing::AMQFrame& frame) |
| { |
| conn->received(frame); |
| } |
| |
| Bridge::Bridge(const std::string& _name, Link* _link, framing::ChannelId _id, |
| CancellationListener l, const _qmf::ArgsLinkBridge& _args, |
| InitializeCallback init, const std::string& _queueName, const string& ae) : |
| link(_link), channel(_id), args(_args), |
| listener(l), name(_name), |
| queueName(_queueName.empty() ? "qpid.bridge_queue_" + name + "_" + link->getBroker()->getFederationTag() |
| : _queueName), |
| altEx(ae), persistenceId(0), |
| connState(0), conn(0), initialize(init), detached(false), |
| useExistingQueue(!_queueName.empty()), |
| sessionName("qpid.bridge_session_" + name + "_" + link->getBroker()->getFederationTag()) |
| { |
| ManagementAgent* agent = link->getBroker()->getManagementAgent(); |
| if (agent != 0) { |
| mgmtObject = _qmf::Bridge::shared_ptr(new _qmf::Bridge |
| (agent, this, link, name, args.i_durable, args.i_src, args.i_dest, |
| args.i_key, args.i_srcIsQueue, args.i_srcIsLocal, |
| args.i_tag, args.i_excludes, args.i_dynamic, args.i_sync)); |
| mgmtObject->set_channelId(channel); |
| agent->addObject(mgmtObject); |
| } |
| QPID_LOG(debug, "Bridge " << name << " created from " << args.i_src << " to " << args.i_dest); |
| } |
| |
| Bridge::~Bridge() |
| { |
| mgmtObject->resourceDestroy(); |
| } |
| |
| void Bridge::create(Connection& c, AsyncStore* const store) |
| { |
| detached = false; // Reset detached in case we are recovering. |
| connState = &c; |
| conn = &c; |
| FieldTable options; |
| if (args.i_sync) options.setInt("qpid.sync_frequency", args.i_sync); |
| SessionHandler& sessionHandler = c.getChannel(channel); |
| sessionHandler.setErrorListener(shared_from_this()); |
| if (args.i_srcIsLocal) { |
| if (args.i_dynamic) |
| throw Exception("Dynamic routing not supported for push routes"); |
| // Point the bridging commands at the local connection handler |
| pushHandler.reset(new PushHandler(&c)); |
| channelHandler.reset(new framing::ChannelHandler(channel, pushHandler.get())); |
| |
| session.reset(new framing::AMQP_ServerProxy::Session(*channelHandler)); |
| peer.reset(new framing::AMQP_ServerProxy(*channelHandler)); |
| |
| session->attach(sessionName, false); |
| session->commandPoint(0,0); |
| } else { |
| sessionHandler.attachAs(sessionName); |
| // Point the bridging commands at the remote peer broker |
| peer.reset(new framing::AMQP_ServerProxy(sessionHandler.out)); |
| } |
| |
| if (args.i_srcIsLocal) sessionHandler.getSession()->disableReceiverTracking(); |
| if (initialize) initialize(*this, sessionHandler); |
| else if (args.i_srcIsQueue) { |
| peer->getMessage().subscribe(args.i_src, args.i_dest, args.i_sync ? 0 : 1, 0, false, "", 0, options); |
| peer->getMessage().flow(args.i_dest, 0, args.i_sync ? 2 * args.i_sync : 0xFFFFFFFF); |
| peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF); |
| QPID_LOG(debug, "Activated bridge " << name << " for route from queue " << args.i_src << " to " << args.i_dest); |
| } else { |
| if (!useExistingQueue) { |
| FieldTable queueSettings; |
| |
| if (args.i_tag.size()) { |
| queueSettings.setString("qpid.trace.id", args.i_tag); |
| } else { |
| const string& peerTag = c.getFederationPeerTag(); |
| if (peerTag.size()) |
| queueSettings.setString("qpid.trace.id", peerTag); |
| } |
| |
| if (args.i_excludes.size()) { |
| queueSettings.setString("qpid.trace.exclude", args.i_excludes); |
| } else { |
| const string& localTag = link->getBroker()->getFederationTag(); |
| if (localTag.size()) |
| queueSettings.setString("qpid.trace.exclude", localTag); |
| } |
| |
| bool durable = false;//should this be an arg, or would we use srcIsQueue for durable queues? |
| bool exclusive = true; // only exclusive if the queue is owned by the bridge |
| bool autoDelete = exclusive && !durable;//auto delete transient queues? |
| peer->getQueue().declare(queueName, altEx, false, durable, exclusive, autoDelete, queueSettings); |
| } |
| if (!args.i_dynamic) |
| peer->getExchange().bind(queueName, args.i_src, args.i_key, FieldTable()); |
| peer->getMessage().subscribe(queueName, args.i_dest, (useExistingQueue && args.i_sync) ? 0 : 1, 0, false, "", 0, options); |
| peer->getMessage().flow(args.i_dest, 0, (useExistingQueue && args.i_sync) ? 2 * args.i_sync : 0xFFFFFFFF); |
| peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF); |
| |
| if (args.i_dynamic) { |
| Exchange::shared_ptr exchange = link->getBroker()->getExchanges().get(args.i_src); |
| if (exchange.get() == 0) |
| throw Exception("Exchange not found for dynamic route"); |
| exchange->registerDynamicBridge(this, store); |
| QPID_LOG(debug, "Activated bridge " << name << " for dynamic route for exchange " << args.i_src); |
| } else { |
| QPID_LOG(debug, "Activated bridge " << name << " for static route from exchange " << args.i_src << " to " << args.i_dest); |
| } |
| } |
| if (args.i_srcIsLocal) sessionHandler.getSession()->enableReceiverTracking(); |
| } |
| |
| void Bridge::cancel(Connection&) |
| { |
| if (resetProxy()) { |
| peer->getMessage().cancel(args.i_dest); |
| peer->getSession().detach(sessionName); |
| } |
| QPID_LOG(debug, "Cancelled bridge " << name); |
| } |
| |
| /** Notify the bridge that the connection has closed */ |
| void Bridge::closed() |
| { |
| if (args.i_dynamic) { |
| Exchange::shared_ptr exchange = link->getBroker()->getExchanges().find(args.i_src); |
| if (exchange.get()) exchange->removeDynamicBridge(this); |
| } |
| QPID_LOG(debug, "Closed bridge " << name); |
| } |
| |
| /** Shut down the bridge */ |
| void Bridge::close() |
| { |
| listener(this); // ask the LinkRegistry to destroy us |
| } |
| |
| void Bridge::setPersistenceId(uint64_t pId) const |
| { |
| persistenceId = pId; |
| } |
| |
| |
| const std::string Bridge::ENCODED_IDENTIFIER("bridge.v2"); |
| const std::string Bridge::ENCODED_IDENTIFIER_V1("bridge"); |
| |
| bool Bridge::isEncodedBridge(const std::string& key) |
| { |
| return key == ENCODED_IDENTIFIER || key == ENCODED_IDENTIFIER_V1; |
| } |
| |
| |
| Bridge::shared_ptr Bridge::decode(LinkRegistry& links, Buffer& buffer) |
| { |
| string kind; |
| buffer.getShortString(kind); |
| |
| string host; |
| uint16_t port; |
| string src; |
| string dest; |
| string key; |
| string id; |
| string excludes; |
| string name; |
| |
| Link::shared_ptr link; |
| if (kind == ENCODED_IDENTIFIER_V1) { |
| /** previous versions identified the bridge by host:port, not by name, and |
| * transport wasn't provided. Try to find a link using those paramters. |
| */ |
| buffer.getShortString(host); |
| port = buffer.getShort(); |
| |
| link = links.getLink(host, port); |
| if (!link) { |
| QPID_LOG(error, "Bridge::decode() failed: cannot find Link for host=" << host << ", port=" << port); |
| return Bridge::shared_ptr(); |
| } |
| } else { |
| string linkName; |
| |
| buffer.getShortString(name); |
| buffer.getShortString(linkName); |
| link = links.getLink(linkName); |
| if (!link) { |
| QPID_LOG(error, "Bridge::decode() failed: cannot find Link named='" << linkName << "'"); |
| return Bridge::shared_ptr(); |
| } |
| } |
| |
| bool durable(buffer.getOctet()); |
| buffer.getShortString(src); |
| buffer.getShortString(dest); |
| buffer.getShortString(key); |
| bool is_queue(buffer.getOctet()); |
| bool is_local(buffer.getOctet()); |
| buffer.getShortString(id); |
| buffer.getShortString(excludes); |
| bool dynamic(buffer.getOctet()); |
| uint16_t sync = buffer.getShort(); |
| |
| if (kind == ENCODED_IDENTIFIER_V1) { |
| /** previous versions did not provide a name for the bridge, so create one |
| */ |
| name = createName(link->getName(), src, dest, key); |
| } |
| |
| return links.declare(name, *link, durable, src, dest, key, is_queue, |
| is_local, id, excludes, dynamic, sync).first; |
| } |
| |
| void Bridge::encode(Buffer& buffer) const |
| { |
| buffer.putShortString(ENCODED_IDENTIFIER); |
| buffer.putShortString(name); |
| buffer.putShortString(link->getName()); |
| buffer.putOctet(args.i_durable ? 1 : 0); |
| buffer.putShortString(args.i_src); |
| buffer.putShortString(args.i_dest); |
| buffer.putShortString(args.i_key); |
| buffer.putOctet(args.i_srcIsQueue ? 1 : 0); |
| buffer.putOctet(args.i_srcIsLocal ? 1 : 0); |
| buffer.putShortString(args.i_tag); |
| buffer.putShortString(args.i_excludes); |
| buffer.putOctet(args.i_dynamic ? 1 : 0); |
| buffer.putShort(args.i_sync); |
| } |
| |
| uint32_t Bridge::encodedSize() const |
| { |
| return ENCODED_IDENTIFIER.size() + 1 // +1 byte length |
| + name.size() + 1 |
| + link->getName().size() + 1 |
| + 1 // durable |
| + args.i_src.size() + 1 |
| + args.i_dest.size() + 1 |
| + args.i_key.size() + 1 |
| + 1 // srcIsQueue |
| + 1 // srcIsLocal |
| + args.i_tag.size() + 1 |
| + args.i_excludes.size() + 1 |
| + 1 // dynamic |
| + 2; // sync |
| } |
| |
| management::ManagementObject::shared_ptr Bridge::GetManagementObject(void) const |
| { |
| return mgmtObject; |
| } |
| |
| management::Manageable::status_t Bridge::ManagementMethod(uint32_t methodId, |
| management::Args& /*args*/, |
| string&) |
| { |
| if (methodId == _qmf::Bridge::METHOD_CLOSE) { |
| //notify that we are closed |
| QPID_LOG(debug, "Bridge::close() method called on bridge '" << name << "'"); |
| close(); |
| return management::Manageable::STATUS_OK; |
| } else { |
| return management::Manageable::STATUS_UNKNOWN_METHOD; |
| } |
| } |
| |
| void Bridge::propagateBinding(const string& key, const string& tagList, |
| const string& op, const string& origin, |
| qpid::framing::FieldTable* extra_args) |
| { |
| const string& localTag = link->getBroker()->getFederationTag(); |
| const string& peerTag = connState->getFederationPeerTag(); |
| |
| if (tagList.find(peerTag) == tagList.npos) { |
| FieldTable bindArgs; |
| if (extra_args) { |
| for (qpid::framing::FieldTable::ValueMap::iterator i=extra_args->begin(); i != extra_args->end(); ++i) { |
| bindArgs.insert((*i)); |
| } |
| } |
| string newTagList(tagList + string(tagList.empty() ? "" : ",") + localTag); |
| |
| bindArgs.setString(QPID_REPLICATE, NONE); |
| bindArgs.setString(qpidFedOp, op); |
| bindArgs.setString(qpidFedTags, newTagList); |
| if (origin.empty()) |
| bindArgs.setString(qpidFedOrigin, localTag); |
| else |
| bindArgs.setString(qpidFedOrigin, origin); |
| |
| conn->requestIOProcessing(boost::bind(&Bridge::ioThreadPropagateBinding, this, |
| queueName, args.i_src, key, bindArgs)); |
| } |
| } |
| |
| void Bridge::sendReorigin() |
| { |
| FieldTable bindArgs; |
| |
| bindArgs.setString(qpidFedOp, fedOpReorigin); |
| bindArgs.setString(qpidFedTags, link->getBroker()->getFederationTag()); |
| |
| conn->requestIOProcessing(boost::bind(&Bridge::ioThreadPropagateBinding, this, |
| queueName, args.i_src, args.i_key, bindArgs)); |
| } |
| bool Bridge::resetProxy() |
| { |
| SessionHandler& sessionHandler = conn->getChannel(channel); |
| if (!sessionHandler.getSession()) peer.reset(); |
| else peer.reset(new framing::AMQP_ServerProxy(sessionHandler.out)); |
| return peer.get(); |
| } |
| |
| void Bridge::ioThreadPropagateBinding(const string& queue, const string& exchange, const string& key, FieldTable args) |
| { |
| if (resetProxy()) { |
| peer->getExchange().bind(queue, exchange, key, args); |
| } else { |
| // link's periodic maintenance visit will attempt to recover |
| } |
| } |
| |
| bool Bridge::containsLocalTag(const string& tagList) const |
| { |
| const string& localTag = link->getBroker()->getFederationTag(); |
| return (tagList.find(localTag) != tagList.npos); |
| } |
| |
| const string& Bridge::getLocalTag() const |
| { |
| return link->getBroker()->getFederationTag(); |
| } |
| |
| // SessionHandler::ErrorListener methods. |
| void Bridge::connectionException( |
| framing::connection::CloseCode code, const std::string& msg) |
| { |
| if (errorListener) errorListener->connectionException(code, msg); |
| } |
| |
| void Bridge::channelException( |
| framing::session::DetachCode code, const std::string& msg) |
| { |
| if (errorListener) errorListener->channelException(code, msg); |
| } |
| |
| void Bridge::executionException( |
| framing::execution::ErrorCode code, const std::string& msg) |
| { |
| if (errorListener) errorListener->executionException(code, msg); |
| } |
| |
| void Bridge::detach() { |
| detached = true; |
| if (errorListener) errorListener->detach(); |
| } |
| |
| std::string Bridge::createName(const std::string& linkName, |
| const std::string& src, |
| const std::string& dest, |
| const std::string& key) |
| { |
| std::stringstream keystream; |
| keystream << linkName << "!" << src << "!" << dest << "!" << key; |
| return keystream.str(); |
| } |
| |
| }} |