package com.sunwin.metro.utils;

import io.prometheus.client.exporter.PushGateway;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @program: test_prometheus
 * @description: 删除pushaway节点
 * @author: xuYJ
 * @create: 2022-01-25 09:30
 **/
public class DeletePush extends KeyedProcessFunction<String, Map<String, List<String>>, String> {
    private transient MapState<String, String> mapState;
    private PushGateway gateway;
    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<String, String> descriptor = new MapStateDescriptor<>("mapState", TypeInformation.of(String.class), TypeInformation.of(String.class));
        mapState = getRuntimeContext().getMapState(descriptor);
        gateway = new PushGateway("192.168.20.65:9091");
        System.out.println("生成 gateway 客户端 ");
        super.open(parameters);
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void processElement(Map<String, List<String>> value, Context ctx, Collector<String> out) throws Exception {
        ArrayList<String> list = new ArrayList<>();
        mapState.keys().forEach(new Consumer<String>() {
            @Override
            public void accept(String s) {
                list.add(s);
            }
        });
        List<String> keyWords = value.get("keyWords");
        for (String keywords : value.get("keyWords")) {
            if(!list.contains(keywords)){
                mapState.put(keywords,"1");
            }
        }
        List<String> collect = list.stream().filter(item -> !keyWords.contains(item)).collect(Collectors.toList());
        for (String s : collect) {
            String[] result = s.split("_");
            gateway.delete(result[0] + "_" + result[1] + "站台_" + result[2] + "设备监控数据");
            mapState.remove(s);
            out.collect("删除job ：" + result[0] + "_" + result[1] + "站台_" + result[2] + "设备监控数据");
        }
    }
}
