package com.zhaoonline.common.es;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.ParseException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.RequestBuilder;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.zhaoonline.common.Utils.IOUtils;
import com.zhaoonline.common.es.bean.BulkResponse;
import com.zhaoonline.common.es.bean.DeleteResponse;
import com.zhaoonline.common.es.bean.GetResponse;
import com.zhaoonline.common.es.bean.IndexResponse;
import com.zhaoonline.common.es.bean.InterSearchHits;
import com.zhaoonline.common.es.bean.InterSearchResponse;
import com.zhaoonline.common.es.bean.QueryResponse;
import com.zhaoonline.common.json.JsonUtils;


/**
 * class name:ESClient <BR>
 * class description:Elastic的客户端 <BR>
 * Remark: <BR>
 * @version 1.00 2016年10月24日
 * @author zhaoonline)yangyoupeng
 */
public class ESHttpClient {
	private static final Logger LOG = LoggerFactory.getLogger(ESHttpClient.class);
	private static final String PATH_SEPERATOR = "/";
	private String _index;
	private String _type;
	private static final AtomicReference<CloseableHttpClient> httpClient = new AtomicReference<CloseableHttpClient>();
	private static final Thread monitorThread=createMonitorThread();
	private ESClientConfiguration config;
	private String path;
	private List<HttpHost> hostsList=new ArrayList<HttpHost>();
	
	
    private static final Timer CONNECTION_MANAGER_TIMER = new Timer(true);
    
    static {
	        CONNECTION_MANAGER_TIMER.schedule(new TimerTask() {
	            @Override
				public void run() {
	                try {
	                    final CloseableHttpClient hc = httpClient.get();
	                    if (hc == null) return;
	                    hc.getConnectionManager().closeExpiredConnections();
	                    hc.getConnectionManager().closeIdleConnections(30, TimeUnit.SECONDS);
	                } catch (Throwable t) {
	                    LOG.error("error closing expired connections", t);
	                }
	            }
	        }, 30000, 5000);
	        Runtime.getRuntime().addShutdownHook(monitorThread);
	}
    
    
    
	public ESHttpClient(String index,String type,ESClientConfiguration config){
		this._index=index;
		this._type=type;
		concatPathWithIndexType();
		this.config=config;
	}

	private static Thread createMonitorThread() {
		Thread monitorThread =new Thread(new Runnable() {
			@Override
			public void run() {
				CONNECTION_MANAGER_TIMER.cancel();
				LOG.info("pool Manager check timer cancel running");
			}
		});
		return monitorThread;
	}
	private void concatPathWithIndexType() {
		this.path=_index+PATH_SEPERATOR+_type;
	}
	public void init(){
		getClient();
		List<String> hostPosts=config.getHostPorts();
		hostsList=buildHostPortListFromString(hostPosts);
		
	}

	private CloseableHttpClient getClient() {
		if(httpClient.get() ==null){
			httpClient.set(HttpClientFactory.newClient());
		}
		return httpClient.get();
	}

	
	private static List<HttpHost> buildHostPortListFromString(List<String> stringhostPortList) {
		List<HttpHost> hostPorts=new ArrayList<HttpHost>(stringhostPortList.size());
		for(String hostPortString: stringhostPortList){
			
			String[] hostPortArray= hostPortString.split(":");
			if(hostPortArray!=null &&  hostPortArray.length>=2){
				String host=String.valueOf(hostPortArray[0]);
				String port=String.valueOf(hostPortArray[1]);
				HttpHost hostPort= new HttpHost(host,Integer.valueOf(port));
				hostPorts.add(hostPort);
			}
		}
		return hostPorts;
	}
	
	/**
	 * Method name: addDoc <BR>
	 * Description:  添加doc，指定id <BR>
	 * Remark: <BR>
	 * @param id
	 * @param docString
	 * @return  boolean<BR>
	 * @throws IOException 
	 * @throws UnsupportedOperationException 
	 * @throws JsonMappingException 
	 * @throws JsonParseException 
	 */
	public boolean addDoc(String id, String docString) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException {
		String pathWithId=path+PATH_SEPERATOR+id;
		CloseableHttpResponse response=	sendPostRequest(docString, pathWithId,this.hostsList);
		
		HttpEntity responseEntity=response.getEntity();
		IndexResponse indexResponse=JsonUtils.toObject(responseEntity.getContent(), IndexResponse.class);
		if(indexResponse !=null){
			return indexResponse.isSuccess();
		}
		IOUtils.closeStream(response);
		return  false;
	}


	public Map searchById(String id) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException{
		String pathWithId=path+PATH_SEPERATOR+id;
		HttpRequest request=RequestBuilder.get(pathWithId).build();
		CloseableHttpResponse httpresponse=	sendRequest(request,this.hostsList);
		if(httpresponse ==null){
			return null;
		}
		HttpEntity responseEntity=httpresponse.getEntity();
		GetResponse response=JsonUtils.toObject(responseEntity.getContent(), GetResponse.class);
		IOUtils.closeStream(httpresponse);
		return response.getSource();
	}
	
	
	/**
	 * Method name: search <BR>
	 * Description: search 利用post方法查询doc<BR>
	 * Remark: <BR>
	 * @param queryString
	 * @return
	 * @throws JsonParseException
	 * @throws JsonMappingException
	 * @throws UnsupportedOperationException
	 * @throws IOException  List<Map><BR>
	 */
	public QueryResponse query(String queryString) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException{
		String pathWithId=path+PATH_SEPERATOR+"_search";
		CloseableHttpResponse response= sendPostRequest(queryString,pathWithId,this.hostsList);
		HttpEntity responseEntity=response.getEntity();
		String str=EntityUtils.toString(responseEntity);
		InterSearchResponse searchResponse=JsonUtils.toObject(str, InterSearchResponse.class);
		if(searchResponse ==null){
			return null;
		}
		QueryResponse queryResponse =new QueryResponse();
		InterSearchHits searchHits=searchResponse.getHits();
		queryResponse.setTotal(searchHits.getTotal());
		queryResponse.addData(searchHits.getHits());
		IOUtils.closeStream(response);
		return queryResponse;
	}
	
	
	/**
	 * Method name: query <BR>
	 * Description: 根据pagesize返回内容 <BR>
	 * Remark: <BR>
	 * @param queryBuilder
	 * @return
	 * @throws JsonParseException
	 * @throws JsonMappingException
	 * @throws UnsupportedOperationException
	 * @throws IOException  QueryResponse<BR>
	 */
	public QueryResponse query(ZhaoQueryBuilder queryBuilder) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException{
		QueryResponse queryResponse=this.query(queryBuilder.toQueryString());
		return queryResponse;
	}
	
	/**
	 * Method name: queryAll <BR>
	 * Description: 返回所有doc <BR>
	 * Remark: <BR>
	 * @param queryBuilder
	 * @return
	 * @throws JsonParseException
	 * @throws JsonMappingException
	 * @throws UnsupportedOperationException
	 * @throws IOException  QueryResponse<BR>
	 */
	public QueryResponse queryAll(ZhaoQueryBuilder queryBuilder) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException{
		QueryResponse returnResponse=new QueryResponse();
		
		int size=queryBuilder.getPage().getSize();
		if(size ==0){
			return null;
		}
		
		LoopCount loopcount=null;
		
		boolean readDone=false;
		while(!readDone){
			QueryResponse queryResponse=this.query(queryBuilder);
			int totalSize=queryResponse.getTotal().intValue();
			int returnDataSize=queryResponse.getDataList().size();
		
			
			returnResponse.addData(queryResponse.getDataList());
			int alreadyReadSize=queryBuilder.getPage().getStart()+returnDataSize;
			if(alreadyReadSize >=totalSize ){
				readDone=true;
				returnResponse.setTotal(new Long(totalSize));
				break;
			}
			//为了防止异常的出现死循环
			if(loopcount==null){
				loopcount=new LoopCount();
				loopcount.maxLoopcount=totalSize/size;
			}else if(loopcount.maxLoopcount <= 0){
				break;
			}else{
				loopcount.maxLoopcount=loopcount.maxLoopcount-1;
			}
			queryBuilder.setPage(alreadyReadSize, size);
			
		}
		return returnResponse;
	}
	
	private  class LoopCount{
		int maxLoopcount=0;
	}
	
	
	/**
	 * Method name: addDoc <BR>
	 * Description: 添加doc，id由ES自动生成 <BR>
	 * Remark: <BR>
	 * @param docString
	 * @return  boolean<BR>
	 * @throws IOException 
	 * @throws UnsupportedOperationException 
	 */
	public boolean addDoc(String docString) throws UnsupportedOperationException, IOException {
		CloseableHttpResponse response= sendPostRequest(docString, this.path,this.hostsList);
		HttpEntity responseEntity=response.getEntity();
		IndexResponse indexResponse=JsonUtils.toObject(responseEntity.getContent(), IndexResponse.class);
		if(indexResponse !=null){
			return true;
		}
		IOUtils.closeStream(response);
		return  false;
	}
	
	
	
	private CloseableHttpResponse sendPostRequest(String source, String path,List<HttpHost> hostsList) {
		HttpEntity entity=new StringEntity(source,ContentType.APPLICATION_JSON);
		HttpRequest request=RequestBuilder.post(path).setEntity(entity).build();
		CloseableHttpResponse response=sendRequest(request,hostsList);
		return response;
	}

	private CloseableHttpResponse sendRequest(HttpRequest request,List<HttpHost> hostsList) {
		CloseableHttpResponse response=null;
		for(HttpHost httpHost:hostsList){
			try {
				response= getClient().execute(httpHost, request);
				if(response !=null){
					return response;
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return response;
	}
	
	public void close(){
		IOUtils.closeStream(getClient());
		httpClient.set(null);
	}

	public boolean deleteDoc(String id) throws JsonParseException, JsonMappingException, UnsupportedOperationException, IOException {
		String pathWithId=path+PATH_SEPERATOR+id;
		HttpRequest request=RequestBuilder.delete(pathWithId).build();
		CloseableHttpResponse response=	sendRequest(request,this.hostsList);
		if(response ==null){
			return false;
		}
		HttpEntity responseEntity=response.getEntity();
		DeleteResponse deleteResponse=JsonUtils.toObject(responseEntity.getContent(), DeleteResponse.class);
		IOUtils.closeStream(response);
		return deleteResponse.isSuccess();
	}

	/**
	 * Method name: bulkAddDoc <BR>
	 * Description: 批量添加指定ID <BR>
	 * Remark: <BR>
	 * @param ids
	 * @param dataList
	 * @return
	 * @throws UnsupportedOperationException
	 * @throws IOException  boolean<BR>
	 */
	public boolean bulkAddDoc(List ids,List dataList) throws UnsupportedOperationException, IOException {
		if(!validateBulkInput(ids, dataList)){
			throw new IllegalArgumentException("the ids size is not equal to dataList ");
		};
		
		StringBuilder builder=buildBulkIndexScript(ids,dataList);
		if(builder.length()==0){
			return false;
		}
		return bulkAdd(builder.toString());
	}

	private boolean validateBulkInput(List ids, List dataList) {
		if(ids ==null || dataList == null){
			return false;
		}
		if(ids.size() != dataList.size() ){
			return false;
		}
		return true;
	}
	
	
	/**
	 * Method name: bulkAddDoc <BR>
	 * Description: 批量添加自动生成ID <BR>
	 * Remark: <BR>
	 * @param dataList
	 * @return
	 * @throws UnsupportedOperationException
	 * @throws IOException  boolean<BR>
	 */
	public boolean bulkAddDoc(List dataList) throws UnsupportedOperationException, IOException {
		StringBuilder builder=buildBulkIndexScript(null,dataList);
		if(builder.length()==0){
			return false;
		}
		return bulkAdd(builder.toString());
	}
	
	public boolean bulkAdd(String bulkScripts) throws ParseException, IOException{
		String bulkPath=path+PATH_SEPERATOR+"_bulk";
		CloseableHttpResponse response= sendPostRequest(bulkScripts,bulkPath,this.hostsList);
		if(response ==null){
			return false;
		}
		HttpEntity responseEntity=response.getEntity();
		BulkResponse bulkResponse=JsonUtils.toObject(responseEntity.getContent(), BulkResponse.class);
		IOUtils.closeStream(response);
		return !bulkResponse.isErrors();
	}
	

	public  StringBuilder buildBulkIndexScript(List ids,List dataList) throws JsonProcessingException {
		StringBuilder bulkBuilder=new StringBuilder();
		if(ids !=null && ids.size() !=0){
			for(int i=0;i<ids.size();i++){
				StringBuilder itemBuilder=new StringBuilder();
				itemBuilder.append("{ \"index\" :").append("{ \"_index\" : \"").
													append(this._index).append("\"").append(",\"_type\" :\"").append(this._type).append("\"")
													.append(",\"_id\" :\"").append(ids.get(i)).append("\"")
													.append("} }").append("\n");
				String objectjson=JsonUtils.toJson(dataList.get(i));
				itemBuilder.append(objectjson).append("\n");
				bulkBuilder.append(itemBuilder);
			}
		}else {
			for(Object object:dataList){
				StringBuilder itemBuilder=new StringBuilder();
				itemBuilder.append("{ \"index\" :").append("{ \"_index\" : \"").
				append(this._index).append("\"").append(",\"_type\" :\"").append(this._type).append("\"")
				//.append(",\"_id\" :\"").append(1).append("\"")
				.append("} }").append("\n");
				String objectjson=JsonUtils.toJson(object);
				bulkBuilder.append(itemBuilder);
			}
		}
		return bulkBuilder;
	}
}