//
// Copyright 2022 DMetaSoul
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package com.dmetasoul.metaspore.dataservice;

import com.dmetasoul.metaspore.annotation.FeatureAnnotation;
import com.dmetasoul.metaspore.common.CommonUtils;
import com.dmetasoul.metaspore.configure.SourceTable;
import com.dmetasoul.metaspore.data.DataContext;
import com.dmetasoul.metaspore.data.DataResult;
import com.dmetasoul.metaspore.data.ServiceRequest;
import com.dmetasoul.metaspore.datasource.DataSource;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * SourceTable的DataService的实现类
 * SourceTable用于调用source数据源信息请求相关数据，sourcetable配置定义了数据的schema和部分条件
 * 注解DataServiceAnnotation 必须设置， value应设置为SourceTable。
 * Created by @author qinyy907 in 14:24 22/08/01.
 */

@Slf4j
@FeatureAnnotation("SourceTable")
public class SourceTableTask extends DataService {
    public static final int DEFAULT_MAX_LIMIT = 200;

    protected int maxLimit;
    private DataSource dataSource;
    protected SourceTable sourceTable;

    @Override
    public boolean initService() {
        sourceTable = (SourceTable) tableConfig;
        this.maxLimit = CommonUtils.getField(sourceTable.getOptions(), "max_limit", DEFAULT_MAX_LIMIT, Integer.class);
        dataSource = serviceManager.getDataSource(sourceTable.getSource());
        for (String col : sourceTable.getColumnNames()) {
            resFields.add(sourceTable.getFieldMap().get(col));
            dataTypes.add(sourceTable.getColumnMap().get(col));
        }
        return true;
    }

    protected List<Map<String, Object>> processRequest(ServiceRequest request, DataContext context) {
        return dataSource.process(request, context);
    }

    public <T> T getOptionOrDefault(String key, T value) {
        return CommonUtils.getField(sourceTable.getOptions(), key, value);
    }

    @Override
    public DataResult process(ServiceRequest request, DataContext context) {
        DataResult dataResult;
        int retryNum = getOptionOrDefault("retryNum", 0);
        long timeOut = getOptionOrDefault("timeOut", 30000L);
        TimeUnit timeOutUnit = TimeUnit.MILLISECONDS;
        retryNum += 1;
        do {
            CompletableFuture<DataResult> future = CompletableFuture.supplyAsync(() -> {
                List<Map<String, Object>> res = processRequest(request, context);
                DataResult result = setDataResult(res);
                if (checkResult(result)) {
                    return result;
                } else {
                    return null;
                }
            }, dataSource.getFeaturePool()).exceptionally(error -> {
                log.error("exec fail at {}, exception:{}!", name, error);
                return null;
            });
            try {
                dataResult = future.get(timeOut, timeOutUnit);
                if (checkResult(dataResult)) {
                    return dataResult;
                }
                retryNum -= 1;
            } catch (InterruptedException | ExecutionException e) {
                log.error("there was an error when executing the CompletableFuture", e);
            } catch (TimeoutException e) {
                log.error("when task timeout!", e);
            }
        } while (retryNum >= 0);
        return null;
    }
}
