package com.citygis.web.controller;

import com.citygis.web.service.KafkaConsumerServer;
import com.citygis.web.service.KafkaMonitorService;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Map;


@CrossOrigin
@RestController
@RequestMapping("/kafkaMonitor")
public class KafkaMonitorController {

    @Resource
    KafkaMonitorService kafkaMonitorService;

    @Resource
    KafkaConsumerServer kafkaCousumerServer;

    /**
     * 获取partition大小
     *
     * @param map
     * @return
     */
    @RequestMapping("/getPartitionSizeValue")
    public String getPartitionSizeValue(@RequestBody Map<String, Object> map) {
        return kafkaMonitorService.getPartitionSizeValue();
    }

    /**
     * 获取集群连接数量
     *
     * @return
     */
    @RequestMapping("/getActiveControllerCount")
    public String getActiveControllerCount() {
        return kafkaMonitorService.getActiveControllerCount();
    }

    /**
     * 关闭连接数
     *
     * @param map
     * @return
     */
    @RequestMapping("/getConnectionCloseRate")
    public String getConnectionCloseRate(@RequestBody Map<String, Object> map) {
        return kafkaMonitorService.getConnectionCloseRate(map.get("clientId").toString());
    }

    /**
     * 磁盘使用量
     *
     * @return
     */
    @RequestMapping("/getDiskUsage")
    public String getDiskUsage() {
        return kafkaMonitorService.getDiskUsage();
    }

    /**
     * cpu使用量
     *
     * @return
     */
    @RequestMapping("/getCpuUsage")
    public String getCpuUsage() {
        return kafkaMonitorService.getCpuUsage();
    }

    /**
     * 缺失副本数
     *
     * @return
     */
    @RequestMapping("/getUnderReplicated")
    public String getUnderReplicated(@RequestBody Map<String, Object> map) {
        return kafkaMonitorService.getUnderReplicated();
    }

    /**
     * 流出流量 消费
     *
     * @return
     */
    @RequestMapping("/getBytesOutPerSe")
    public String getBytesOutPerSe() throws Exception {
        return kafkaMonitorService.getBytesOutPerSe();
    }

    /**
     * 流入流量 生产
     *
     * @return
     */
    @RequestMapping("/getBytesInPerSec")
    public String getBytesInPerSec() throws Exception {
        return kafkaMonitorService.getBytesInPerSec();
    }

    /**
     * 消息 消费
     *
     * @return
     */
    @RequestMapping("/getConsumerMessage")
    public String getConsumerMessage() throws Exception {
        return kafkaMonitorService.getConsumerMessage();
    }

    /**
     * 生产限流
     *
     * @throws Exception
     */
    @RequestMapping("/getProducerThrottledRate")
    public void getProducerThrottledRate() throws Exception {
        kafkaMonitorService.getProducerThrottledRate();
    }

    /**
     * 消费限流
     *
     * @throws Exception
     */
    @RequestMapping("/getConsumerThrottledRate")
    public void getConsumerThrottledRate() throws Exception {
        kafkaMonitorService.getConsumerThrottledRate();
    }

    /**
     * 每个topic每个未消息的量
     *
     * @throws Exception
     */
    @RequestMapping("/getAllLagMessage")
    public void getAllLagMessage() throws Exception {
        kafkaCousumerServer.getAllLagMessage();
    }


}
