package com.tsi.comm.server.process.dbms;

import com.tsi.app.common.app.AppUtils;
import com.tsi.app.common.utils.Elapsed;
import com.tsi.app.common.utils.TimeUtils;
import com.tsi.comm.server.mybatis.TsiDatabaseMapper;
import com.tsi.comm.server.mybatis.vo.*;
import com.tsi.comm.server.process.AbstractTsiCvimWorker;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
@Getter
public class TsiCvimDbmsWorker extends AbstractTsiCvimWorker implements Runnable {

    //private ConcurrentLinkedQueue<AbstractTsiPacket> DATA_QUEUE;
    private final LinkedBlockingQueue<Object> DATA_QUEUE;
    private final TsiDatabaseMapper tsiDatabaseMapper;

    public TsiCvimDbmsWorker(int idx, int qSize) {
        this.idx = idx;
        this.qSize = qSize;
        this.DATA_QUEUE = new LinkedBlockingQueue<>(qSize);
        this.tsiDatabaseMapper = (TsiDatabaseMapper) AppUtils.getBean(TsiDatabaseMapper.class);
    }

    @Override
    public void run() {
        log.info("DbmsWorker({}): {} Start. QSIZE: {}", this.idx, Thread.currentThread().getName(), this.qSize);
        try {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    Object object = this.DATA_QUEUE.poll(1, TimeUnit.SECONDS); // 인터럽트 감지 가능
                    if (object != null) {
                        process(object);
                    } else {
                        Thread.yield(); // 대기 중 CPU 양보
                    }
                } catch (InterruptedException ie) {
                    log.warn("DbmsWorker({}): {} Interrupted. Exiting...", this.idx, Thread.currentThread().getName());
                    Thread.currentThread().interrupt(); // 상태 복구
                    break;
                } catch (Exception e) {
                    log.error("DbmsWorker({}): {} Exception: {}", this.idx, Thread.currentThread().getName(), e.getMessage());
                }
            }
        } finally {
            log.info("DbmsWorker({}): {} Stopped.", this.idx, Thread.currentThread().getName());
        }
    }

//    @Override
//    public void run() {
//        log.info("{} Start. QSIZE: {}", Thread.currentThread().getName(), this.qSize);
//        while (true) {
//            try {
//                Object object = this.DATA_QUEUE.take();
//                if (object != null) {
//                    process(object);
//                }
//                else {
//                    Thread.yield();
//                }
//            }
//            catch (Exception e) {
//                log.error("Exception(DbmsWorker while): {}", e.getMessage());
//            }
//        }
//    }

    /*
     *  작업큐에 데이터 추가
     */
    public boolean add(Object object) {
        boolean offer = false;
        try {
            offer = this.DATA_QUEUE.offer(object);  //offer => full -> return, add => full -> wait
            if (!offer) {
                log.error("DbmsWorker({}): Queue.offer(Full): {}/{}/{}, {}, {}",
                        this.idx, this.DATA_QUEUE.size(), this.qSize, this.DATA_QUEUE.remainingCapacity(),
                        object, Thread.currentThread().getName());
            }
        } catch (Exception e) {
            log.error("DbmsWorker({}): Queue.offer: Exception: {}, {}, {}", this.idx, object.toString(), Thread.currentThread().getName(), e.getMessage());
        }
        return offer;
    }

    public void process(Object object) {

        Elapsed elapsed = new Elapsed();
        String job = "Unknown";
        int count = 0;

        if (object instanceof AbstractDbmsVo) {
            AbstractDbmsVo dbmsVo = (AbstractDbmsVo)object;
            try {
                if (dbmsVo.getDbmsType() == AbstractDbmsVo.DBMS_TPMS_STAT_1S) {
                    job = "CommTpHs";
                    count = 1;
                    this.tsiDatabaseMapper.insertCommTpsHs((CommStatusVo) dbmsVo);
                }
                else if (dbmsVo.getDbmsType() == AbstractDbmsVo.DBMS_KAFKA_TRANS_HS) {
                    job = "KafkaTransHs";
                    count = 1;
                    this.tsiDatabaseMapper.insertKafkaTransHs((KafkaTransVo) dbmsVo);
                }
                else if (dbmsVo.getDbmsType() == AbstractDbmsVo.DBMS_NODE_STATUS) {
                    job = "NodeStatus";
                    count = 1;
                    this.tsiDatabaseMapper.updateNodeStatus((NodeStatusVo)object);
                }
                else if (dbmsVo.getDbmsType() == AbstractDbmsVo.DBMS_ALARM_OCCR_HS) {
                    job = "AlarmOccr";
                    count = 1;
                    this.tsiDatabaseMapper.insertAlarmOccrHs((AlarmOccrVo)object);
                }
                else if (dbmsVo.getDbmsType() == AbstractDbmsVo.DBMS_NODE_IP_UPDATE) {
                    job = "NodeIpAddr";
                    count = 1;
                    this.tsiDatabaseMapper.updateNodeIpAddr((NodeIpAddrVo)object);
                }
            }
            catch (Exception e) {
                log.error("DbmsWorker.process: {}, {}, Exception: {}", job, dbmsVo, e.getMessage());
            }
            calcProcessTime(dbmsVo.getCreateTime());
        }
        else {
            log.error("DbmsWorker.process: Unknown Object Class: {}", object.getClass().getSimpleName());
        }
        if (elapsed.milliSeconds() > 500) {
            // 로그가 너무 많아 일정 시간 이상 걸리는 경우만 로그로 남긴다.
            log.warn("{}: {} EA. {} {}", job, count, TimeUtils.elapsedTimeStr(elapsed.nanoSeconds()), Thread.currentThread().getName());
        }
    }

    public void report() {
        log.info("DbmsWorker({}), Queue Total/Size/Remain: {}/{}/{}, Average: {}, {}",
                this.idx,
                this.qSize, this.DATA_QUEUE.size(), this.DATA_QUEUE.remainingCapacity(),
                TimeUtils.elapsedTimeStr(this.avgTime),
                Thread.currentThread().getName());
    }

}
