Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/rajith_jms_client@1128369 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 0 insertions, 323 deletions

diff --git a/cpp/src/qpid/broker/Bridge.cpp b/cpp/src/qpid/broker/Bridge.cpp
deleted file mode 100644
index 7fbbf4e2c4..0000000000
--- a/cpp/src/qpid/broker/Bridge.cpp
+++ /dev/null
@@ -1,323 +0,0 @@
-/*
- *
- * 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/FedOps.h"
-#include "qpid/broker/ConnectionState.h"
-#include "qpid/broker/Connection.h"
-#include "qpid/broker/Link.h"
-#include "qpid/broker/LinkRegistry.h"
-#include "qpid/broker/SessionState.h"
-
-#include "qpid/management/ManagementAgent.h"
-#include "qpid/framing/Uuid.h"
-#include "qpid/log/Statement.h"
-#include <iostream>
-
-using qpid::framing::FieldTable;
-using qpid::framing::Uuid;
-using qpid::framing::Buffer;
-using qpid::management::ManagementAgent;
-using std::string;
-namespace _qmf = qmf::org::apache::qpid::broker;
-
-namespace qpid {
-namespace broker {
-
-void Bridge::PushHandler::handle(framing::AMQFrame& frame)
-{
-    conn->received(frame);
-}
-
-Bridge::Bridge(Link* _link, framing::ChannelId _id, CancellationListener l,
-               const _qmf::ArgsLinkBridge& _args) : 
-    link(_link), id(_id), args(_args), mgmtObject(0),
-    listener(l), name(Uuid(true).str()), queueName("bridge_queue_"), persistenceId(0)
-{
-    std::stringstream title;
-    title << id << "_" << link->getBroker()->getFederationTag();
-    queueName += title.str();
-    ManagementAgent* agent = link->getBroker()->getManagementAgent();
-    if (agent != 0) {
-        mgmtObject = new _qmf::Bridge
-            (agent, this, link, id, 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);
-        agent->addObject(mgmtObject);
-    }
-    QPID_LOG(debug, "Bridge created from " << args.i_src << " to " << args.i_dest);
-}
-
-Bridge::~Bridge() 
-{
-    mgmtObject->resourceDestroy(); 
-}
-
-void Bridge::create(Connection& c)
-{
-    connState = &c;
-    conn = &c;
-    FieldTable options;
-    if (args.i_sync) options.setInt("qpid.sync_frequency", args.i_sync);
-    SessionHandler& sessionHandler = c.getChannel(id);
-    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(id, pushHandler.get()));
-
-        session.reset(new framing::AMQP_ServerProxy::Session(*channelHandler));
-        peer.reset(new framing::AMQP_ServerProxy(*channelHandler));
-        
-        session->attach(name, false);
-        session->commandPoint(0,0);
-    } else {
-        sessionHandler.attachAs(name);
-        // 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 (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, 0xFFFFFFFF);
-        peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF);
-        QPID_LOG(debug, "Activated route from queue " << args.i_src << " to " << args.i_dest);
-    } else {
-        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 autoDelete = !durable;//auto delete transient queues?
-        peer->getQueue().declare(queueName, "", false, durable, true, autoDelete, queueSettings);
-        if (!args.i_dynamic)
-            peer->getExchange().bind(queueName, args.i_src, args.i_key, FieldTable());
-        peer->getMessage().subscribe(queueName, args.i_dest, 1, 0, false, "", 0, FieldTable());
-        peer->getMessage().flow(args.i_dest, 0, 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);
-            QPID_LOG(debug, "Activated dynamic route for exchange " << args.i_src);
-        } else {
-            QPID_LOG(debug, "Activated 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(name);
-    }
-}
-
-void Bridge::closed()
-{
-    if (args.i_dynamic) {
-        Exchange::shared_ptr exchange = link->getBroker()->getExchanges().get(args.i_src);
-        if (exchange.get() != 0)
-            exchange->removeDynamicBridge(this);
-    }
-}
-
-void Bridge::destroy()
-{
-    listener(this);
-}
-
-void Bridge::setPersistenceId(uint64_t pId) const
-{
-    persistenceId = pId;
-}
-
-const string& Bridge::getName() const
-{
-    return name;
-}
-
-Bridge::shared_ptr Bridge::decode(LinkRegistry& links, Buffer& buffer)
-{
-    string   host;
-    uint16_t port;
-    string   src;
-    string   dest;
-    string   key;
-    string   id;
-    string   excludes;
-
-    buffer.getShortString(host);
-    port = buffer.getShort();
-    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();
-
-    return links.declare(host, port, durable, src, dest, key,
-                         is_queue, is_local, id, excludes, dynamic, sync).first;
-}
-
-void Bridge::encode(Buffer& buffer) const 
-{
-    buffer.putShortString(string("bridge"));
-    buffer.putShortString(link->getHost());
-    buffer.putShort(link->getPort());
-    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 link->getHost().size() + 1 // short-string (host)
-        + 7                // short-string ("bridge")
-        + 2                // port
-        + 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* Bridge::GetManagementObject (void) const
-{
-    return (management::ManagementObject*) 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
-        destroy();
-        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(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(id);
-    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 {
-        QPID_LOG(error, "Cannot propagate binding for dynamic bridge as session has been detached, deleting dynamic bridge");
-        destroy();
-    }
-}
-
-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();
-}
-
-}}