
#include "MediaF3Kafka.h"

namespace mfs
{
    static int _num = 0;
    class HashPartitionerCb : public RdKafka::PartitionerCb
    {
    public:
        int32_t partitioner_cb (const RdKafka::Topic *topic, const std::string *key,
                                int32_t partition_cnt, void *msg_opaque)
        {
            char msg[128] = {0};
            sprintf(msg, "HashPartitionerCb:[%s][%s][%d]", topic->name().c_str(),
                    key->c_str(), partition_cnt);
            std::cout << msg << std::endl;
            return generate_hash(key->c_str(), key->size()) % partition_cnt;
        }
    private:

        static inline unsigned int generate_hash(const char *str, size_t len)
        {
            unsigned int hash = 5381;
            for (size_t i = 0 ; i < len ; i++)
                hash = ((hash << 5) + hash) + str[i];
            return hash;
        }
    };


    class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb
    {
    public:
        void dr_cb(RdKafka::Message &message)
        {
            if (message.err())
                std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
            else
                std::cerr << "Message delivered to topic " << message.topic_name()
                        << " [" << message.partition() << "] at offset "
                        << message.offset() << std::endl;
        }
    };

    class ProducerEventCb : public RdKafka::EventCb
    {
    public:
        void event_cb(RdKafka::Event &event)
        {
            switch(event.type())
            {
            case RdKafka::Event::EVENT_ERROR:
                std::cout << "RdKafka::Event::EVENT_ERROR: " << RdKafka::err2str(event.err()) << std::endl;
                break;
            case RdKafka::Event::EVENT_STATS:
                std::cout << "RdKafka::Event::EVENT_STATS: " << event.str() << std::endl;
                break;
            case RdKafka::Event::EVENT_LOG:
                std::cout << "RdKafka::Event::EVENT_LOG " << event.fac() << std::endl;
                break;
            case RdKafka::Event::EVENT_THROTTLE:
                std::cout << "RdKafka::Event::EVENT_THROTTLE " << event.broker_name() << std::endl;
                break;
            }
        }
    };

    static std::map<std::string, std::shared_ptr<MediaF3Kafka>> _instances;
    static std::mutex _instancesMutex;

    std::shared_ptr<MediaF3Kafka> MediaF3Kafka::Instance(const std::string& clientid)
    {
        auto iter = _instances.find(clientid);
        if (iter == _instances.end())
        {
            std::lock_guard<std::mutex> gl(_instancesMutex);
            iter = _instances.find(clientid);
            if (iter == _instances.end())
            {
                auto instance = std::make_shared<MediaF3Kafka>();
                instance->init(AppSetting::Instance()->kafkaHost, "client_"+clientid+"_to_paas", 0);
                _instances.insert(std::pair<std::string, std::shared_ptr<MediaF3Kafka>>(clientid, instance));
                iter = _instances.find(clientid);
            }
        }
        if (iter == _instances.end()){
            LOG_ERROR << "MediaF3Kafka is NULL!" << LOG_ENDL;
            return nullptr;
        }
        return iter->second;
    }
    MediaF3Kafka::~MediaF3Kafka()
    {
        std::cout << "MediaF3Kafka::~MediaF3Kafka" << std::endl;
        delete m_topic;
        delete m_topicConfig;
        delete m_producer;
        delete m_dr_cb;
        delete m_event_cb;
        delete m_partitioner_cb;
        delete m_config;
    }
    
    int MediaF3Kafka::init(const std::string& brokers, const std::string& topic, int partition)
    {
        std::cout << "MediaF3Kafka::init" << std::endl;
        m_brokers = brokers;
        m_topicStr = topic;
        m_partition = partition;
        // 创建Kafka Conf对象
        m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        if(m_config == NULL)
        {
            std::cout << "Create RdKafka Conf failed." << std::endl;
        }
        // 创建Topic Conf对象
        m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
        if(m_topicConfig == NULL)
        {
            std::cout << "Create RdKafka Topic Conf failed." << std::endl;
        }
        // 设置Broker属性
        RdKafka::Conf::ConfResult errCode;
        m_dr_cb = new ProducerDeliveryReportCb;
        std::string errorStr;
        errCode = m_config->set("dr_cb", m_dr_cb, errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }
        m_event_cb = new ProducerEventCb;
        errCode = m_config->set("event_cb", m_event_cb, errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }

        m_partitioner_cb = new HashPartitionerCb;
        errCode = m_topicConfig->set("partitioner_cb", m_partitioner_cb, errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }

        errCode = m_config->set("statistics.interval.ms", "10000", errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }

        errCode = m_config->set("message.max.bytes", "10240000", errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }
        errCode = m_config->set("bootstrap.servers", m_brokers, errorStr);
        if(errCode != RdKafka::Conf::CONF_OK)
        {
            std::cout << "Conf set failed:" << errorStr << std::endl;
        }
        // 创建Producer
        m_producer = RdKafka::Producer::create(m_config, errorStr);
        if(m_producer == NULL)
        {
            std::cout << "Create Producer failed:" << errorStr << std::endl;
        }
        // 创建Topic对象
        m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, errorStr);
        if(m_topic == NULL)
        {
            std::cout << "Create Topic failed:" << errorStr << std::endl;
        }
        return 0;
    }


    void MediaF3Kafka::pushMessage(const std::string& str, const std::string& key)
    {    
        LOG_DEBUG << str << LOG_ENDL;
        int32_t len = str.length();
        void* payload = const_cast<void*>(static_cast<const void*>(str.data()));
        RdKafka::ErrorCode errorCode = m_producer->produce(m_topic, RdKafka::Topic::PARTITION_UA,
                                    RdKafka::Producer::RK_MSG_COPY,
                                    payload, len, &key, NULL);
        m_producer->poll(0);
        if (errorCode != RdKafka::ERR_NO_ERROR)
        {
            std::cerr << "Produce failed: " << RdKafka::err2str(errorCode) << std::endl;
            if(errorCode ==  RdKafka::ERR__QUEUE_FULL)
            {
                // m_producer->poll(0);
            }
        }
        else 
        {
            std::cout << "pushMessage Success" << std::endl;
        }
        if (m_producer->outq_len()<=0)
        {
            
        }
    }

    // 9101
    void MediaF3Kafka::PlayRealTime(const MessageBaseSetting& baseSetting, const std::string&  streamType, const std::string&  dataType, int port, const std::string& clientId, const std::string& monitorId)
    {
        std::stringstream ss;
        int mnum = _num++;
        ss<<"{\"data\":{\"channelNum\":"<<baseSetting.channelNum<<",\"serverIp\":\""<<AppSetting::Instance()->rtpIpSend
        <<"\",\"streamType\":"<<streamType<<",\"tcpPort\":"<< port <<",\"type\":"<<dataType
        <<",\"udpPort\":0},\"desc\":{\"monitorId\": "<<monitorId<<",\"sendType\":1,\"clientId\":\""<<clientId<<"\""<<",\"msgSn\":"<<mnum<<",\"messageType\":0,\"msgId\":"
        <<EF3CommandID::EF3CommandID_REALTIME<<",\"sysTime\":\""<<getNowStr()<<"\",\"type\":\"1\",\"deviceId\":\""
        << baseSetting.deviceID <<"\"}}";
        MessageInfo info  ={0};
        info.number = mnum;
        info.messgae = ss.str();
        info.answer = baseSetting.answer;

        if (MediaF3Kafka::Instance(clientId))
        {
            MediaF3Kafka::Instance(clientId)->pushMessage(info.messgae, "");
        }
    }
    // 9102
    void MediaF3Kafka::ControlRealTime(const MessageBaseSetting& baseSetting, const std::string& streamType, const std::string& closeVideoType,const std::string& control, const std::string& clientId, const std::string& monitorId)
    {
        std::stringstream ss;
        int mnum = _num++;
        ss<<"{\"data\":{\"channelNum\":"<<baseSetting.channelNum<<",\"changeStreamType\":"<<streamType<<",\"closeVideoType\":"<<closeVideoType
        <<",\"deviceType\":"<<"\"1\""<<",\"control\":"<<control<<"},\"desc\":{\"monitorId\": "<<monitorId<<",\"sendType\":1,\"clientId\":\""<<clientId<<"\""<<",\"msgSn\":"<<mnum<<",\"messageType\":0,\"msgId\":"
        <<EF3CommandID::EF3CommandID_REALTIME_C<<",\"sysTime\":\""<<getNowStr()<<"\",\"type\":\"0\",\"deviceId\":\""
        << baseSetting.deviceID <<"\"}}";

        MessageInfo info  ={0};
        info.number = mnum;
        info.messgae = ss.str();
        info.answer = baseSetting.answer;

        if (MediaF3Kafka::Instance(clientId))
        {
            MediaF3Kafka::Instance(clientId)->pushMessage(info.messgae, "");
        }
    }
    // 9201
    void MediaF3Kafka::PlayPlayback(const MessageBaseSetting& baseSetting, const std::string& endTime, const std::string& startTime, const std::string& forwardOrRewind, const std::string& remoteMode, const std::string& streamType, const std::string& storageType, int port, const std::string& clientId, const std::string& monitorId){
        std::stringstream ss;
        int mnum = _num++;
        auto st = atoi(streamType.c_str());
        st++;
        ss<<"{\"data\":{\"channelNum\":"<<baseSetting.channelNum<<",\"deviceType\":"<<"\"1\""<<",\"endTime\":\""<<endTime
        <<"\",\"startTime\":\""<<startTime<<"\",\"forwardOrRewind\":"<<forwardOrRewind<<",\"remoteMode\":"<<remoteMode<<",\"serverIp\":\""
        <<AppSetting::Instance()->rtpIpSend<<"\",\"storageType\":"<<storageType<<",\"streamType\":"<<std::to_string(st)<<",\"tcpPort\":"<< port
        <<",\"udpPort\":0,\"videoType\":0},\"desc\":{\"monitorId\": "<<monitorId<<",\"sendType\":1,\"clientId\":\""<<clientId<<"\""<<",\"msgSn\":"<<mnum<<",\"messageType\":0,\"msgId\":"
        <<EF3CommandID::EF3CommandID_PLAYBACK<<",\"sysTime\":\""<<getNowStr()<<"\",\"type\":\"0\",\"deviceId\":\""
        << baseSetting.deviceID <<"\"}}";

        MessageInfo info  ={0};
        info.number = mnum;
        info.messgae = ss.str();
        info.answer = baseSetting.answer;

        if (MediaF3Kafka::Instance(clientId))
        {
            MediaF3Kafka::Instance(clientId)->pushMessage(info.messgae, "");
        }
    }
    // 9202
    void MediaF3Kafka::ControlPlayback(const MessageBaseSetting& baseSetting, const std::string& dragPlaybackTime, const std::string& forwardOrRewind,const std::string& remote, const std::string& clientId, const std::string& monitorId)
    {
        std::stringstream ss;
        int mnum = _num++;
        ss<<"{\"data\":{\"channelNum\":"<<baseSetting.channelNum<<",\"dragPlaybackTime\":\""<<dragPlaybackTime
        <<"\",\"forwardOrRewind\":"<<forwardOrRewind<<",\"remote\":"<<remote<<"},\"desc\":{\"monitorId\": "
        <<monitorId<<",\"sendType\":1,\"clientId\":\""<<clientId<<"\""<<",\"msgSn\":"<<mnum<<",\"messageType\":0,\"msgId\":"
        <<EF3CommandID::EF3CommandID_PLAYBACK_C<<",\"sysTime\":\""<<getNowStr()<<"\",\"type\":\"0\",\"deviceId\":\""
        << baseSetting.deviceID <<"\"}}";

        MessageInfo info  ={0};
        info.number = mnum;
        info.messgae = ss.str();
        info.answer = baseSetting.answer;

        if (MediaF3Kafka::Instance(clientId))
        {
            MediaF3Kafka::Instance(clientId)->pushMessage(info.messgae, "");
        }
    }

}