WisdomAdapter/src/main/java/com/gyee/wisdom/adapter/dao/TaosLatestDao.java

package com.gyee.wisdom.adapter.dao;

import com.gyee.wisdom.adapter.common.config.TaosConfiguration;
import com.gyee.wisdom.adapter.common.config.ThreadPoolTaskConfig;
import com.gyee.wisdom.adapter.common.exception.WisdomException;
import com.gyee.wisdom.adapter.common.util.TaosCovertUtil;
import com.gyee.wisdom.adapter.model.adapter.*;
import com.gyee.wisdom.adapter.timeseries.dao.ILatestDao;
import com.gyee.wisdom.adapter.timeseries.TaosDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.Statement;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Slf4j
@Component
@TaosDao
public class TaosLatestDao implements ILatestDao {

    @Autowired
    private TaosConfiguration config;
    @Autowired
    private ThreadPoolTaskConfig taskConfig;


    /**
     * 查询多测点最新数据
     * @param tsPoints
     * @return
     * @throws Exception
     */
    @Override
    public Map<String, TsData> getTsDataLatest(List<TsPoint> tsPoints) throws Exception {
        Map<String, TsData> result = new HashMap<>();
        Map<TsDataType, List<TsPoint>> pointGroup = tsPoints.stream().collect(Collectors.groupingBy(TsPoint::getTsDataType));
        for (Map.Entry<TsDataType, List<TsPoint>> entry : pointGroup.entrySet()) {
            String[] tagNames = entry.getValue().stream().map(TsPoint::getId).toArray(String[]::new);
            if (entry.getKey() == TsDataType.DOUBLE) {
                result.putAll(getDoubleTsDataSnapshots(tagNames));
            } else if (entry.getKey() == TsDataType.LONG) {
                result.putAll(getLongTsDataSnapshots(tagNames));
            } else if (entry.getKey() == TsDataType.BOOLEAN) {
                result.putAll(getBooleanTsDataSnapshots(tagNames));
            } else {
                throw new WisdomException("Taos不支持数据类型:" + entry.getKey());
            }
        }

        return result;
    }


    @Override
    public int writeDoubleLatest(List<TsPointData> list) throws Exception {
        boolean flag = false;
        Connection connection = config.getConnect();

        try {
            Statement st = connection.createStatement();

            StringBuilder sb = new StringBuilder();
            sb.append("insert into ");
            for (TsPointData obj : list) {
                long time = obj.getTsData().getTs();
                String point = obj.getTagName();
                double value = obj.getTsData().getDoubleValue().get();
                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
                sb.append(point).append(" values (");
                sb.append(time).append(",").append(value).append(")");
            }

            flag = st.execute(sb.toString());

        } catch (Exception e) {
        } finally {
            if (connection != null)
                connection.close();
        }

        return flag == true ? list.size() : 0;
    }

    @Override
    public int writeStringLatest(List<TsPointData> list) throws Exception {
        boolean flag = false;
        Connection connection = config.getConnect();

        try {
            Statement st = connection.createStatement();

            StringBuilder sb = new StringBuilder();
            sb.append("insert into ");
            for (TsPointData obj : list) {
                long time = obj.getTsData().getTs();
                String point = obj.getTagName();
                String value = obj.getTsData().getStringValue().get();
                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
                sb.append(point).append(" values (");
                sb.append(time).append(",'").append(value).append("') ");
            }

            flag = st.execute(sb.toString());

        } catch (Exception e) {
        } finally {
            if (connection != null)
                connection.close();
        }

        return flag == true ? list.size() : 0;
    }

    @Override
    public int writeBooleanLatest(List<TsPointData> list) throws Exception {
        boolean flag = false;
        Connection connection = config.getConnect();

        try {
            Statement st = connection.createStatement();

            StringBuilder sb = new StringBuilder();
            sb.append("insert into ");
            for (TsPointData obj : list) {
                long time = obj.getTsData().getTs();
                String point = obj.getTagName();
                boolean value = obj.getTsData().getBooleanValue().get();
                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
                sb.append(point).append(" values (");
                sb.append(time).append(",").append(value).append(")");
            }

            flag = st.execute(sb.toString());

        } catch (Exception e) {
        } finally {
            if (connection != null)
                connection.close();
        }

        return flag == true ? list.size() : 0;
    }

    @Override
    public int writeLongLatest(List<TsPointData> list) throws Exception {
        boolean flag = false;
        Connection connection = config.getConnect();

        try {
            Statement st = connection.createStatement();

            StringBuilder sb = new StringBuilder();
            sb.append("insert into ");
            for (TsPointData obj : list) {
                long time = obj.getTsData().getTs();
                String point = obj.getTagName();
                long value = obj.getTsData().getLongValue().get();
                sb.append(TaosCovertUtil.coverStationPrefix(point)).append(".");
                sb.append(point).append(" values (");
                sb.append(time).append(",").append(value).append(")");
            }

            flag = st.execute(sb.toString());

        } catch (Exception e) {
        } finally {
            if (connection != null)
                connection.close();
        }

        return flag == true ? list.size() : 0;
    }


    /**
     * 由于taos数据库中全部是DOUBLE类型，故不需要区分类型
     *
     * @param dataList
     * @return
     * @throws Exception
     */
    @Override
    public boolean writeLatest(List<TsPointData> dataList) throws Exception {
        int writeCount = 0;

        Map<TsDataType, List<TsPointData>> pointGroup = dataList.stream().collect(Collectors.groupingBy(TsPointData::findDataType));
        for (Map.Entry<TsDataType, List<TsPointData>> entry : pointGroup.entrySet()) {
            List<TsPointData> pointDataList = entry.getValue();
//            if (entry.getKey() == TsDataType.DOUBLE) {
            int count = writeDoubleLatest(pointDataList);
            writeCount = writeCount + count;
//            } else if (entry.getKey() == TsDataType.BOOLEAN) {
//                int count = writeBooleanLatest(pointDataList);
//                writeCount = writeCount + count;
//            } else if (entry.getKey() == TsDataType.STRING) {
//                int count = writeStringLatest(pointDataList);
//                writeCount = writeCount + count;
//            } else if (entry.getKey() == TsDataType.BLOB) {
//                int count = writeBlobLatest(pointDataList);
//                writeCount = writeCount + count;
//            } else if (entry.getKey() == TsDataType.COORDINATE) {
//                int count = writeCoordinateLatest(pointDataList);
//                writeCount = writeCount + count;
//            } else if (entry.getKey() == TsDataType.LONG) {
//                int count = writeLongLatest(pointDataList);
//                writeCount = writeCount + count;
//            }
        }
        return writeCount > 0 ? true : false;
    }


    public Map<String, TsData> getDoubleTsDataSnapshots(String... tagNames) throws Exception {
        Map<String, TsData> tsDataMap = new HashMap<>();

        //存储线程的返回值
        List<Future<Map<String, TsData>>> results = new LinkedList<>();

        for (String tag : tagNames) {
            String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
            String sql = "select last_row(*) from " + point;
            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.DOUBLE);
            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
            results.add(submit);
        }

        //返回结果
        for (int i = 0; i < results.size(); i++) {
            tsDataMap.putAll(results.get(i).get());
        }

        return tsDataMap;
    }

    public Map<String, TsData> getLongTsDataSnapshots(String... tagNames) throws Exception {
        Map<String, TsData> tsDataMap = new HashMap<>();

        //存储线程的返回值
        List<Future<Map<String, TsData>>> results = new LinkedList<>();

        for (String tag : tagNames) {
            String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
            String sql = "select last_row(*) from " + point;
            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.LONG);
            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
            results.add(submit);
        }

        //返回结果
        for (int i = 0; i < results.size(); i++) {
            tsDataMap.putAll(results.get(i).get());
        }

        return tsDataMap;
    }


    public Map<String, TsData> getBooleanTsDataSnapshots(String... tagNames) throws Exception {
        Map<String, TsData> tsDataMap = new HashMap<>();

        //存储线程的返回值
        List<Future<Map<String, TsData>>> results = new LinkedList<>();

        for (String tag : tagNames) {
            String point = TaosCovertUtil.coverStationPrefix(tag) + "." + tag.replace(".", "_");
            String sql = "select last_row(*) from " + point;
            TaskCallable task = new TaskCallable(config.getConnect(), sql, 0, tag, TsDataType.BOOLEAN);
            Future<Map<String, TsData>> submit = taskConfig.getExecutor().submit(task);
            results.add(submit);
        }

        //返回结果
        for (int i = 0; i < results.size(); i++) {
            tsDataMap.putAll(results.get(i).get());
        }

        return tsDataMap;
    }

}