package com.yeejoin.amos.api.alarm.service.impl;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * @description: 监听设备告警信息
 * @author: tw
 * @createDate: 2023/6/27
 */

@Service
public class AlarmKafkaConsumer {

	@Autowired
	PointSystemServiceImpl pointSystemServiceImpl;
	
	@Value("${warning.wait.time:1000}")
	private long waitTime;

	// 消费者来处理消息
	@KafkaListener(id = "alarmInfo", topics = { "${kafka.equipment.alarm}" })
	public void message1(String record, Acknowledgment ack) throws InterruptedException {
		// 处理业务
		String date = record;
		
		// 异步触发预警
		pointSystemServiceImpl.sendWarningAsync(date);
		
		// 等待1s
		Thread.sleep(waitTime);
		
		// 手动提交
		ack.acknowledge();
	}

	@KafkaListener(id = "alarmInfo2", topics = { "${kafka.equipment.test}" })
	public void message2(String record, Acknowledgment ack) {
		String date = record;
		System.out.println("收到告警信息" + date);
	}

//    public void message1( ConsumerRecord<?, ?> record, Acknowledgment ack){
//        // 消费的哪个topic、partition的消息,打印出消息内容
//
//        StringBuffer sb = new StringBuffer();
//        // 主题
//        sb.append(record.topic() + "-");
//        // 分区
//        sb.append(record.partition() + "-");
//        // 需要消费的值
//        sb.append(record.value() + "-");
//        // 位移
//        sb.append(record.offset());
//
//        System.out.println( "消费者进行消费："+ sb);
//        ack.acknowledge();
//
//    }
//    // 简单消费者，groupId可以任意起
//    @KafkaListener(id = "Consumer0", groupId = "jf0-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"0"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer0(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }
//
//    @KafkaListener(id = "Consumer1", groupId = "jf1-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"1"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer1(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }
//
//    @KafkaListener(id = "Consumer2", groupId = "jf2-group", topics = "jf1", topicPartitions = {
//            @TopicPartition(topic = "jf1", partitions = {"2"}),
//    }, containerFactory = "kafkaListenerContainerFactory")
//    public void consumer3(ConsumerRecord<String, String> records, Acknowledgment ack) {
//        this.message1(records,ack);
//    }

}
