void AclReader::printConnectionRules(const std::string name, const AclData::bwHostRuleSet& rules) const {
     QPID_LOG(debug, "ACL: " << name << " Connection Rule list : " << rules.size() << " rules found :");
     int cnt = 1;
     for (AclData::bwHostRuleSetItr i=rules.begin(); i<rules.end(); i++,cnt++) {
         QPID_LOG(debug, "ACL:   " << std::setfill(' ') << std::setw(2) << cnt << " " << i->toString());
     }
 }

void TimerWarnings::log() {
    if (!taskStats.empty() && nextReport < now()) {
        for (TaskStatsMap::iterator i = taskStats.begin(); i != taskStats.end(); ++i) {
            std::string task = i->first;
            TaskStats& stats = i->second;
            if (stats.lateDelay.count)
                QPID_LOG(info, task << " task late "
                         << stats.lateDelay.count << " times by "
                         << stats.lateDelay.average()/TIME_MSEC << "ms on average.");

            if (stats.overranOverrun.count)
                QPID_LOG(info, task << " task overran "
                         << stats.overranOverrun.count << " times by "
                         << stats.overranOverrun.average()/TIME_MSEC << "ms (taking "
                         << stats.overranTime.average() << "ns) on average.");

            if (stats.lateAndOverranOverrun.count)
                QPID_LOG(info, task << " task late and overran "
                         << stats.lateAndOverranOverrun.count << " times: late "
                         << stats.lateAndOverranDelay.average()/TIME_MSEC << "ms, overran "
                         << stats.lateAndOverranOverrun.average()/TIME_MSEC << "ms (taking "
                         << stats.lateAndOverranTime.average() << "ns) on average.");

        }
        nextReport = AbsTime(now(), interval);
        taskStats.clear();
    }
}

// Handler for deliverEventQueue.
// This thread decodes frames from events.
void Cluster::deliveredEvent(const Event& e) {
    if (e.isCluster()) {
        EventFrame ef(e, e.getFrame());
        // Stop the deliverEventQueue on update offers.
        // This preserves the connection decoder fragments for an update.
        // Only do this for the two brokers that are directly involved in this
        // offer: the one making the offer, or the one receiving it.
        const ClusterUpdateOfferBody* offer = castUpdateOffer(ef.frame.getBody());
        if (offer && ( e.getMemberId() == self || MemberId(offer->getUpdatee()) == self) ) {
            QPID_LOG(info, *this << " stall for update offer from " << e.getMemberId()
                     << " to " << MemberId(offer->getUpdatee()));
            deliverEventQueue.stop();
        }
        deliverFrame(ef);
    }
    else if(!discarding) {
        if (e.isControl())
            deliverFrame(EventFrame(e, e.getFrame()));
        else {
            try { decoder.decode(e, e.getData()); }
            catch (const Exception& ex) {
                // Close a connection that is sending us invalid data.
                QPID_LOG(error, *this << " aborting connection "
                         << e.getConnectionId() << ": " << ex.what());
                framing::AMQFrame abort((ClusterConnectionAbortBody()));
                deliverFrame(EventFrame(EventHeader(CONTROL, e.getConnectionId()), abort));
            }
        }
    }
}

qpid::sys::ConnectionCodec* ProtocolImpl::create(const qpid::framing::ProtocolVersion& v, qpid::sys::OutputControl& out, const std::string& id, const qpid::sys::SecuritySettings& external)
{
    if (v == qpid::framing::ProtocolVersion(1, 0)) {
        if (v.getProtocol() == qpid::framing::ProtocolVersion::SASL) {
            if (broker.getOptions().auth) {
                QPID_LOG(info, "Using AMQP 1.0 (with SASL layer)");
                return new qpid::broker::amqp::Sasl(out, id, broker, *interconnects,
                                                    qpid::SaslFactory::getInstance().createServer(broker.getOptions().realm,broker.getOptions().requireEncrypted, external),
                                                    domain);
            } else {
                std::auto_ptr<SaslServer> authenticator(new qpid::NullSaslServer(broker.getOptions().realm));
                QPID_LOG(info, "Using AMQP 1.0 (with dummy SASL layer)");
                return new qpid::broker::amqp::Sasl(out, id, broker, *interconnects, authenticator, domain);
            }
        } else {
            if (broker.getOptions().auth) {
                throw qpid::Exception("SASL layer required!");
            } else {
                QPID_LOG(info, "Using AMQP 1.0 (no SASL layer)");
                return new qpid::broker::amqp::Connection(out, id, broker, *interconnects, false, domain);
            }
        }
    }
    return 0;
}

//delivery-annotations, message-annotations, application-properties, amqp-value
bool MessageReader::onStartMap(uint32_t count, const CharSequence& elements, const CharSequence& raw, const Descriptor* descriptor)
{
    if (delegate) {
        return delegate->onStartMap(count, elements, raw, descriptor);
    } else {
        if (!descriptor) {
            QPID_LOG(warning, "Expected described type but got no descriptor for map.");
            return false;
        } else if (descriptor->match(DELIVERY_ANNOTATIONS_SYMBOL, DELIVERY_ANNOTATIONS_CODE)) {
            onDeliveryAnnotations(elements, raw);
            return false;
        } else if (descriptor->match(MESSAGE_ANNOTATIONS_SYMBOL, MESSAGE_ANNOTATIONS_CODE)) {
            onMessageAnnotations(elements, raw);
            return false;
        } else if (descriptor->match(FOOTER_SYMBOL, FOOTER_CODE)) {
            onFooter(elements, raw);
            return false;
        } else if (descriptor->match(APPLICATION_PROPERTIES_SYMBOL, APPLICATION_PROPERTIES_CODE)) {
            onApplicationProperties(elements, raw);
            return false;
        } else if (descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
            onAmqpValue(elements, qpid::amqp::typecodes::MAP_NAME);
            return false;
        } else {
            QPID_LOG(warning, "Unexpected described map: " << *descriptor);
            return false;
        }
    }
}

 // Debug aid
 void AclReader::printRules() const {
     QPID_LOG(debug, "ACL: Rule list: " << rules.size() << " ACL rules found:");
     int cnt = 1;
     for (rlCitr i=rules.begin(); i<rules.end(); i++,cnt++) {
         QPID_LOG(debug, "ACL:   " << std::setfill(' ') << std::setw(2) << cnt << " " << (*i)->toString());
     }
 }

//
// closed - called during Connection's destructor
//
void ConnectionCounter::closed(broker::Connection& connection) {
    QPID_LOG(trace, "ACL ConnectionCounter closed: " << connection.getMgmtId()
        << ", userId:" << connection.getUserId());

    Mutex::ScopedLock locker(dataLock);

    connectCountsMap_t::iterator eRef = connectProgressMap.find(connection.getMgmtId());
    if (eRef != connectProgressMap.end()) {
        if ((*eRef).second == C_OPENED){
            // Normal case: connection was created and opened.
            // Decrement user in-use counts
            releaseLH(connectByNameMap,
                      connection.getUserId());
        } else {
            // Connection was created but not opened.
            // Don't decrement user count.
        }

        // Decrement host in-use count.
        releaseLH(connectByHostMap,
                  getClientHost(connection.getMgmtId()));

        // destroy connection progress indicator
        connectProgressMap.erase(eRef);

    } else {
        // connection not found in progress map
        QPID_LOG(notice, "ACL ConnectionCounter closed info for '" << connection.getMgmtId()
            << "' not found in connection state pool");
    }

    // total connections
    totalCurrentConnections -= 1;
}

/**
 * Called when a delivery is writable
 */
void OutgoingFromRelay::handle(pn_delivery_t* delivery)
{
    void* context = pn_delivery_get_context(delivery);
    BufferedTransfer* transfer = reinterpret_cast<BufferedTransfer*>(context);
    assert(transfer);
    if (pn_delivery_writable(delivery)) {
        if (transfer->write(link)) {
            outgoingMessageSent();
            QPID_LOG(debug, "Sent relayed message " << name << " [" << relay.get() << "]");
        } else {
            QPID_LOG(error, "Failed to send relayed message " << name << " [" << relay.get() << "]");
        }
    }
    if (pn_delivery_updated(delivery)) {
        uint64_t d = transfer->updated();
        switch (d) {
          case PN_ACCEPTED:
            outgoingMessageAccepted();
            break;
          case PN_REJECTED:
          case PN_RELEASED://TODO: not quite true...
          case PN_MODIFIED://TODO: not quite true...
            outgoingMessageRejected();
            break;
          default:
            QPID_LOG(warning, "Unhandled disposition: " << d);
        }
    }
}

//
// approveConnection
//  check total connections, connections from IP, connections by user and
//  disallow if over any limit
//
bool ConnectionCounter::approveConnection(
        const broker::Connection& connection,
        bool enforcingConnectionQuotas,
        uint16_t connectionUserQuota )
{
    const std::string& hostName(getClientHost(connection.getMgmtId()));
    const std::string& userName(              connection.getUserId());

    Mutex::ScopedLock locker(dataLock);

    // Bump state from CREATED to OPENED
    (void) countConnectionLH(connectProgressMap, connection.getMgmtId(),
                             C_OPENED, false, false);

    // Approve total connections
    bool okTotal  = true;
    if (totalLimit > 0) {
        okTotal = totalCurrentConnections <= totalLimit;
        QPID_LOG(trace, "ACL ConnectionApprover totalLimit=" << totalLimit
                 << " curValue=" << totalCurrentConnections
                 << " result=" << (okTotal ? "allow" : "deny"));
    }

    // Approve by IP host connections
    bool okByIP   = limitApproveLH(connectByHostMap, hostName, hostLimit, true);

    // Count and Approve the connection by the user
    bool okByUser = countConnectionLH(connectByNameMap, userName,
                                      connectionUserQuota, true,
                                      enforcingConnectionQuotas);

    // Emit separate log for each disapproval
    if (!okTotal) {
        QPID_LOG(error, "Client max total connection count limit of " << totalLimit
                 << " exceeded by '"
                 << connection.getMgmtId() << "', user: '"
                 << userName << "'. Connection refused");
    }
    if (!okByIP) {
        QPID_LOG(error, "Client max per-host connection count limit of "
                 << hostLimit << " exceeded by '"
                 << connection.getMgmtId() << "', user: '"
                 << userName << "'. Connection refused.");
    }
    if (!okByUser) {
        QPID_LOG(error, "Client max per-user connection count limit of "
                 << connectionUserQuota << " exceeded by '"
                 << connection.getMgmtId() << "', user: '"
                 << userName << "'. Connection refused.");
    }

    // Count/Event once for each disapproval
    bool result = okTotal && okByIP && okByUser;
    if (!result) {
        acl.reportConnectLimit(userName, hostName);
    }

    return result;
}

 void received(Message& message)
 {
     QPID_LOG(debug, "received: " << message.getData() << " for " << message.getDestination());
     if (message.getDestination() == source) {
         receivedFromSource(message);
     } else if (message.getDestination() == control) {
         receivedFromControl(message);
     } else {
         QPID_LOG(error, "Unexpected message: " << message.getData() << " to " << message.getDestination());
     }
 }

void Quorum::start(boost::shared_ptr<sys::Poller> p) {
    poller = p;
    QPID_LOG(debug, "Connecting to quorum service.");
    cman = cman_init(0);
    if (cman == 0) throw ErrnoException("Can't connect to cman service");
    if (!cman_is_quorate(cman)) {
        QPID_LOG(notice, "Waiting for cluster quorum.");
        while(!cman_is_quorate(cman)) sys::sleep(5);
    }
    int err = cman_start_notification(cman, cmanCallbackFn);
    if (err != 0) throw ErrnoException("Can't register for cman notifications");
    watch(getFd());
}

void ConnectionContext::attach(pn_session_t* /*session*/, pn_link_t* link, int credit)
{
    qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
    QPID_LOG(debug, "Attaching link " << link << ", state=" << pn_link_state(link));
    pn_link_open(link);
    QPID_LOG(debug, "Link attached " << link << ", state=" << pn_link_state(link));
    if (credit) pn_link_flow(link, credit);
    wakeupDriver();
    while (pn_link_state(link) & PN_REMOTE_UNINIT) {
        QPID_LOG(debug, "waiting for confirmation of link attach for " << link << ", state=" << pn_link_state(link));
        wait();
    }
}

    void enqueue(TransactionContext* ,
                 const boost::intrusive_ptr<PersistableMessage>& pmsg,
                 const PersistableQueue& )
    {
        qpid::broker::amqp_0_10::MessageTransfer* msg = dynamic_cast<qpid::broker::amqp_0_10::MessageTransfer*>(pmsg.get());
        assert(msg);

        // Dump the message if there is a dump file.
        if (dump.get()) {
            msg->getFrames().getMethod()->print(*dump);
            *dump  << endl << "  ";
            msg->getFrames().getHeaders()->print(*dump);
            *dump << endl << "  ";
            *dump << msg->getFrames().getContentSize() << endl;
        }

        // Check the message for special instructions.
        string data = msg->getFrames().getContent();
        size_t i = string::npos;
        size_t j = string::npos;
        if (strncmp(data.c_str(), TEST_STORE_DO.c_str(), strlen(TEST_STORE_DO.c_str())) == 0
            && (i = data.find(name+"[")) != string::npos
            && (j = data.find("]", i)) != string::npos)
        {
            size_t start = i+name.size()+1;
            string action = data.substr(start, j-start);

            if (action == EXCEPTION) {
                throw Exception(QPID_MSG("TestStore " << name << " throwing exception for: " << data));
            }
            else if (action == EXIT_PROCESS) {
                // FIXME aconway 2009-04-10: this is a dubious way to
                // close the process at best, it can cause assertions or seg faults
                // rather than clean exit.
                QPID_LOG(critical, "TestStore " << name << " forcing process exit for: " << data);
                exit(0);
            }
            else if (strncmp(action.c_str(), ASYNC.c_str(), strlen(ASYNC.c_str())) == 0) {
                std::string delayStr(action.substr(ASYNC.size()));
                int delay = boost::lexical_cast<int>(delayStr);
                threads.push_back(Thread(*new Completer(msg, delay)));
            }
            else {
                QPID_LOG(error, "TestStore " << name << " unknown action " << action);
                msg->enqueueComplete();
            }
        }
        else
            msg->enqueueComplete();
    }

/**
 * Expects lock to be held by caller
 */
void ConnectionContext::wakeupDriver()
{
    switch (state) {
      case CONNECTED:
        haveOutput = true;
        transport->activateOutput();
        QPID_LOG(debug, "wakeupDriver()");
        break;
      case DISCONNECTED:
      case CONNECTING:
        QPID_LOG(error, "wakeupDriver() called while not connected");
        break;
    }
}

bool SaslClient::onStartList(uint32_t count, const CharSequence& arguments, const CharSequence& /*full raw data*/, const Descriptor* descriptor)
{
    if (!descriptor) {
        QPID_LOG(error, "Expected described type in SASL negotiation but got no descriptor");
    } else if (descriptor->match(SASL_MECHANISMS_SYMBOL, SASL_MECHANISMS_CODE)) {
        QPID_LOG(trace, "Reading SASL-MECHANISMS");
        Decoder decoder(arguments.data, arguments.size);
        if (count != 1) QPID_LOG(error, "Invalid SASL-MECHANISMS frame; exactly one field expected, got " << count);
        SaslMechanismsReader reader(*this);
        decoder.read(reader);
    } else if (descriptor->match(SASL_CHALLENGE_SYMBOL, SASL_CHALLENGE_CODE)) {
        QPID_LOG(trace, "Reading SASL-CHALLENGE");
        Decoder decoder(arguments.data, arguments.size);
        if (count != 1) QPID_LOG(error, "Invalid SASL-CHALLENGE frame; exactly one field expected, got " << count);
        SaslChallengeReader reader(*this);
        decoder.read(reader);
    } else if (descriptor->match(SASL_OUTCOME_SYMBOL, SASL_OUTCOME_CODE)) {
        QPID_LOG(trace, "Reading SASL-OUTCOME");
        Decoder decoder(arguments.data, arguments.size);
        if (count == 1) {
            SaslOutcomeReader reader(*this, false);
            decoder.read(reader);
        } else if (count == 2) {
            SaslOutcomeReader reader(*this, true);
            decoder.read(reader);
        } else {
            QPID_LOG(error, "Invalid SASL-OUTCOME frame; got " << count << " fields");
        }
    } else {
        QPID_LOG(error, "Unexpected descriptor in SASL negotiation: " << *descriptor);
    }
    return false;
}

void Cluster::memberUpdate(Lock& l) {
    // Ignore config changes while we are joining.
    if (state < CATCHUP) return;
    QPID_LOG(info, *this << " member update: " << map);
    size_t aliveCount = map.aliveCount();
    assert(map.isAlive(self));
    failoverExchange->updateUrls(getUrls(l));

    // Mark store clean if I am the only broker, dirty otherwise.
    if (store.hasStore()) {
        if (aliveCount == 1) {
            if (store.getState() != STORE_STATE_CLEAN_STORE) {
                QPID_LOG(notice, *this << "Sole member of cluster, marking store clean.");
                store.clean(Uuid(true));
            }
        }
        else {
            if (store.getState() != STORE_STATE_DIRTY_STORE) {
                QPID_LOG(notice, "Running in a cluster, marking store dirty.");
                store.dirty();
            }
        }
    }

    // If I am the last member standing, set queue policies.
    if (aliveCount == 1 && lastAliveCount > 1 && state >= CATCHUP) {
        QPID_LOG(notice, *this << " last broker standing, update queue policies");
        lastBroker = true;
        broker.getQueues().updateQueueClusterState(true);
    }
    else if (aliveCount > 1 && lastBroker) {
        QPID_LOG(notice, *this << " last broker standing joined by " << aliveCount-1
                 << " replicas, updating queue policies.");
        lastBroker = false;
        broker.getQueues().updateQueueClusterState(false);
    }
    lastAliveCount = aliveCount;

    // Close connections belonging to members that have left the cluster.
    ConnectionMap::iterator i = connections.begin();
    while (i != connections.end()) {
        ConnectionMap::iterator j = i++;
        MemberId m = j->second->getId().getMember();
        if (m != self && !map.isMember(m)) {
            j->second->close();
            erase(j->second->getId(), l);
        }
    }
}

// Called in update thread or deliver thread.
void Cluster::checkUpdateIn(Lock& l) {
    if (state != UPDATEE) return; // Wait till we reach the stall point.
    if (!updateClosed) return;  // Wait till update connection closes.
    if (updatedMap) { // We're up to date
        map = *updatedMap;
        mcast.mcastControl(ClusterReadyBody(ProtocolVersion(), myUrl.str()), self);
        state = CATCHUP;
        /* In CATCHUP mode the update has finished, and we are consuming
        ** whatever backlog of messages has built up during the update.
        ** We should enable queue events here, or messages that are received
        ** during this phase will not be replicated properly. ( If there are
        ** relevant replication queues. ) */
        broker.getQueueEvents().enable();
        memberUpdate(l);
        // Must be called *after* memberUpdate() to avoid sending an extra update.
        failoverExchange->setReady();
        // NB: don't updateMgmtMembership() here as we are not in the deliver
        // thread. It will be updated on delivery of the "ready" we just mcast.
        broker.setClusterUpdatee(false);
        broker.setAcl(acl);     // Restore ACL
        discarding = false;     // OK to set, we're stalled for update.
        QPID_LOG(notice, *this << " update complete, starting catch-up.");
        QPID_LOG(debug, debugSnapshot()); // OK to call because we're stalled.
        if (mAgent) {
            // Update management agent now, after all update activity is complete.
            updateDataExchange->updateManagementAgent(mAgent);
            mAgent->suppress(false); // Enable management output.
            mAgent->clusterUpdate();
        }
        // Restore alternate exchange settings on exchanges.
        broker.getExchanges().eachExchange(
            boost::bind(&broker::Exchange::recoveryComplete, _1,
                        boost::ref(broker.getExchanges())));
        enableClusterSafe();    // Enable cluster-safe assertions
        deliverEventQueue.start();
        // FIXME aconway 2012-04-04: unregister/delete Update[Data]Exchange
        updateDataExchange.reset();
        broker.getExchanges().destroy(UpdateDataExchange::EXCHANGE_NAME);
        broker.getExchanges().destroy(UpdateClient::UPDATE);
    }
    else if (updateRetracted) { // Update was retracted, request another update
        updateRetracted = false;
        updateClosed = false;
        state = JOINER;
        QPID_LOG(notice, *this << " update retracted, sending new update request.");
        mcast.mcastControl(ClusterUpdateRequestBody(ProtocolVersion(), myUrl.str()), self);
        deliverEventQueue.start();
    }
}

 void receivedFromSource(Message& message)
 {
     QPID_LOG(debug, "transfering  " << (transfered+1) << " of " << expected);
     message.getDeliveryProperties().setRoutingKey(destination);
     async(sourceSubscription.getSession()).messageTransfer(arg::content=message);
     if (++transfered == expected) {
         QPID_LOG(info, "completed job: " << transfered << " messages shifted from " <<
                  source << " to " << destination);
         sourceSubscription.accept(sourceSubscription.getUnaccepted());
         sourceSubscription.getSession().txCommit();
         sourceSubscription.cancel();
         //grant credit to allow broker to send us another control message
         controlSubscription.grantMessageCredit(1);
     }
 }

boost::shared_ptr<FieldValue> convertString(const std::string& value, const std::string& encoding)
{
    bool large = value.size() > std::numeric_limits<uint16_t>::max();
    if (encoding.empty() || encoding == amqp0_10_binary || encoding == binary) {
        if (large) {
            return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
        } else {
            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x90));
        }
    } else if (encoding == utf8) {
        if (!large)
            return boost::shared_ptr<FieldValue>(new Str16Value(value));
        throw Exception(QPID_MSG("Could not encode utf8 character string - too long (" << value.size() << " bytes)"));
    } else if (encoding == utf16) {
        if (!large)
            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x96));
        throw Exception(QPID_MSG("Could not encode utf16 character string - too long (" << value.size() << " bytes)"));
    } else if (encoding == iso885915) {
        if (!large)
            return boost::shared_ptr<FieldValue>(new Var16Value(value, 0x94));
        throw Exception(QPID_MSG("Could not encode iso-8859-15 character string - too long (" << value.size() << " bytes)"));
    } else {
        // the encoding was not recognised
        QPID_LOG(warning, "Unknown byte encoding: [" << encoding << "], encoding as vbin32.");
        return boost::shared_ptr<FieldValue>(new Var32Value(value, 0xa0));
    }
}

size_t Sasl::encode(char* buffer, size_t size)
{
    if (state == AUTHENTICATED) {
        if (securityLayer.get()) return securityLayer->encode(buffer, size);
        else return connection.encode(buffer, size);
    } else {
        size_t encoded = 0;
        if (writeHeader) {
            encoded += writeProtocolHeader(buffer, size);
            if (!encoded) return 0;
            writeHeader = false;
        }
        if (encoded < size) {
            encoded += write(buffer + encoded, size - encoded);
        }
        if (state == SUCCESS_PENDING) {
            state = AUTHENTICATED;
        } else if (state == FAILURE_PENDING) {
            state = FAILED;
        } else {
            haveOutput = (encoded == size);
        }
        QPID_LOG(trace, id << " Sasl::encode(" << size << "): " << encoded);
        return encoded;
    }
}