package org.frameworkset.elasticsearch.imp;
/**
 * Copyright 2008 biaoping.yin
 * <p>
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.
 */

import org.frameworkset.elasticsearch.ElasticSearchHelper;
import org.frameworkset.tran.ExportResultHandler;
import org.frameworkset.tran.config.ImportBuilder;
import org.frameworkset.tran.metrics.TaskMetrics;
import org.frameworkset.tran.plugin.db.input.DBInputConfig;
import org.frameworkset.tran.plugin.es.output.ElasticsearchOutputConfig;
import org.frameworkset.tran.schedule.ExternalScheduler;
import org.frameworkset.tran.schedule.ImportIncreamentConfig;
import org.frameworkset.tran.schedule.quartz.AbstractQuartzJobHandler;
import org.frameworkset.tran.task.TaskCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>Description: 使用quartz等外部环境定时运行导入数据，需要调试测试quatz作业同步功能，按如下配置进行操作：</p>
 *  * 1.在配置文件中添加quartz作业配置-resources/org/frameworkset/task/quarts-task.xml相关内容
 *  * <list>
 *  *
 *  * 			<property name="QuartzImportTask" jobid="QuartzImportTask"
 *  * 							  bean-name="QuartzImportTask"
 *  * 							  method="execute"
 *  * 							  cronb_time="${quartzImportTask.crontime:*\/20 * * * * ?}" used="false"
 *  * 							  shouldRecover="false"
 *  * 					/>
 *  *
 *  * </list>
 *  *
 *  * 	<property name="QuartzImportTask" class="org.frameworkset.elasticsearch.imp.QuartzImportTask"
 *  * 			  destroy-method="destroy"
 *  * 			  init-method="init"
 *  * 	/>
 *  *
 *  * 2.添加一个带main方法的作业运行
 *  * public class QuartzTest {
 *  * 	public static void main(String[] args){
 *  * 		TaskService.getTaskService().startService();
 *  *        }
 *  * }
 *  * 然后运行main方法即可
 * <p></p>
 * <p>Copyright (c) 2018</p>
 * @Date 2019/4/13 13:45
 * @author biaoping.yin
 * @version 1.0
 */
public class QuartzImportTask extends AbstractQuartzJobHandler {
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	public void init(){
		externalScheduler = new ExternalScheduler();
		externalScheduler.dataStream((Object params)->{
			ImportBuilder importBuilder = new ImportBuilder() ;
			//增量定时任务不要删表，但是可以通过删表来做初始化操作
//		if(dropIndice) {
			try {
				//清除测试表,导入的时候回重建表，测试的时候加上为了看测试效果，实际线上环境不要删表
				String repsonse = ElasticSearchHelper.getRestClientUtil().dropIndice("quartz");
				System.out.println(repsonse);
			} catch (Exception e) {
			}
//		}


			//指定导入数据的sql语句，必填项，可以设置自己的提取逻辑，
			// 设置增量变量log_id，增量变量名称#[log_id]可以多次出现在sql语句的不同位置中，例如：
			// select * from td_sm_log where log_id > #[log_id] and parent_id = #[log_id]
			// log_id和数据库对应的字段一致,就不需要设置setLastValueColumn信息，
			// 但是需要设置setLastValueType告诉工具增量字段的类型
			importBuilder.setExportResultHandler(new ExportResultHandler() {
				@Override
				public void success(TaskCommand taskCommand, Object result) {
					System.out.println("success");
					TaskMetrics taskMetrics = taskCommand.getTaskMetrics();
					logger.info(taskMetrics.toString());
				}

				@Override
				public void error(TaskCommand taskCommand, Object result) {
					System.out.println("error");
					TaskMetrics taskMetrics = taskCommand.getTaskMetrics();
					logger.info(taskMetrics.toString());
				}

				@Override
				public void exception(TaskCommand taskCommand, Throwable exception) {
					System.out.println("exception");
					TaskMetrics taskMetrics = taskCommand.getTaskMetrics();
					logger.info(taskMetrics.toString());
				}


			});
			DBInputConfig dbInputConfig = new DBInputConfig();
			dbInputConfig.setSql("select * from td_sm_log");
			importBuilder.setInputConfig(dbInputConfig);

			importBuilder.addIgnoreFieldMapping("remark1");
//		importBuilder.setSql("select * from td_sm_log ");
			/**
			 * es相关配置
			 */
			ElasticsearchOutputConfig elasticsearchOutputConfig = new ElasticsearchOutputConfig();
			elasticsearchOutputConfig
					.setIndex("quartz") ;//必填项
//					.setIndexType("quartz") //es 7以后的版本不需要设置indexType，es7以前的版本必需设置indexType
//				.setRefreshOption("refresh")//可选项，null表示不实时刷新，importBuilder.setRefreshOption("refresh");表示实时刷新
			elasticsearchOutputConfig.setEsIdField("log_id");//设置文档主键，不设置，则自动产生文档id

			elasticsearchOutputConfig.setDebugResponse(false);//设置是否将每次处理的reponse打印到日志文件中，默认false
			elasticsearchOutputConfig.setDiscardBulkResponse(true);//设置是否需要批量处理的响应报文，不需要设置为false，true为需要，默认false
			/**
			 elasticsearchOutputConfig.setEsIdGenerator(new EsIdGenerator() {
			 //如果指定EsIdGenerator，则根据下面的方法生成文档id，
			 // 否则根据setEsIdField方法设置的字段值作为文档id，
			 // 如果默认没有配置EsIdField和如果指定EsIdGenerator，则由es自动生成文档id

			 @Override
			 public Object genId(Context context) throws Exception {
			 return SimpleStringUtil.getUUID();//返回null，则由es自动生成文档id
			 }
			 });
			 */
			importBuilder.setOutputConfig(elasticsearchOutputConfig);

			importBuilder.setUseJavaName(true) //可选项,将数据库字段名称转换为java驼峰规范的名称，true转换，false不转换，默认false，例如:doc_id -> docId
					.setUseLowcase(false)  //可选项，true 列名称转小写，false列名称不转换小写，默认false，只要在UseJavaName为false的情况下，配置才起作用
					.setPrintTaskLog(true) //可选项，true 打印任务执行日志（耗时，处理记录数） false 不打印，默认值false
					.setBatchSize(10);  //可选项,批量导入es的记录数，默认为-1，逐条处理，> 0时批量处理

			//定时任务配置，
			//采用内部定时任务
//		importBuilder.setFixedRate(false)//参考jdk timer task文档对fixedRate的说明
////					 .setScheduleDate(date) //指定任务开始执行时间：日期
//				.setDeyLay(1000L) // 任务延迟执行deylay毫秒后执行
//				.setPeriod(10000L); //每隔period毫秒执行，如果不设置，只执行一次
			//采用外部定时任务
			importBuilder.setExternalTimer(true);
			//定时任务配置结束
//
//		//设置任务执行拦截器，可以添加多个
//		importBuilder.addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Throwable e) {
//				System.out.println("throwException");
//			}
//		}).addCallInterceptor(new CallInterceptor() {
//			@Override
//			public void preCall(TaskContext taskContext) {
//				System.out.println("preCall 1");
//			}
//
//			@Override
//			public void afterCall(TaskContext taskContext) {
//				System.out.println("afterCall 1");
//			}
//
//			@Override
//			public void throwException(TaskContext taskContext, Throwable e) {
//				System.out.println("throwException 1");
//			}
//		});
//		//设置任务执行拦截器结束，可以添加多个
			//增量配置开始
//		importBuilder.setLastValueColumn("log_id");//手动指定数字增量查询字段，默认采用上面设置的sql语句中的增量变量名称作为增量查询字段的名称，指定以后就用指定的字段
			importBuilder.setFromFirst(true);//setFromfirst(false)，如果作业停了，作业重启后从上次截止位置开始采集数据，
			//setFromfirst(true) 如果作业停了，作业重启后，重新开始采集数据
			importBuilder.setLastValueStorePath("logtable_import");//记录上次采集的增量字段值的文件路径，作为下次增量（或者重启后）采集数据的起点，不同的任务这个路径要不一样
//		importBuilder.setLastValueStoreTableName("logs");//记录上次采集的增量字段值的表，可以不指定，采用默认表名increament_tab
			importBuilder.setLastValueType(ImportIncreamentConfig.NUMBER_TYPE);//如果没有指定增量查询字段名称，则需要指定字段类型：ImportIncreamentConfig.NUMBER_TYPE 数字类型
			// 或者ImportIncreamentConfig.TIMESTAMP_TYPE 日期类型
			//增量配置结束

			//映射和转换配置开始
//		/**
//		 * db-es mapping 表字段名称到es 文档字段的映射：比如document_id -> docId
//		 * 可以配置mapping，也可以不配置，默认基于java 驼峰规则进行db field-es field的映射和转换
//		 */
//		importBuilder.addFieldMapping("document_id","docId")
//				.addFieldMapping("docwtime","docwTime")
//				.addIgnoreFieldMapping("channel_id");//添加忽略字段
//
//
//		/**
//		 * 为每条记录添加额外的字段和值
//		 * 可以为基本数据类型，也可以是复杂的对象
//		 */
//		importBuilder.addFieldValue("testF1","f1value");
//		importBuilder.addFieldValue("testInt",0);
//		importBuilder.addFieldValue("testDate",new Date());
//		importBuilder.addFieldValue("testFormateDate","yyyy-MM-dd HH",new Date());
//		TestObject testObject = new TestObject();
//		testObject.setId("testid");
//		testObject.setName("jackson");
//		importBuilder.addFieldValue("testObject",testObject);
//
//		/**
//		 * 重新设置es数据结构
//		 */
//		importBuilder.setDataRefactor(new DataRefactor() {
//			public void refactor(Context context) throws Exception  {
//				CustomObject customObject = new CustomObject();
//				customObject.setAuthor((String)context.getValue("author"));
//				customObject.setTitle((String)context.getValue("title"));
//				customObject.setSubtitle((String)context.getValue("subtitle"));
//				customObject.setIds(new int[]{1,2,3});
//				context.addFieldValue("docInfo",customObject);//如果还需要构建更多的内部对象，可以继续构建
//
//				//上述三个属性已经放置到docInfo中，如果无需再放置到索引文档中，可以忽略掉这些属性
//				context.addIgnoreFieldMapping("author");
//				context.addIgnoreFieldMapping("title");
//				context.addIgnoreFieldMapping("subtitle");
//			}
//		});
			//映射和转换配置结束

			/**
			 * 内置线程池配置，实现多线程并行数据导入功能，作业完成退出时自动关闭该线程池
			 */
			importBuilder.setParallel(true);//设置为多线程并行批量导入,false串行
			importBuilder.setQueue(10);//设置批量导入线程池等待队列长度
			importBuilder.setThreadCount(5);//设置批量导入线程池工作线程数量
			importBuilder.setContinueOnError(true);//任务出现异常，是否继续执行作业：true（默认值）继续执行 false 中断作业执行

			/**
			 * 执行数据库表数据导入es操作
			 */
			return importBuilder;
		});

	}


}
