BrokerConsumerStatsImpl.cc

/**
 * 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 <lib/BrokerConsumerStatsImpl.h>
#include <boost/date_time/local_time/local_time.hpp>

namespace pulsar {
BrokerConsumerStatsImpl::BrokerConsumerStatsImpl()
    : validTill_(boost::posix_time::microsec_clock::universal_time()){};

BrokerConsumerStatsImpl::BrokerConsumerStatsImpl(double msgRateOut, double msgThroughputOut,
                                                 double msgRateRedeliver, std::string consumerName,
                                                 uint64_t availablePermits, uint64_t unackedMessages,
                                                 bool blockedConsumerOnUnackedMsgs, std::string address,
                                                 std::string connectedSince, const std::string& type,
                                                 double msgRateExpired, uint64_t msgBacklog)
    : msgRateOut_(msgRateOut),
      msgThroughputOut_(msgThroughputOut),
      msgRateRedeliver_(msgRateRedeliver),
      consumerName_(consumerName),
      availablePermits_(availablePermits),
      unackedMessages_(unackedMessages),
      blockedConsumerOnUnackedMsgs_(blockedConsumerOnUnackedMsgs),
      address_(address),
      connectedSince_(connectedSince),
      type_(convertStringToConsumerType(type)),
      msgRateExpired_(msgRateExpired),
      msgBacklog_(msgBacklog) {}

bool BrokerConsumerStatsImpl::isValid() const {
    return boost::posix_time::microsec_clock::universal_time() <= validTill_;
}

std::ostream& operator<<(std::ostream& os, const BrokerConsumerStatsImpl& obj) {
    os << "\nBrokerConsumerStatsImpl ["
       << "validTill_ = " << obj.isValid() << ", msgRateOut_ = " << obj.getMsgRateOut()
       << ", msgThroughputOut_ = " << obj.getMsgThroughputOut()
       << ", msgRateRedeliver_ = " << obj.getMsgRateRedeliver()
       << ", consumerName_ = " << obj.getConsumerName()
       << ", availablePermits_ = " << obj.getAvailablePermits()
       << ", unackedMessages_ = " << obj.getUnackedMessages()
       << ", blockedConsumerOnUnackedMsgs_ = " << obj.isBlockedConsumerOnUnackedMsgs()
       << ", address_ = " << obj.getAddress() << ", connectedSince_ = " << obj.getConnectedSince()
       << ", type_ = " << obj.getType() << ", msgRateExpired_ = " << obj.getMsgRateExpired()
       << ", msgBacklog_ = " << obj.getMsgBacklog() << "]";
    return os;
}

double BrokerConsumerStatsImpl::getMsgRateOut() const { return msgRateOut_; }

double BrokerConsumerStatsImpl::getMsgThroughputOut() const { return msgThroughputOut_; }

double BrokerConsumerStatsImpl::getMsgRateRedeliver() const { return msgRateRedeliver_; }

const std::string BrokerConsumerStatsImpl::getConsumerName() const { return consumerName_; }

uint64_t BrokerConsumerStatsImpl::getAvailablePermits() const { return availablePermits_; }

uint64_t BrokerConsumerStatsImpl::getUnackedMessages() const { return unackedMessages_; }

bool BrokerConsumerStatsImpl::isBlockedConsumerOnUnackedMsgs() const { return blockedConsumerOnUnackedMsgs_; }

const std::string BrokerConsumerStatsImpl::getAddress() const { return address_; }

const std::string BrokerConsumerStatsImpl::getConnectedSince() const { return connectedSince_; }

const ConsumerType BrokerConsumerStatsImpl::getType() const { return type_; }

double BrokerConsumerStatsImpl::getMsgRateExpired() const { return msgRateExpired_; }

uint64_t BrokerConsumerStatsImpl::getMsgBacklog() const { return msgBacklog_; }

void BrokerConsumerStatsImpl::setCacheTime(uint64_t cacehTimeInMs) {
    validTill_ =
        boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(cacehTimeInMs);
}

ConsumerType BrokerConsumerStatsImpl::convertStringToConsumerType(const std::string& str) {
    if (str == "ConsumerFailover" || str == "Failover") {
        return ConsumerFailover;
    } else if (str == "ConsumerShared" || str == "Shared") {
        return ConsumerShared;
    } else if (str == "ConsumerKeyShared" || str == "KeyShared") {
        return ConsumerKeyShared;
    } else {
        return ConsumerExclusive;
    }
}
}  // namespace pulsar