package com.sunwin.metro.process;

import com.alibaba.fastjson.JSONObject;
import com.sunwin.metro.utils.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import java.util.List;
import java.util.Map;

/**
 * @program: boryou-flink
 * @description: 处理不同关键词匹配
 * @author: xuYJ
 * @create: 2020-10-19 16:23
 **/
public class BroadcastMeterMes extends BroadcastProcessFunction<String, Map<String, List<String>>, Tuple2<String,String>> {

    private List<String> keyWords;

    @Override
    public void processElement(String value, ReadOnlyContext ctx, Collector<Tuple2<String,String>> collector) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        try {
                if (!StringUtils.isNullOrEmpty(keyWords)) {
                    for (String tuple2 : keyWords) {
                        String[] split = tuple2.split("_");
                        String num = split[0];
                        String station = split[1].trim();
                        String name = split[2].trim();
                        String appType = split[3].trim();
                        if(jsonObject.containsKey(name.split("-")[0].trim())){
                            JSONObject json = JSONObject.parseObject(jsonObject.getString(name.split("-")[0].trim()));
                            json.put("num",num);
                            json.put("station",station);
                            json.put("name",name);
                            json.put("appType",appType);
                            collector.collect(new Tuple2<>((num+"_"+station+"_"+name+"_"+appType),json.toString()));
                        }else {
                            System.out.println("没有该设备数据：" + tuple2);
                        }
                    }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("解析该设备数据失败");
        }
    }

    @Override
    public void processBroadcastElement(Map<String, List<String>> map, Context ctx, Collector<Tuple2<String,String>> out) throws Exception {
        try {
            if (!StringUtils.isNullOrEmpty(map)) {
                List<String> keyWords = map.get("keyWords");
                if (!StringUtils.isNullOrEmpty(keyWords)) {
                    this.keyWords = keyWords;
                } else {
                    System.out.println("查询数据库为空");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
