package com.huazheng.project.greenplum.source.mysql;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Arrays;
import java.util.List;


import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import com.huazheng.project.HZDataStream;

import cn.hutool.core.thread.ThreadUtil;

import lombok.extern.log4j.Log4j2;

@Log4j2
@Service
public class TransformNewNodeSource implements SourceFunction<String> {

	private static final long serialVersionUID = 1L;
	
	
	
	public String getErrorInfoFromException(Exception e) {
		try {
			StringWriter sw = new StringWriter();
			PrintWriter pw = new PrintWriter(sw);
			e.printStackTrace(pw);
			return "\r\n" + sw.toString() + "\r\n";
		} catch (Exception e2) {
			return "bad getErrorInfoFromException";
		}
	}
	
	@Override
	public void run(SourceContext<String> ctx) throws Exception {
		DefaultRedisScript<String> script = new DefaultRedisScript<String>();
		script.setResultType(String.class);
		script.setScriptSource(new ResourceScriptSource(new ClassPathResource("luascript/vbap.lua")));
		List<String> keys = Arrays.asList("huazheng:TransformNewNode:sendcount", "huazheng:TransformNewNode:id", "huazheng:list:TransformNewNode", "huazheng:TransformNewNode:receivecount");
		while (true) {
			try {
				String value = HZDataStream.redis1Template.execute(script, keys, "");
				String[] values = value.toString().split("==========");
				String checkString = values[0];
				String[] split = checkString.split(", ");
				boolean check = split[0].split(":")[1].equals(split[1].split(":")[1]);
				if (values.length > 1) { // 有数据字符串
//					log.info(msg + " " + value.toString() + " " + check);
					log.info(checkString + " " + check);
					ctx.collect(values[1]);
				} else { // 没有数据字符串
					ThreadUtil.sleep(1000); // 没有数据了，休眠一下
				}
			} catch (Exception e) {
				HZDataStream.redis1Template.opsForHash().put("huazheng:TransformNewNode:error", "receivecount_elseerror", getErrorInfoFromException(e));
			}
		}
	}

	@Override
	public void cancel() {
		
	}
	
}
