text: /*
   *
   * 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/Broker.h"
  
  #include "qpid/broker/AclModule.h"
  #include "qpid/broker/BrokerOptions.h"
  #include "qpid/broker/Connection.h"
  #include "qpid/broker/DirectExchange.h"
  #include "qpid/broker/FanOutExchange.h"
  #include "qpid/broker/HeadersExchange.h"
  #include "qpid/broker/MessageStoreModule.h"
  #include "qpid/broker/NameGenerator.h"
  #include "qpid/broker/NullMessageStore.h"
  #include "qpid/broker/RecoveryManagerImpl.h"
  #include "qpid/broker/SaslAuthenticator.h"
  #include "qpid/broker/TopicExchange.h"
  #include "qpid/broker/Link.h"
  #include "qpid/broker/PersistableObject.h"
  #include "qpid/broker/QueueFlowLimit.h"
  #include "qpid/broker/QueueSettings.h"
  #include "qpid/broker/TransactionObserver.h"
  #include "qpid/broker/MessageGroupManager.h"
  
  #include "qmf/org/apache/qpid/broker/Package.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerConnect.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerCreate.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerDelete.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerQuery.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerEcho.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerGetLogLevel.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerQueueMoveMessages.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerSetLogLevel.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerGetLogHiresTimestamp.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerSetLogHiresTimestamp.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerSetTimestampConfig.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerGetTimestampConfig.h"
  #include "qmf/org/apache/qpid/broker/ArgsBrokerQueueRedirect.h"
  #include "qmf/org/apache/qpid/broker/EventExchangeDeclare.h"
  #include "qmf/org/apache/qpid/broker/EventExchangeDelete.h"
  #include "qmf/org/apache/qpid/broker/EventBind.h"
  #include "qmf/org/apache/qpid/broker/EventUnbind.h"
  #include "qmf/org/apache/qpid/broker/EventQueueRedirect.h"
  #include "qmf/org/apache/qpid/broker/EventQueueRedirectCancelled.h"
  #include "qpid/amqp_0_10/Codecs.h"
  #include "qpid/management/ManagementDirectExchange.h"
  #include "qpid/management/ManagementTopicExchange.h"
  #include "qpid/log/Logger.h"
  #include "qpid/log/Options.h"
  #include "qpid/log/Statement.h"
  #include "qpid/framing/AMQFrame.h"
  #include "qpid/framing/FieldTable.h"
  #include "qpid/framing/ProtocolInitiation.h"
  #include "qpid/framing/reply_exceptions.h"
  #include "qpid/framing/Uuid.h"
  #include "qpid/sys/TransportFactory.h"
  #include "qpid/sys/Poller.h"
  #include "qpid/sys/Dispatcher.h"
  #include "qpid/sys/Thread.h"
  #include "qpid/sys/Time.h"
  #include "qpid/sys/Timer.h"
  #include "qpid/sys/ConnectionInputHandlerFactory.h"
  #include "qpid/sys/SystemInfo.h"
  #include "qpid/Address.h"
  #include "qpid/StringUtils.h"
  #include "qpid/Url.h"
  #include "qpid/Version.h"
  #include "config.h"
  
  #include <boost/bind.hpp>
  #include <boost/format.hpp>
  
  #include <iostream>
  #include <memory>
  #include <set>
  
  using qpid::sys::TransportAcceptor;
  using qpid::sys::TransportConnector;
  using qpid::sys::Poller;
  using qpid::sys::Dispatcher;
  using qpid::sys::Thread;
  using qpid::framing::FrameHandler;
  using qpid::framing::ChannelId;
  using qpid::management::ManagementAgent;
  using qpid::management::ManagementObject;
  using qpid::management::Manageable;
  using qpid::management::Args;
  using qpid::management::getCurrentPublisher;
  using qpid::types::Variant;
  using std::string;
  using std::make_pair;
  
  namespace _qmf = qmf::org::apache::qpid::broker;
  
  namespace qpid {
  namespace broker {
  
  const std::string empty;
  const std::string amq_direct("amq.direct");
  const std::string amq_topic("amq.topic");
  const std::string amq_fanout("amq.fanout");
  const std::string amq_match("amq.match");
  const std::string qpid_management("qpid.management");
  const std::string knownHostsNone("none");
  
  BrokerOptions::BrokerOptions(const std::string& name) :
      qpid::Options(name),
      noDataDir(0),
      port(DEFAULT_PORT),
      workerThreads(5),
      connectionBacklog(10),
      enableMgmt(1),
      mgmtPublish(1),
      mgmtPubInterval(10*sys::TIME_SEC),
      queueCleanInterval(60*sys::TIME_SEC*10),//10 minutes
      auth(SaslAuthenticator::available()),
      realm("QPID"),
      replayFlushLimit(0),
      replayHardLimit(0),
      queueLimit(100*1048576/*100M default limit*/),
      tcpNoDelay(true),
      requireEncrypted(false),
      knownHosts(knownHostsNone),
      qmf2Support(true),
      qmf1Support(false),
      queueFlowStopRatio(80),
      queueFlowResumeRatio(70),
      queueThresholdEventRatio(80),
      defaultMsgGroup("qpid.no-group"),
      timestampRcvMsgs(false),    // set the 0.10 timestamp delivery property
      linkMaintenanceInterval(2*sys::TIME_SEC),
      linkHeartbeatInterval(120*sys::TIME_SEC),
      dtxDefaultTimeout(60),      // 60s
      dtxMaxTimeout(3600),        // 3600s
      maxNegotiateTime(10000),    // 10s
      sessionMaxUnacked(5000) 
  {
      int c = sys::SystemInfo::concurrency();
      workerThreads=c+1;
      std::string home = getHome();
  
      if (home.length() == 0)
          dataDir += DEFAULT_DATA_DIR_LOCATION;
      else
          dataDir += home;
      dataDir += DEFAULT_DATA_DIR_NAME;
  
      addOptions()
          ("data-dir", optValue(dataDir,"DIR"), "Directory to contain persistent data generated by the broker")
          ("no-data-dir", optValue(noDataDir), "Don't use a data directory.  No persistent configuration will be loaded or stored")
          ("paging-dir", optValue(pagingDir,"DIR"), "Directory in which paging files will be created for paged queues")
          ("port,p", optValue(port,"PORT"), "Tells the broker to listen on PORT")
          ("interface", optValue(listenInterfaces, "<interface name>|<interface address>"), "Which network interfaces to use to listen for incoming connections")
          ("listen-disable", optValue(listenDisabled, "<transport name>"), "Transports to disable listening")
          ("protocols", optValue(protocols, "<protocol name+version>"), "Which protocol versions to allow")
          ("worker-threads", optValue(workerThreads, "N"), "Sets the broker thread pool size")
          ("connection-backlog", optValue(connectionBacklog, "N"), "Sets the connection backlog limit for the server socket")
          ("mgmt-enable,m", optValue(enableMgmt,"yes|no"), "Enable Management")
          ("mgmt-publish", optValue(mgmtPublish,"yes|no"), "Enable Publish of Management Data ('no' implies query-only)")
          ("mgmt-qmf2", optValue(qmf2Support,"yes|no"), "Enable broadcast of management information over QMF v2")
          ("mgmt-qmf1", optValue(qmf1Support,"yes|no"), "Enable broadcast of management information over QMF v1")
          ("mgmt-pub-interval", optValue(mgmtPubInterval, "SECONDS"), "Management Publish Interval")
          ("queue-purge-interval", optValue(queueCleanInterval, "SECONDS"),
           "Interval between attempts to purge any expired messages from queues")
          ("auth", optValue(auth, "yes|no"), "Enable authentication, if disabled all incoming connections will be trusted")
          ("realm", optValue(realm, "REALM"), "Use the given realm when performing authentication")
          ("sasl-service-name", optValue(saslServiceName, "NAME"), "The service name to specify for SASL")
          ("default-queue-limit", optValue(queueLimit, "BYTES"), "Default maximum size for queues (in bytes)")
          ("tcp-nodelay", optValue(tcpNoDelay), "Set TCP_NODELAY on TCP connections")
          ("require-encryption", optValue(requireEncrypted), "Only accept connections that are encrypted")
          ("known-hosts-url", optValue(knownHosts, "URL or 'none'"), "URL to send as 'known-hosts' to clients ('none' implies empty list)")
          ("sasl-config", optValue(saslConfigPath, "DIR"), "Allows SASL config path, if supported by platform, to be overridden.  For default location on Linux, see Cyrus SASL documentation.  There is no SASL config dir on Windows.")
          ("default-flow-stop-threshold", optValue(queueFlowStopRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is activated.")
          ("default-flow-resume-threshold", optValue(queueFlowResumeRatio, "PERCENT"), "Percent of queue's maximum capacity at which flow control is de-activated.")
          ("default-event-threshold-ratio", optValue(queueThresholdEventRatio, "%age of limit"), "The ratio of any specified queue limit at which an event will be raised")
          ("default-message-group", optValue(defaultMsgGroup, "GROUP-IDENTIFER"), "Group identifier to assign to messages delivered to a message group queue that do not contain an identifier.")
          ("enable-timestamp", optValue(timestampRcvMsgs, "yes|no"), "Add current time to each received message.")
          ("link-maintenance-interval", optValue(linkMaintenanceInterval, "SECONDS"),
           "Interval to check federation link health and re-connect if need be")
          ("link-heartbeat-interval", optValue(linkHeartbeatInterval, "SECONDS"),
           "Heartbeat interval for a federation link")
          ("dtx-default-timeout", optValue(dtxDefaultTimeout, "SECONDS"), "Default timeout for DTX transaction before aborting it")
          ("dtx-max-timeout", optValue(dtxMaxTimeout, "SECONDS"), "Maximum allowed timeout for DTX transaction. A value of zero disables maximum timeout limit checks and allows arbitrarily large timeout settings.")
          ("max-negotiate-time", optValue(maxNegotiateTime, "MILLISECONDS"), "Maximum time a connection can take to send the initial protocol negotiation")
          ("federation-tag", optValue(fedTag, "NAME"), "Override the federation tag")
          ("session-max-unacked", optValue(sessionMaxUnacked, "DELIVERIES"), "Maximum number of un-acknowledged outoing messages per sesssion")
          ;
  }
  
  namespace {
  // Arguments to declare a non-replicated exchange.
  framing::FieldTable noReplicateArgs() {
      framing::FieldTable args;
      args.setString("qpid.replicate", "none");
      return args;
  }
  }
  
  Broker::LogPrefix::LogPrefix() :
      std::string(Msg() << "Broker (pid=" << sys::SystemInfo::getProcessId() << ") ") {
      QPID_LOG(notice, *this << "start-up");
  }
  
  Broker::LogPrefix::~LogPrefix() { QPID_LOG(notice, *this << "shut-down"); }
  
  Broker::Broker(const BrokerOptions& conf) :
      poller(new Poller),
      timer(new qpid::sys::Timer),
      config(conf),
      managementAgent(conf.enableMgmt ? new ManagementAgent(conf.qmf1Support,
                                                            conf.qmf2Support)
                                      : 0),
      disabledListeningTransports(conf.listenDisabled.begin(), conf.listenDisabled.end()),
      store(new NullMessageStore),
      acl(0),
      dataDir(conf.noDataDir ? std::string() : conf.dataDir),
      pagingDir(!conf.pagingDir.empty() ? conf.pagingDir :
                dataDir.isEnabled() ? dataDir.getPath() + BrokerOptions::DEFAULT_PAGED_QUEUE_DIR :
                std::string() ),
      queues(this),
      exchanges(this),
      links(this),
      dtxManager(*timer.get(), conf.dtxDefaultTimeout),
      sessionManager(
          qpid::SessionState::Configuration(
              conf.replayFlushLimit*1024, // convert kb to bytes.
              conf.replayHardLimit*1024),
          *this),
      queueCleaner(queues, poller, timer.get()),
      recoveryInProgress(false),
      protocolRegistry(std::set<std::string>(conf.protocols.begin(), conf.protocols.end()), this),
      timestampRcvMsgs(conf.timestampRcvMsgs),
      getKnownBrokers(boost::bind(&Broker::getKnownBrokersImpl, this))
  {
      if (!dataDir.isEnabled()) {
          QPID_LOG (info, "No data directory - Disabling persistent configuration");
      }
      System* system = new System (dataDir.isEnabled() ? dataDir.getPath() : string(), this);
      systemObject = System::shared_ptr(system);
      try {
      if (conf.enableMgmt) {
          QPID_LOG(info, "Management enabled");
          managementAgent->configure(dataDir.isEnabled() ? dataDir.getPath() : string(), conf.mgmtPublish,
                                     conf.mgmtPubInterval/sys::TIME_SEC, this, conf.workerThreads + 3);
          managementAgent->setName("apache.org", "qpidd");
          _qmf::Package packageInitializer(managementAgent.get());
  
          mgmtObject = _qmf::Broker::shared_ptr(new _qmf::Broker(managementAgent.get(), this, system, "amqp-broker"));
          mgmtObject->set_systemRef(system->GetManagementObject()->getObjectId());
          mgmtObject->set_port(conf.port);
          mgmtObject->set_workerThreads(conf.workerThreads);
          mgmtObject->set_connBacklog(conf.connectionBacklog);
          mgmtObject->set_mgmtPubInterval(conf.mgmtPubInterval/sys::TIME_SEC);
          mgmtObject->set_mgmtPublish(conf.mgmtPublish);
          mgmtObject->set_version(qpid::version);
          if (dataDir.isEnabled())
              mgmtObject->set_dataDir(dataDir.getPath());
          else
              mgmtObject->clr_dataDir();
  
          managementAgent->addObject(mgmtObject, 0, true);
  
          // Since there is currently no support for virtual hosts, a placeholder object
          // representing the implied single virtual host is added here to keep the
          // management schema correct.
          Vhost* vhost = new Vhost(this, this);
          vhostObject = Vhost::shared_ptr(vhost);
          if (conf.fedTag.empty()) {
              framing::Uuid uuid(managementAgent->getUuid());
              federationTag = uuid.str();
          } else
              federationTag = conf.fedTag;
          vhostObject->setFederationTag(federationTag);
  
          queues.setParent(vhost);
          exchanges.setParent(vhost);
          links.setParent(vhost);
      } else {
          // Management is disabled so there is no broker management ID.
          // Create a unique uuid to use as the federation tag.
          if (conf.fedTag.empty()) {
              framing::Uuid uuid(true);
              federationTag = uuid.str();
          } else
              federationTag = conf.fedTag;
      }
  
      // Early-Initialize plugins
      Plugin::earlyInitAll(*this);
  
      QueueFlowLimit::setDefaults(conf.queueLimit, conf.queueFlowStopRatio, conf.queueFlowResumeRatio);
      MessageGroupManager::setDefaults(conf.defaultMsgGroup);
  
      // If no plugin store module registered itself, set up the null store.
      if (NullMessageStore::isNullStore(store.get()))
          setStore();
  
      framing::FieldTable args;
  
      // Default exchnge is not replicated.
      exchanges.declare(empty, DirectExchange::typeName, false, false, noReplicateArgs());
  
      RecoveredObjects objects;
      if (store.get() != 0) {
          RecoveryManagerImpl recoverer(
              queues, exchanges, links, dtxManager, protocolRegistry, objects);
          recoveryInProgress = true;
          store->recover(recoverer);
          recoveryInProgress = false;
      }
  
      //ensure standard exchanges exist (done after recovery from store)
      declareStandardExchange(amq_direct, DirectExchange::typeName);
      declareStandardExchange(amq_topic, TopicExchange::typeName);
      declareStandardExchange(amq_fanout, FanOutExchange::typeName);
      declareStandardExchange(amq_match, HeadersExchange::typeName);
  
      if(conf.enableMgmt) {
          exchanges.declare(qpid_management, ManagementTopicExchange::typeName, false, false, noReplicateArgs());
          Exchange::shared_ptr mExchange = exchanges.get(qpid_management);
          Exchange::shared_ptr dExchange = exchanges.get(amq_direct);
          managementAgent->setExchange(mExchange, dExchange);
          boost::dynamic_pointer_cast<ManagementTopicExchange>(mExchange)->setManagmentAgent(managementAgent.get(), 1);
  
          std::string qmfTopic("qmf.default.topic");
          std::string qmfDirect("qmf.default.direct");
  
          std::pair<Exchange::shared_ptr, bool> topicPair(
              exchanges.declare(qmfTopic, ManagementTopicExchange::typeName, false, false, noReplicateArgs()));
          std::pair<Exchange::shared_ptr, bool> directPair(
              exchanges.declare(qmfDirect, ManagementDirectExchange::typeName, false, false, noReplicateArgs()));
  
          boost::dynamic_pointer_cast<ManagementDirectExchange>(directPair.first)->setManagmentAgent(managementAgent.get(), 2);
          boost::dynamic_pointer_cast<ManagementTopicExchange>(topicPair.first)->setManagmentAgent(managementAgent.get(), 2);
  
          managementAgent->setExchangeV2(topicPair.first, directPair.first);
      }
      else
          QPID_LOG(info, "Management not enabled");
  
      // this feature affects performance, so let's be sure that gets logged!
      if (conf.timestampRcvMsgs) {
          QPID_LOG(notice, "Receive message timestamping is ENABLED.");
      }
  
      /**
       * SASL setup, can fail and terminate startup
       */
      if (conf.auth) {
          SaslAuthenticator::init(qpid::saslName, conf.saslConfigPath);
          QPID_LOG(info, "SASL enabled");
      } else {
          QPID_LOG(notice, "SASL disabled: No Authentication Performed");
      }
  
      // Initialize plugins
      Plugin::initializeAll(*this);
      //recover any objects via object factories
      objects.restore(*this);
  
      // Assign to queues their users who created them (can be done after ACL is loaded in Plugin::initializeAll above
      if ((getAcl()) && (store.get())) {
          queues.eachQueue(boost::bind(&qpid::broker::Queue::updateAclUserQueueCount, _1));
      }
  
      if(conf.enableMgmt) {
          if (getAcl()) {
              mgmtObject->set_maxConns(getAcl()->getMaxConnectTotal());
          }
      }
  
      if (conf.queueCleanInterval) {
          queueCleaner.start(conf.queueCleanInterval);
      }
  
      if (!conf.knownHosts.empty() && conf.knownHosts != knownHostsNone) {
          knownBrokers.push_back(Url(conf.knownHosts));
      }
  
      } catch (const std::exception& e) {
          QPID_LOG(critical, logPrefix << "start-up failed: " << e.what());
          finalize();
          throw;
      }
      QPID_LOG(info, logPrefix << "initialized");
  }
  
  void Broker::declareStandardExchange(const std::string& name, const std::string& type)
  {
      bool storeEnabled = store.get() != NULL;
      framing::FieldTable args;
      // Standard exchanges are not replicated.
      std::pair<Exchange::shared_ptr, bool> status =
          exchanges.declare(name, type, storeEnabled, false, noReplicateArgs());
      if (status.second && storeEnabled) {
          store->create(*status.first, framing::FieldTable ());
      }
  }
  
  bool Broker::isAuthenticating() const
  {
      return config.auth;
  }
  
  bool Broker::requireEncrypted() const
  {
      return config.requireEncrypted;
  }
  
  std::string Broker::getRealm() const
  {
      return config.realm;
  }
  
  std::string Broker::getSaslServiceName() const
  {
      return config.saslServiceName;
  }
  
  bool Broker::getTcpNoDelay() const
  {
      return config.tcpNoDelay;
  }
  
  uint32_t Broker::getMaxNegotiateTime() const
  {
      return config.maxNegotiateTime;
  }
  
  size_t Broker::getSessionMaxUnacked() const
  {
      return config.sessionMaxUnacked;
  }
  
  uint16_t Broker::getPortOption() const
  {
      return config.port;
  }
  
  const std::vector<std::string>& Broker::getListenInterfaces() const
  {
      return config.listenInterfaces;
  }
  
  int Broker::getConnectionBacklog() const
  {
      return config.connectionBacklog;
  }
  
  sys::Duration Broker::getLinkMaintenanceInterval() const
  {
      return config.linkMaintenanceInterval;
  }
  
  sys::Duration Broker::getLinkHeartbeatInterval() const
  {
      return config.linkHeartbeatInterval;
  }
  
  uint32_t Broker::getDtxMaxTimeout() const
  {
      return config.dtxMaxTimeout;
  }
  
  uint16_t Broker::getQueueThresholdEventRatio() const
  {
      return config.queueThresholdEventRatio;
  }
  
  uint Broker::getQueueLimit() const
  {
      return config.queueLimit;
  }
  
  boost::intrusive_ptr<Broker> Broker::create(int16_t port)
  {
      BrokerOptions config;
      config.port=port;
      return create(config);
  }
  
  boost::intrusive_ptr<Broker> Broker::create(const BrokerOptions& opts)
  {
      return boost::intrusive_ptr<Broker>(new Broker(opts));
  }
  
  void Broker::setStore (const boost::shared_ptr<MessageStore>& _store)
  {
      // Exit now if multiple store plugins are attempting to load
      if (!NullMessageStore::isNullStore(store.get())) {
          QPID_LOG(error, "Multiple store plugins are not supported");
          throw Exception(QPID_MSG("Failed to start broker: Multiple store plugins were loaded"));
      }
  
      store.reset(new MessageStoreModule (_store));
      setStore();
  }
  
  void Broker::setStore () {
      queues.setStore     (store.get());
      dtxManager.setStore (store.get());
      links.setStore      (store.get());
  }
  
  void Broker::run() {
      if (config.workerThreads > 0) {
          QPID_LOG(info, logPrefix << "running");
          Dispatcher d(poller);
          int numIOThreads = config.workerThreads;
          std::vector<Thread> t(numIOThreads-1);
  
          // Run n-1 io threads
          for (int i=0; i<numIOThreads-1; ++i)
              t[i] = Thread(d);
  
          // Run final thread
          d.run();
  
          // Now wait for n-1 io threads to exit
          for (int i=0; i<numIOThreads-1; ++i) {
              t[i].join();
          }
          QPID_LOG(info, logPrefix << "stopped");
      } else {
          throw Exception((boost::format("Invalid value for worker-threads: %1%") % config.workerThreads).str());
      }
  }
  
  void Broker::shutdown() {
      // NB: this function must be async-signal safe, it must not
      // call any function that is not async-signal safe.
      // Any unsafe shutdown actions should be done in the destructor.
      poller->shutdown();
  }
  
  Broker::~Broker() {
      QPID_LOG(info, logPrefix << "shutting down");
      if (mgmtObject != 0)
          mgmtObject->debugStats("destroying");
      shutdown();
      finalize();                 // Finalize any plugins.
      if (config.auth)
          SaslAuthenticator::fini();
      timer->stop();
      managementAgent.reset();
  }
  
  ManagementObject::shared_ptr Broker::GetManagementObject(void) const
  {
      return mgmtObject;
  }
  
  Manageable* Broker::GetVhostObject(void) const
  {
      return vhostObject.get();
  }
  
  Manageable::status_t Broker::ManagementMethod (uint32_t methodId,
                                                 Args&    args,
                                                 string&  text)
  {
      Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
  
      switch (methodId)
      {
      case _qmf::Broker::METHOD_ECHO :
          QPID_LOG (debug, "Broker::echo("
                    << dynamic_cast<_qmf::ArgsBrokerEcho&>(args).io_sequence
                    << ", "
                    << dynamic_cast<_qmf::ArgsBrokerEcho&>(args).io_body
                    << ")");
          status = Manageable::STATUS_OK;
          break;
      case _qmf::Broker::METHOD_CONNECT : {
          /** Management is creating a Link to a remote broker using the host and port of
           * the remote.  This (old) interface does not allow management to specify a name
           * for the link, nor does it allow multiple Links to the same remote.  Use the
           * "create()" broker method if these features are needed.
           * TBD: deprecate this interface.
           */
          QPID_LOG(info, "The Broker::connect() method will be removed in a future release of QPID."
                   " Please use the Broker::create() method with type='link' instead.");
          _qmf::ArgsBrokerConnect& hp=
              dynamic_cast<_qmf::ArgsBrokerConnect&>(args);
  
          string transport = hp.i_transport.empty() ? TCP_TRANSPORT : hp.i_transport;
          QPID_LOG (debug, "Broker::connect() " << hp.i_host << ":" << hp.i_port << "; transport=" << transport <<
                          "; durable=" << (hp.i_durable?"T":"F") << "; authMech=\"" << hp.i_authMechanism << "\"");
          if (!getTransportInfo(transport).connectorFactory) {
              QPID_LOG(error, "Transport '" << transport << "' not supported");
              text = "transport type not supported";
              return  Manageable::STATUS_NOT_IMPLEMENTED;
          }
  
          // Does a link to the remote already exist?  If so, re-use the existing link
          // - this behavior is backward compatible with previous releases.
          if (!links.getLink(hp.i_host, hp.i_port, transport)) {
              // new link, need to generate a unique name for it
              std::pair<Link::shared_ptr, bool> response =
                links.declare(Link::createName(transport, hp.i_host, hp.i_port),
                              hp.i_host, hp.i_port, transport,
                              hp.i_durable, hp.i_authMechanism, hp.i_username, hp.i_password);
              if (!response.first) {
                  text = "Unable to create Link";
                  status = Manageable::STATUS_PARAMETER_INVALID;
                  break;
              }
          }
          status = Manageable::STATUS_OK;
          break;
        }
      case _qmf::Broker::METHOD_QUEUEMOVEMESSAGES : {
          _qmf::ArgsBrokerQueueMoveMessages& moveArgs=
              dynamic_cast<_qmf::ArgsBrokerQueueMoveMessages&>(args);
          QPID_LOG (debug, "Broker::queueMoveMessages()");
          if (queueMoveMessages(moveArgs.i_srcQueue, moveArgs.i_destQueue, moveArgs.i_qty,
                                moveArgs.i_filter, getCurrentPublisher()) >=0)
              status = Manageable::STATUS_OK;
          else
              return Manageable::STATUS_PARAMETER_INVALID;
          break;
        }
      case _qmf::Broker::METHOD_SETLOGLEVEL :
          setLogLevel(dynamic_cast<_qmf::ArgsBrokerSetLogLevel&>(args).i_level);
          QPID_LOG (debug, "Broker::setLogLevel()");
          status = Manageable::STATUS_OK;
          break;
      case _qmf::Broker::METHOD_GETLOGLEVEL :
          dynamic_cast<_qmf::ArgsBrokerGetLogLevel&>(args).o_level = getLogLevel();
          QPID_LOG (debug, "Broker::getLogLevel()");
          status = Manageable::STATUS_OK;
          break;
      case _qmf::Broker::METHOD_CREATE :
        {
            _qmf::ArgsBrokerCreate& a = dynamic_cast<_qmf::ArgsBrokerCreate&>(args);
            createObject(a.i_type, a.i_name, a.i_properties, a.i_strict, getCurrentPublisher());
            status = Manageable::STATUS_OK;
            break;
        }
      case _qmf::Broker::METHOD_DELETE :
        {
            _qmf::ArgsBrokerDelete& a = dynamic_cast<_qmf::ArgsBrokerDelete&>(args);
            deleteObject(a.i_type, a.i_name, a.i_options, getCurrentPublisher());
            status = Manageable::STATUS_OK;
            break;
        }
      case _qmf::Broker::METHOD_QUERY :
        {
            _qmf::ArgsBrokerQuery& a = dynamic_cast<_qmf::ArgsBrokerQuery&>(args);
            status = queryObject(a.i_type, a.i_name, a.o_results, getCurrentPublisher());
            break;
        }
      case _qmf::Broker::METHOD_GETTIMESTAMPCONFIG:
          {
            _qmf::ArgsBrokerGetTimestampConfig& a = dynamic_cast<_qmf::ArgsBrokerGetTimestampConfig&>(args);
            status = getTimestampConfig(a.o_receive, getCurrentPublisher());
            break;
          }
      case _qmf::Broker::METHOD_SETTIMESTAMPCONFIG:
          {
            _qmf::ArgsBrokerSetTimestampConfig& a = dynamic_cast<_qmf::ArgsBrokerSetTimestampConfig&>(args);
            status = setTimestampConfig(a.i_receive, getCurrentPublisher());
            break;
          }
  
      case _qmf::Broker::METHOD_GETLOGHIRESTIMESTAMP:
      {
          dynamic_cast<_qmf::ArgsBrokerGetLogHiresTimestamp&>(args).o_logHires = getLogHiresTimestamp();
          QPID_LOG (debug, "Broker::getLogHiresTimestamp()");
          status = Manageable::STATUS_OK;
          break;
      }
      case _qmf::Broker::METHOD_SETLOGHIRESTIMESTAMP:
      {
          setLogHiresTimestamp(dynamic_cast<_qmf::ArgsBrokerSetLogHiresTimestamp&>(args).i_logHires);
          QPID_LOG (debug, "Broker::setLogHiresTimestamp()");
          status = Manageable::STATUS_OK;
          break;
      }
      case _qmf::Broker::METHOD_QUEUEREDIRECT:
      {
          string srcQueue(dynamic_cast<_qmf::ArgsBrokerQueueRedirect&>(args).i_sourceQueue);
          string tgtQueue(dynamic_cast<_qmf::ArgsBrokerQueueRedirect&>(args).i_targetQueue);
          QPID_LOG (debug, "Broker::queueRedirect source queue:" << srcQueue << " to target queue " << tgtQueue);
          status =  queueRedirect(srcQueue, tgtQueue, getCurrentPublisher());
          break;
      }
      case _qmf::Broker::METHOD_SHUTDOWN :
      {
          QPID_LOG (info, "Broker received shutdown command");
          shutdown();
      }
      default:
          QPID_LOG (debug, "Broker ManagementMethod not implemented: id=" << methodId << "]");
          status = Manageable::STATUS_NOT_IMPLEMENTED;
          break;
      }
  
      return status;
  }
  
  namespace
  {
  const std::string TYPE_QUEUE("queue");
  const std::string TYPE_EXCHANGE("exchange");
  const std::string TYPE_TOPIC("topic");
  const std::string TYPE_BINDING("binding");
  const std::string TYPE_LINK("link");
  const std::string TYPE_BRIDGE("bridge");
  const std::string DURABLE("durable");
  const std::string AUTO_DELETE("auto-delete");
  const std::string ALTERNATE_EXCHANGE("alternate-exchange");
  const std::string EXCHANGE_TYPE("exchange-type");
  const std::string QUEUE_NAME("queue");
  const std::string EXCHANGE_NAME("exchange");
  
  const std::string ATTRIBUTE_TIMESTAMP_0_10("timestamp-0.10");
  
  const std::string _TRUE("true");
  const std::string _FALSE("false");
  
  // parameters for creating a Link object, see mgmt schema
  const std::string HOST("host");
  const std::string PORT("port");
  const std::string TRANSPORT("transport");
  const std::string AUTH_MECHANISM("authMechanism");
  const std::string USERNAME("username");
  const std::string PASSWORD("password");
  
  // parameters for creating a Bridge object, see mgmt schema
  const std::string LINK("link");
  const std::string SRC("src");
  const std::string DEST("dest");
  const std::string KEY("key");
  const std::string TAG("tag");
  const std::string EXCLUDES("excludes");
  const std::string SRC_IS_QUEUE("srcIsQueue");
  const std::string SRC_IS_LOCAL("srcIsLocal");
  const std::string DYNAMIC("dynamic");
  const std::string SYNC("sync");
  const std::string CREDIT("credit");
  
  // parameters for deleting a Queue object
  const std::string IF_EMPTY("if_empty");
  const std::string IF_UNUSED("if_unused");
  }
  
  struct InvalidBindingIdentifier : public qpid::Exception
  {
      InvalidBindingIdentifier(const std::string& name) : qpid::Exception(name) {}
      std::string getPrefix() const { return "invalid binding"; }
  };
  
  struct BindingIdentifier
  {
      std::string exchange;
      std::string queue;
      std::string key;
  
      BindingIdentifier(const std::string& name)
      {
          std::vector<std::string> path;
          split(path, name, "/");
          switch (path.size()) {
            case 1:
              queue = path[0];
              break;
            case 2:
              exchange = path[0];
              queue = path[1];
              break;
            case 3:
              exchange = path[0];
              queue = path[1];
              key = path[2];
              break;
            default:
              throw InvalidBindingIdentifier(name);
          }
      }
  };
  
  struct ObjectAlreadyExists : public qpid::Exception
  {
      ObjectAlreadyExists(const std::string& name) : qpid::Exception(name) {}
      std::string getPrefix() const { return "object already exists"; }
  };
  
  struct UnknownObjectType : public qpid::Exception
  {
      UnknownObjectType(const std::string& type) : qpid::Exception(type) {}
      std::string getPrefix() const { return "unknown object type"; }
  };
  
  struct ReservedObjectName : public qpid::Exception
  {
      ReservedObjectName(const std::string& type) : qpid::Exception(type) {}
      std::string getPrefix() const { return std::string("names prefixed with '")
            + QPID_NAME_PREFIX + std::string("' are reserved"); }
  };
  
  struct UnsupportedTransport : public qpid::Exception
  {
      UnsupportedTransport(const std::string& type) : qpid::Exception(type) {}
      std::string getPrefix() const { return "transport is not supported"; }
  };
  
  struct InvalidParameter : public qpid::Exception
  {
      InvalidParameter(const std::string& type) : qpid::Exception(type) {}
      std::string getPrefix() const { return "invalid parameter to method call"; }
  };
  
  void Broker::createObject(const std::string& type, const std::string& name,
                            const Variant::Map& properties, bool /*strict*/, const Connection* context)
  {
      std::string userId;
      std::string connectionId;
      if (context) {
          userId = context->getUserId();
          connectionId = context->getMgmtId();
      }
      //TODO: implement 'strict' option (check there are no unrecognised properties)
      QPID_LOG (debug, "Broker::create(" << type << ", " << name << "," << properties << ")");
      if (objectFactory.createObject(*this, type, name, properties, userId, connectionId)) {
          QPID_LOG (debug, "Broker::create(" << type << ", " << name << "," << properties << ") handled by registered factory");
      } else if (type == TYPE_QUEUE) {
          bool durable(false);
          bool autodelete(false);
          std::string alternateExchange;
          Variant::Map extensions;
          for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
              // extract durable, auto-delete and alternate-exchange properties
              if (i->first == DURABLE) durable = i->second;
              else if (i->first == AUTO_DELETE) autodelete = i->second;
              else if (i->first == ALTERNATE_EXCHANGE) alternateExchange = i->second.asString();
              //treat everything else as extension properties
              else extensions[i->first] = i->second;
          }
          QueueSettings settings(durable, autodelete);
          Variant::Map unused;
          settings.populate(extensions, unused);
          qpid::amqp_0_10::translate(unused, settings.storeSettings);
          //TODO: unused doesn't take store settings into account... so can't yet implement strict
          QPID_LOG(debug, "Broker did not use the following settings (store module may): " << unused);
  
          std::pair<boost::shared_ptr<Queue>, bool> result =
              createQueue(name, settings, 0, alternateExchange, userId, connectionId);
          if (!result.second) {
              throw ObjectAlreadyExists(name);
          }
      } else if (type == TYPE_EXCHANGE || type == TYPE_TOPIC) {
          bool durable(false);
          bool autodelete(false);
          std::string exchangeType("topic");
          std::string alternateExchange;
          Variant::Map extensions;
          for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
              // extract durable, auto-delete and alternate-exchange properties
              if (i->first == DURABLE) durable = i->second;
              else if (i->first == AUTO_DELETE) autodelete = i->second;
              else if (i->first == EXCHANGE_TYPE) exchangeType = i->second.asString();
              else if (i->first == ALTERNATE_EXCHANGE) alternateExchange = i->second.asString();
              //treat everything else as extension properties
              else extensions[i->first] = i->second;
          }
          framing::FieldTable arguments;
          qpid::amqp_0_10::translate(extensions, arguments);
  
          try {
              std::pair<boost::shared_ptr<Exchange>, bool> result =
                  createExchange(name, exchangeType, durable, autodelete, alternateExchange, arguments, userId, connectionId);
              if (!result.second) {
                  throw ObjectAlreadyExists(name);
              }
          } catch (const UnknownExchangeTypeException&) {
              throw Exception(QPID_MSG("Invalid exchange type: " << exchangeType));
          }
      } else if (type == TYPE_BINDING) {
          BindingIdentifier binding(name);
          std::string exchangeType("topic");
          Variant::Map extensions;
          for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
              // extract durable, auto-delete and alternate-exchange properties
              if (i->first == EXCHANGE_TYPE) exchangeType = i->second.asString();
              //treat everything else as extension properties
              else extensions[i->first] = i->second;
          }
          framing::FieldTable arguments;
          qpid::amqp_0_10::translate(extensions, arguments);
  
          bind(binding.queue, binding.exchange, binding.key, arguments, 0, userId, connectionId);
  
      } else if (type == TYPE_LINK) {
  
          QPID_LOG (debug, "createObject: Link; name=" << name << "; args=" << properties );
  
          if (name.compare(0, QPID_NAME_PREFIX.length(), QPID_NAME_PREFIX) == 0) {
              QPID_LOG(error, "Link name='" << name << "' cannot use the reserved prefix '" << QPID_NAME_PREFIX << "'");
              throw ReservedObjectName(name);
          }
  
          std::string host;
          uint16_t port = 0;
          std::string transport = TCP_TRANSPORT;
          bool durable = false;
          std::string authMech, username, password;
  
          for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
              if (i->first == HOST) host = i->second.asString();
              else if (i->first == PORT) port = i->second.asUint16();
              else if (i->first == TRANSPORT) transport = i->second.asString();
              else if (i->first == DURABLE) durable = bool(i->second);
              else if (i->first == AUTH_MECHANISM) authMech = i->second.asString();
              else if (i->first == USERNAME) username = i->second.asString();
              else if (i->first == PASSWORD) password = i->second.asString();
              else {
                  // TODO: strict checking here
              }
          }
  
          if (!getTransportInfo(transport).connectorFactory) {
              QPID_LOG(error, "Transport '" << transport << "' not supported.");
              throw UnsupportedTransport(transport);
          }
  
          std::pair<boost::shared_ptr<Link>, bool> rc;
          rc = links.declare(name, host, port, transport, durable, authMech, username, password);
          if (!rc.first) {
              QPID_LOG (error, "Failed to create Link object, name=" << name << " remote=" << host << ":" << port <<
                        "; transport=" << transport << "; durable=" << (durable?"T":"F") << "; authMech=\"" << authMech << "\"");
              throw InvalidParameter(name);
          }
          if (!rc.second) {
              QPID_LOG (error, "Failed to create a new Link object, name=" << name << " already exists.");
              throw ObjectAlreadyExists(name);
          }
  
      } else if (type == TYPE_BRIDGE) {
  
          QPID_LOG (debug, "createObject: Bridge; name=" << name << "; args=" << properties );
  
          if (name.compare(0, QPID_NAME_PREFIX.length(), QPID_NAME_PREFIX) == 0) {
              QPID_LOG(error, "Bridge name='" << name << "' cannot use the reserved prefix '" << QPID_NAME_PREFIX << "'");
              throw ReservedObjectName(name);
          }
  
          std::string linkName;
          std::string src;
          std::string dest;
          std::string key;
          std::string id;
          std::string excludes;
          std::string queueName;
          bool durable = false;
          bool srcIsQueue = false;
          bool srcIsLocal = false;
          bool dynamic = false;
          uint16_t sync = 0;
          uint32_t credit = LinkRegistry::INFINITE_CREDIT;
  
          for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
  
              if (i->first == LINK) linkName = i->second.asString();
              else if (i->first == SRC) src = i->second.asString();
              else if (i->first == DEST) dest = i->second.asString();
              else if (i->first == KEY) key = i->second.asString();
              else if (i->first == TAG) id = i->second.asString();
              else if (i->first == EXCLUDES) excludes = i->second.asString();
              else if (i->first == SRC_IS_QUEUE) srcIsQueue = bool(i->second);
              else if (i->first == SRC_IS_LOCAL) srcIsLocal = bool(i->second);
              else if (i->first == DYNAMIC) dynamic = bool(i->second);
              else if (i->first == SYNC) sync = i->second.asUint16();
              else if (i->first == CREDIT) credit = i->second.asUint32();
              else if (i->first == DURABLE) durable = bool(i->second);
              else if (i->first == QUEUE_NAME) queueName = i->second.asString();
              else {
                  // TODO: strict checking here
              }
          }
  
          boost::shared_ptr<Link> link;
          if (linkName.empty() || !(link = links.getLink(linkName))) {
              QPID_LOG(error, "Link '" << linkName << "' not found; bridge create failed.");
              throw InvalidParameter(name);
          }
          std::pair<Bridge::shared_ptr, bool> rc =
            links.declare(name, *link, durable, src, dest, key, srcIsQueue, srcIsLocal, id, excludes,
                          dynamic, sync, credit,
                          0,
                          queueName);
  
          if (!rc.first) {
              QPID_LOG (error, "Failed to create Bridge object, name=" << name << " link=" << linkName <<
                        "; src=" << src << "; dest=" << dest << "; key=" << key);
              throw InvalidParameter(name);
          }
          if (!rc.second) {
              QPID_LOG (error, "Failed to create a new Bridge object, name=" << name << " already exists.");
              throw ObjectAlreadyExists(name);
          }
      } else {
          throw UnknownObjectType(type);
      }
  }
  
  void Broker::deleteObject(const std::string& type, const std::string& name,
                            const Variant::Map& options, const Connection* context)
  {
      std::string userId;
      std::string connectionId;
      if (context) {
          userId = context->getUserId();
          connectionId = context->getMgmtId();
      }
      QPID_LOG (debug, "Broker::delete(" << type << ", " << name << "," << options << ")");
      if (objectFactory.deleteObject(*this, type, name, options, userId, connectionId)) {
          QPID_LOG (debug, "Broker::delete(" << type << ", " << name << "," << options << ") handled by registered factory");
      } else if (type == TYPE_QUEUE) {
          // extract ifEmpty and ifUnused from options
      	bool ifUnused = false, ifEmpty = false;
          for (Variant::Map::const_iterator i = options.begin(); i != options.end(); ++i) {
              if (i->first == IF_UNUSED) ifUnused = i->second.asBool();
              else if (i->first == IF_EMPTY) ifEmpty = i->second.asBool();
          }
          deleteQueue(name, userId, connectionId,
                      boost::bind(&Broker::checkDeleteQueue, this, _1, ifUnused, ifEmpty));
      } else if (type == TYPE_EXCHANGE || type == TYPE_TOPIC) {
          deleteExchange(name, userId, connectionId);
      } else if (type == TYPE_BINDING) {
          BindingIdentifier binding(name);
          unbind(binding.queue, binding.exchange, binding.key, 0, userId, connectionId);
      } else if (type == TYPE_LINK) {
          boost::shared_ptr<Link> link = links.getLink(name);
          if (link) {
              link->close();
          }
      } else if (type == TYPE_BRIDGE) {
          boost::shared_ptr<Bridge> bridge = links.getBridge(name);
          if (bridge) {
              bridge->close();
          }
      } else {
          throw UnknownObjectType(type);
      }
  }
  
  void Broker::checkDeleteQueue(Queue::shared_ptr queue, bool ifUnused, bool ifEmpty)
  {
      if(ifEmpty && queue->getMessageCount() > 0) {
          throw qpid::framing::PreconditionFailedException(QPID_MSG("Cannot delete queue "
                                                                    << queue->getName() << "; queue not empty"));
      } else if(ifUnused && queue->getConsumerCount() > 0) {
          throw qpid::framing::PreconditionFailedException(QPID_MSG("Cannot delete queue "
                                                                    << queue->getName() << "; queue in use"));
      }
  }
  
  Manageable::status_t Broker::queryObject(const std::string& type,
                                           const std::string& name,
                                           Variant::Map& results,
                                           const Connection* context)
  {
      std::string userId;
      std::string connectionId;
      if (context) {
          userId = context->getUserId();
          connectionId = context->getMgmtId();
      }
      QPID_LOG (debug, "Broker::query(" << type << ", " << name << ")");
  
      if (type == TYPE_QUEUE)
          return queryQueue( name, userId, connectionId, results );
  
      if (type == TYPE_EXCHANGE ||
          type == TYPE_TOPIC ||
          type == TYPE_BINDING)
          return Manageable::STATUS_NOT_IMPLEMENTED;
  
      throw UnknownObjectType(type);
  }
  
  Manageable::status_t Broker::queryQueue( const std::string& name,
                                           const std::string& userId,
                                           const std::string& /*connectionId*/,
                                           Variant::Map& results )
  {
      (void) results;
      if (acl) {
          if (!acl->authorise(userId, acl::ACT_ACCESS, acl::OBJ_QUEUE, name, NULL) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue query request from " << userId));
      }
  
      boost::shared_ptr<Queue> q(queues.find(name));
      if (!q) {
          QPID_LOG(error, "Query failed: queue not found, name=" << name);
          return Manageable::STATUS_UNKNOWN_OBJECT;
      }
      q->query( results );
      return Manageable::STATUS_OK;;
  }
  
  Manageable::status_t Broker::getTimestampConfig(bool& receive,
                                                  const Connection* context)
  {
      std::string name;   // none needed for broker
      std::string userId = context->getUserId();
      if (acl && !acl->authorise(userId, acl::ACT_ACCESS, acl::OBJ_BROKER, name, NULL))  {
          throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied broker timestamp get request from " << userId));
      }
      receive = timestampRcvMsgs;
      return Manageable::STATUS_OK;
  }
  
  Manageable::status_t Broker::setTimestampConfig(const bool receive,
                                                  const Connection* context)
  {
      std::string name;   // none needed for broker
      std::string userId = context->getUserId();
      if (acl && !acl->authorise(userId, acl::ACT_UPDATE, acl::OBJ_BROKER, name, NULL)) {
          throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied broker timestamp set request from " << userId));
      }
      timestampRcvMsgs = receive;
      QPID_LOG(notice, "Receive message timestamping is " << ((timestampRcvMsgs) ? "ENABLED." : "DISABLED."));
      return Manageable::STATUS_OK;
  }
  
  void Broker::setLogLevel(const std::string& level)
  {
      QPID_LOG(notice, "Changing log level to " << level);
      std::vector<std::string> selectors;
      split(selectors, level, ", ");
      qpid::log::Logger::instance().reconfigure(selectors);
  }
  
  std::string Broker::getLogLevel()
  {
      std::string level;
      std::string sep("");
      const std::vector<std::string>& selectors = qpid::log::Logger::instance().getOptions().selectors;
      for (std::vector<std::string>::const_iterator i = selectors.begin(); i != selectors.end(); ++i) {
          level += sep + *i;
          sep = ",";
      }
      const std::vector<std::string>& disselectors = qpid::log::Logger::instance().getOptions().deselectors;
      for (std::vector<std::string>::const_iterator i = disselectors.begin(); i != disselectors.end(); ++i) {
          level += sep + "!" + *i;
          sep = ",";
      }
      return level;
  }
  
  void Broker::setLogHiresTimestamp(bool enabled)
  {
      QPID_LOG(notice, "Changing log hires timestamp to " << enabled);
      qpid::log::Logger::instance().setHiresTimestamp(enabled);
  }
  
  bool Broker::getLogHiresTimestamp()
  {
      return qpid::log::Logger::instance().getHiresTimestamp();
  }
  
  
  Manageable::status_t Broker::queueRedirect(const std::string& srcQueue,
                                             const std::string& tgtQueue,
                                             const Connection* context)
  {
      Queue::shared_ptr srcQ(queues.find(srcQueue));
      if (!srcQ) {
          QPID_LOG(error, "Queue redirect failed: source queue not found: "
              << srcQueue);
          return Manageable::STATUS_UNKNOWN_OBJECT;
      }
  
      if (!tgtQueue.empty()) {
          // NonBlank target queue creates partnership
          Queue::shared_ptr tgtQ(queues.find(tgtQueue));
          if (!tgtQ) {
              QPID_LOG(error, "Queue redirect failed: target queue not found: "
                  << tgtQueue);
              return Manageable::STATUS_UNKNOWN_OBJECT;
          }
  
          if (srcQueue.compare(tgtQueue) == 0) {
              QPID_LOG(error, "Queue redirect source queue: "
                  << tgtQueue << " cannot be its own target");
              return Manageable::STATUS_USER;
          }
  
          if (srcQ->isAutoDelete()) {
              QPID_LOG(error, "Queue redirect source queue: "
                  << srcQueue << " is autodelete and can not be part of redirect");
              return Manageable::STATUS_USER;
          }
  
          if (tgtQ->isAutoDelete()) {
              QPID_LOG(error, "Queue redirect target queue: "
                  << tgtQueue << " is autodelete and can not be part of redirect");
              return Manageable::STATUS_USER;
          }
  
          if (srcQ->getRedirectPeer()) {
              QPID_LOG(error, "Queue redirect source queue: "
                  << srcQueue << " is already redirected");
              return Manageable::STATUS_USER;
          }
  
          if (tgtQ->getRedirectPeer()) {
              QPID_LOG(error, "Queue redirect target queue: "
                  << tgtQueue << " is already redirected");
              return Manageable::STATUS_USER;
          }
  
          if (acl) {
              std::map<acl::Property, std::string> params;
              params.insert(make_pair(acl::PROP_QUEUENAME, tgtQ->getName()));
              if (!acl->authorise((context)?context->getUserId():"", acl::ACT_REDIRECT, acl::OBJ_QUEUE, srcQ->getName(), &params))
                  throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied redirect request from " << ((context)?context->getUserId():"(uknown)")));
          }
  
          // Start the backup overflow partnership
          srcQ->setRedirectPeer(tgtQ, true);
          tgtQ->setRedirectPeer(srcQ, false);
  
          // Set management state
          srcQ->setMgmtRedirectState(tgtQueue, true, true);
          tgtQ->setMgmtRedirectState(srcQueue, true, false);
  
          // Management event
          if (managementAgent.get()) {
              managementAgent->raiseEvent(_qmf::EventQueueRedirect(srcQueue, tgtQueue));
          }
  
          QPID_LOG(info, "Queue redirect complete. queue: "
              << srcQueue << " target queue: " << tgtQueue);
          return Manageable::STATUS_OK;
      } else {
          // Blank target queue destroys partnership
          Queue::shared_ptr tgtQ(srcQ->getRedirectPeer());
          if (!tgtQ) {
              QPID_LOG(error, "Queue redirect source queue: "
                  << srcQueue << " is not in redirected");
              return Manageable::STATUS_USER;
          }
  
          if (!srcQ->isRedirectSource()) {
              QPID_LOG(error, "Queue redirect source queue: "
                  << srcQueue << " is not a redirect source");
              return Manageable::STATUS_USER;
          }
  
          if (acl) {
              std::map<acl::Property, std::string> params;
              params.insert(make_pair(acl::PROP_QUEUENAME, tgtQ->getName()));
              if (!acl->authorise((context)?context->getUserId():"", acl::ACT_REDIRECT, acl::OBJ_QUEUE, srcQ->getName(), &params))
                  throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied redirect request from " << ((context)?context->getUserId():"(uknown)")));
          }
  
          queueRedirectDestroy(srcQ, tgtQ, true);
  
          return Manageable::STATUS_OK;
      }
  }
  
  
  void Broker::queueRedirectDestroy(Queue::shared_ptr srcQ,
                                    Queue::shared_ptr tgtQ,
                                    bool moveMsgs) {
      QPID_LOG(notice, "Queue redirect destroyed. queue: " << srcQ->getName()
          << " target queue: " << tgtQ->getName());
  
      tgtQ->setMgmtRedirectState(empty, false, false);
      srcQ->setMgmtRedirectState(empty, false, false);
  
      if (moveMsgs) {
          // TODO: this 'move' works in the static case but has no
          // actual locking that does what redirect needs when
          // there is a lot of traffic in flight.
          tgtQ->move(srcQ, 0);
      }
  
      Queue::shared_ptr np;
  
      tgtQ->setRedirectPeer(np, false);
      srcQ->setRedirectPeer(np, false);
  
      if (managementAgent.get()) {
          managementAgent->raiseEvent(_qmf::EventQueueRedirectCancelled(srcQ->getName(), tgtQ->getName()));
      }
  }
  
  const Broker::TransportInfo& Broker::getTransportInfo(const std::string& name) const {
      static TransportInfo nullTransportInfo;
      TransportMap::const_iterator i
          = name.empty() ? transportMap.begin() : transportMap.find(name);
      if (i == transportMap.end()) return nullTransportInfo;
      else return i->second;
  }
  
  uint16_t Broker::getPort(const std::string& name) const  {
      if (int p = getTransportInfo(name).port) {
          return p;
      } else {
          throw NoSuchTransportException(QPID_MSG("No such transport: '" << name << "'"));
      }
  }
  
  bool Broker::shouldListen(std::string transport) {
      return disabledListeningTransports.count(transport)==0;
  }
  
  void Broker::disableListening(std::string transport) {
      disabledListeningTransports.insert(transport);
  }
  
  void Broker::registerTransport(const std::string& name, boost::shared_ptr<TransportAcceptor> a, boost::shared_ptr<TransportConnector> c, uint16_t p) {
      transportMap[name] = TransportInfo(a, c, p);
      Url::addProtocol(name);
  }
  
  void Broker::accept() {
      unsigned accepting = 0;
      for (TransportMap::const_iterator i = transportMap.begin(); i != transportMap.end(); i++) {
          if (i->second.acceptor) {
              i->second.acceptor->accept(poller, &protocolRegistry);
              ++accepting;
          }
      }
      if ( accepting==0 ) {
          throw Exception(QPID_MSG("Failed to start broker: No transports are listening for incoming connections"));
      }
  }
  
  void Broker::connect(
      const std::string& name,
      const std::string& host, const std::string& port, const std::string& transport,
      boost::function2<void, int, std::string> failed)
  {
      connect(name, host, port, transport, &protocolRegistry, failed);
  }
  
  void Broker::connect(
      const std::string& name,
      const std::string& host, const std::string& port, const std::string& transport,
      sys::ConnectionCodec::Factory* f, boost::function2<void, int, std::string> failed)
  {
      boost::shared_ptr<TransportConnector> tcf = getTransportInfo(transport).connectorFactory;
      if (tcf) tcf->connect(poller, name, host, port, f, failed);
      else throw NoSuchTransportException(QPID_MSG("Unsupported transport type: " << transport));
  }
  
  int32_t Broker::queueMoveMessages(
       const std::string& srcQueue,
       const std::string& destQueue,
       uint32_t  qty,
       const Variant::Map& filter,
       const Connection* context)
  {
      Queue::shared_ptr src_queue = queues.find(srcQueue);
      if (!src_queue)
          return -1;
      Queue::shared_ptr dest_queue = queues.find(destQueue);
      if (!dest_queue)
          return -1;
  
      if (acl) {
          std::map<acl::Property, std::string> params;
          params.insert(make_pair(acl::PROP_QUEUENAME, dest_queue->getName()));
          if (!acl->authorise((context)?context->getUserId():"", acl::ACT_MOVE, acl::OBJ_QUEUE, src_queue->getName(), &params))
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied move request from " << ((context)?context->getUserId():"(uknown)")));
      }
  
      return (int32_t) src_queue->move(dest_queue, qty, &filter);
  }
  
  
  boost::shared_ptr<sys::Poller> Broker::getPoller() { return poller; }
  
  std::vector<Url>
  Broker::getKnownBrokersImpl()
  {
      return knownBrokers;
  }
  
  bool Broker::deferDeliveryImpl(const std::string&, const Message&)
  { return false; }
  
  const std::string Broker::TCP_TRANSPORT("tcp");
  
  std::pair<boost::shared_ptr<Queue>, bool> Broker::createQueue(
      const std::string& name,
      const QueueSettings& constSettings,
      const OwnershipToken* owner,
      const std::string& alternateExchange,
      const std::string& userId,
      const std::string& connectionId)
  {
      QueueSettings settings(constSettings); // So we can modify them
      if (acl) {
          std::map<acl::Property, std::string> params;
          params.insert(make_pair(acl::PROP_ALTERNATE, alternateExchange));
          params.insert(make_pair(acl::PROP_DURABLE, settings.durable ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_EXCLUSIVE, owner ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_AUTODELETE, settings.autodelete ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_POLICYTYPE, settings.getLimitPolicy()));
          params.insert(make_pair(acl::PROP_PAGING, settings.paging ? _TRUE : _FALSE));
          if (settings.paging) {
              params.insert(make_pair(acl::PROP_MAXPAGES, boost::lexical_cast<string>(settings.maxPages ? settings.maxPages : DEFAULT_MAX_PAGES)));
              params.insert(make_pair(acl::PROP_MAXPAGEFACTOR, boost::lexical_cast<string>(settings.pageFactor ? settings.pageFactor : DEFAULT_PAGE_FACTOR)));
          }
          if (settings.maxDepth.hasCount())
              params.insert(make_pair(acl::PROP_MAXQUEUECOUNT, boost::lexical_cast<string>(settings.maxDepth.getCount() ? settings.maxDepth.getCount() : std::numeric_limits<uint64_t>::max())));
          if (settings.maxDepth.hasSize())
              params.insert(make_pair(acl::PROP_MAXQUEUESIZE, boost::lexical_cast<string>(settings.maxDepth.getSize() ? settings.maxDepth.getSize() :  std::numeric_limits<uint64_t>::max())));
          else
              params.insert(make_pair(acl::PROP_MAXQUEUESIZE, boost::lexical_cast<string>(config.queueLimit)));
          if (settings.durable) {
              params.insert(make_pair(acl::PROP_MAXFILECOUNT, boost::lexical_cast<string>(settings.maxFileCount ? settings.maxFileCount : 8)));
              params.insert(make_pair(acl::PROP_MAXFILESIZE, boost::lexical_cast<string>(settings.maxFileSize ? settings.maxFileSize : 24)));
          }
  
          if (!acl->authorise(userId,acl::ACT_CREATE,acl::OBJ_QUEUE,name,&params) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
  
          if (!queues.find(name))
              if (!acl->approveCreateQueue(userId,name) )
                  throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue create request from " << userId));
      }
  
      Exchange::shared_ptr alternate;
      if (!alternateExchange.empty()) {
          alternate = exchanges.get(alternateExchange);
          if (!alternate) throw framing::NotFoundException(QPID_MSG("Alternate exchange does not exist: " << alternateExchange));
      }
  
      // Identify queues that won't survive a failover: exclusive, auto-delete with no delay.
      if (owner && settings.autodelete && !settings.autoDeleteDelay)
          settings.isTemporary = true;
  
      std::pair<Queue::shared_ptr, bool> result =
          queues.declare(name, settings, alternate, false/*recovering*/,
                         owner, connectionId, userId);
      if (result.second) {
          //add default binding:
          result.first->bind(exchanges.getDefault(), name);
          QPID_LOG_CAT(debug, model, "Create queue. name:" << name
              << " user:" << userId
              << " rhost:" << connectionId
              << " durable:" << (settings.durable ? "T" : "F")
              << " owner:" << owner
              << " autodelete:" << (settings.autodelete ? "T" : "F")
              << " alternateExchange:" << alternateExchange );
      }
      return result;
  }
  
  void Broker::deleteQueue(boost::shared_ptr<Queue> queue, const std::string& userId,
                           const std::string& connectionId, QueueFunctor check)
  {
      const std::string& name = queue->getName();
      QPID_LOG_CAT(debug, model, "Deleting queue. name:" << name
                   << " user:" << userId
                   << " rhost:" << connectionId
      );
      if (acl) {
          std::map<acl::Property, std::string> params;
          boost::shared_ptr<Exchange> altEx = queue->getAlternateExchange();
          params.insert(make_pair(acl::PROP_ALTERNATE, (altEx) ? altEx->getName() : "" ));
          params.insert(make_pair(acl::PROP_DURABLE, queue->isDurable() ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_EXCLUSIVE, queue->hasExclusiveOwner() ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_AUTODELETE, queue->isAutoDelete() ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_POLICYTYPE, queue->getSettings().getLimitPolicy()));
  
          if (!acl->authorise(userId,acl::ACT_DELETE,acl::OBJ_QUEUE,name,&params) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied queue delete request from " << userId));
      }
      if (check) check(queue);
      if (acl)
          acl->recordDestroyQueue(name);
      Queue::shared_ptr peerQ(queue->getRedirectPeer());
      if (peerQ)
          queueRedirectDestroy(queue->isRedirectSource() ? queue : peerQ,
                               queue->isRedirectSource() ? peerQ : queue,
                               false);
      queues.destroy(queue, connectionId, userId);
  
  }
  
  void Broker::deleteQueue(const std::string& name, const std::string& userId,
                           const std::string& connectionId, QueueFunctor check)
  {
      Queue::shared_ptr queue = queues.find(name);
      if (queue) {
          deleteQueue(queue, userId, connectionId, check);
      } else {
          throw framing::NotFoundException(QPID_MSG("Delete failed. No such queue: " << name));
      }
  }
  
  std::pair<Exchange::shared_ptr, bool> Broker::createExchange(
      const std::string& name,
      const std::string& type,
      bool durable,
      bool autodelete,
      const std::string& alternateExchange,
      const qpid::framing::FieldTable& arguments,
      const std::string& userId,
      const std::string& connectionId)
  {
      if (acl) {
          std::map<acl::Property, std::string> params;
          params.insert(make_pair(acl::PROP_TYPE, type));
          params.insert(make_pair(acl::PROP_ALTERNATE, alternateExchange));
          params.insert(make_pair(acl::PROP_DURABLE, durable ? _TRUE : _FALSE));
          params.insert(make_pair(acl::PROP_AUTODELETE, autodelete ? _TRUE : _FALSE));
          if (!acl->authorise(userId,acl::ACT_CREATE,acl::OBJ_EXCHANGE,name,&params) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied exchange create request from " << userId));
      }
  
      Exchange::shared_ptr alternate;
      if (!alternateExchange.empty()) {
          alternate = exchanges.get(alternateExchange);
          if (!alternate) throw framing::NotFoundException(QPID_MSG("Alternate exchange does not exist: " << alternateExchange));
      }
  
      std::pair<Exchange::shared_ptr, bool> result;
      result = exchanges.declare(
          name, type, durable, autodelete, arguments, alternate, connectionId, userId);
      if (result.second) {
          if (durable) {
              store->create(*result.first, arguments);
          }
          QPID_LOG_CAT(debug, model, "Create exchange. name:" << name
              << " user:" << userId
              << " rhost:" << connectionId
              << " type:" << type
              << " alternateExchange:" << alternateExchange
              << " durable:" << (durable ? "T" : "F")
              << " autodelete:" << (autodelete ? "T" : "F"));
      }
      return result;
  }
  
  void Broker::deleteExchange(const std::string& name, const std::string& userId,
                             const std::string& connectionId)
  {
      QPID_LOG_CAT(debug, model, "Deleting exchange. name:" << name
          << " user:" << userId
          << " rhost:" << connectionId);
      if (name.empty()) {
          throw framing::InvalidArgumentException(QPID_MSG("Delete not allowed for default exchange"));
      }
      Exchange::shared_ptr exchange(exchanges.get(name));
      if (!exchange) throw framing::NotFoundException(QPID_MSG("Delete failed. No such exchange: " << name));
  
      if (acl) {
          std::map<acl::Property, std::string> params;
          Exchange::shared_ptr altEx = exchange->getAlternate();
          params.insert(make_pair(acl::PROP_TYPE, exchange->getType()));
          params.insert(make_pair(acl::PROP_ALTERNATE, (altEx) ? altEx->getName() : "" ));
          params.insert(make_pair(acl::PROP_DURABLE, exchange->isDurable() ? _TRUE : _FALSE));
  
          if (!acl->authorise(userId,acl::ACT_DELETE,acl::OBJ_EXCHANGE,name,&params) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied exchange delete request from " << userId));
      }
  
      if (exchange->inUseAsAlternate()) throw framing::NotAllowedException(QPID_MSG("Cannot delete " << name <<", in use as alternate-exchange."));
      if (exchange->isDurable()) store->destroy(*exchange);
      if (exchange->getAlternate()) exchange->getAlternate()->decAlternateUsers();
      exchanges.destroy(name, connectionId,  userId);
  }
  
  void Broker::bind(const std::string& queueName,
                    const std::string& exchangeName,
                    const std::string& key,
                    const qpid::framing::FieldTable& arguments,
                    const OwnershipToken* owner,
                    const std::string& userId,
                    const std::string& connectionId)
  {
      if (acl) {
          std::map<acl::Property, std::string> params;
          params.insert(make_pair(acl::PROP_QUEUENAME, queueName));
          params.insert(make_pair(acl::PROP_ROUTINGKEY, key));
  
          if (!acl->authorise(userId,acl::ACT_BIND,acl::OBJ_EXCHANGE,exchangeName,&params))
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied exchange bind request from " << userId));
      }
      if (exchangeName.empty()) {
          throw framing::InvalidArgumentException(QPID_MSG("Bind not allowed for default exchange"));
      }
  
      Queue::shared_ptr queue = queues.find(queueName);
      Exchange::shared_ptr exchange = exchanges.get(exchangeName);
      if (!queue) {
          throw framing::NotFoundException(QPID_MSG("Bind failed. No such queue: " << queueName));
      } else if (!exchange) {
          throw framing::NotFoundException(QPID_MSG("Bind failed. No such exchange: " << exchangeName));
      } else if (queue->hasExclusiveOwner() && !queue->isExclusiveOwner(owner)) {
          throw framing::ResourceLockedException(QPID_MSG("Cannot bind queue "
                                                 << queue->getName() << "; it is exclusive to another session"));
      } else {
          if (queue->bind(exchange, key, arguments)) {
              getBrokerObservers().bind(exchange, queue, key, arguments);
              if (managementAgent.get()) {
                  managementAgent->raiseEvent(_qmf::EventBind(connectionId, userId, exchangeName,
                                                    queueName, key, ManagementAgent::toMap(arguments)));
              }
              QPID_LOG_CAT(debug, model, "Create binding. exchange:" << exchangeName
                  << " queue:" << queueName
                  << " key:" << key
                  << " arguments:" << arguments
                  << " user:" << userId
                  << " rhost:" << connectionId);
          }
      }
  }
  
  void Broker::unbind(const std::string& queueName,
                      const std::string& exchangeName,
                      const std::string& key,
                      const OwnershipToken* owner,
                      const std::string& userId,
                      const std::string& connectionId)
  {
      if (acl) {
          std::map<acl::Property, std::string> params;
          params.insert(make_pair(acl::PROP_QUEUENAME, queueName));
          params.insert(make_pair(acl::PROP_ROUTINGKEY, key));
          if (!acl->authorise(userId,acl::ACT_UNBIND,acl::OBJ_EXCHANGE,exchangeName,&params) )
              throw framing::UnauthorizedAccessException(QPID_MSG("ACL denied exchange unbind request from " << userId));
      }
      if (exchangeName.empty()) {
          throw framing::InvalidArgumentException(QPID_MSG("Unbind not allowed for default exchange"));
      }
      Queue::shared_ptr queue = queues.find(queueName);
      Exchange::shared_ptr exchange = exchanges.get(exchangeName);
      if (!queue) {
          throw framing::NotFoundException(QPID_MSG("Unbind failed. No such queue: " << queueName));
      } else if (!exchange) {
          throw framing::NotFoundException(QPID_MSG("Unbind failed. No such exchange: " << exchangeName));
      } else if (queue->hasExclusiveOwner() && !queue->isExclusiveOwner(owner)) {
          throw framing::ResourceLockedException(QPID_MSG("Cannot unbind queue "
                                                 << queue->getName() << "; it is exclusive to another session"));
      } else {
          if (exchange->unbind(queue, key, 0)) {
              if (exchange->isDurable() && queue->isDurable()) {
                  store->unbind(*exchange, *queue, key, qpid::framing::FieldTable());
              }
              getBrokerObservers().unbind(
                  exchange, queue, key, framing::FieldTable());
              if (managementAgent.get()) {
                  managementAgent->raiseEvent(_qmf::EventUnbind(connectionId, userId, exchangeName, queueName, key));
              }
              QPID_LOG_CAT(debug, model, "Delete binding. exchange:" << exchangeName
                  << " queue:" << queueName
                  << " key:" << key
                  << " user:" << userId
                  << " rhost:" << connectionId);
          }
      }
  }
  
  // FIXME aconway 2012-04-27: access to linkClientProperties is
  // not properly thread safe, you could lose fields if 2 threads
  // attempt to add a field concurrently.
  
  framing::FieldTable Broker::getLinkClientProperties() const {
      sys::Mutex::ScopedLock l(linkClientPropertiesLock);
      return linkClientProperties;
  }
  
  void Broker::setLinkClientProperties(const framing::FieldTable& ft) {
      sys::Mutex::ScopedLock l(linkClientPropertiesLock);
      linkClientProperties = ft;
  }
  
  }} // namespace qpid::broker
  
  