#include <sys/types.h>
#include <sys/stat.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <net/if.h>
#include <unistd.h> 
#include <stdio.h> 
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <sys/time.h>
#include <time.h>
#include <string.h>
#include <openssl/sha.h>
#include <openssl/md5.h>

#include <MESA/MESA_prof_load.h>

#include "doris_client_http.h"

void easy_string_destroy(struct easy_string *estr)
{
	if(estr->buff != NULL)
	{
		free(estr->buff);
		estr->buff = NULL;
 		estr->len = estr->size = 0;
	}
}

void easy_string_savedata(struct easy_string *estr, const char *data, size_t len)
{
	if(estr->size-estr->len < len+1)
	{
		estr->size += len*4+1;
		estr->buff = (char*)realloc(estr->buff, estr->size);
	}
	
	memcpy(estr->buff+estr->len, data, len);
	estr->len += len;
	estr->buff[estr->len]='\0';
}

static inline void drsclient_set_sockopt_keepalive(int sd, int keepidle, int keepintvl, int keepcnt)
{
	int keepalive = 1;
	setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (void*)&keepalive, sizeof(keepalive));
	setsockopt(sd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepidle, sizeof(keepidle));
	setsockopt(sd, SOL_TCP, TCP_KEEPINTVL, (void*)&keepintvl, sizeof(keepintvl));
	setsockopt(sd, SOL_TCP, TCP_KEEPCNT, (void*)&keepcnt, sizeof(keepcnt));
}

int32_t param_get_connected_hosts(struct doris_http_parameter *param)
{
	return param->connected_hosts;
}
int32_t param_get_failed_hosts(struct doris_http_parameter *param)
{
	return param->failed_hosts;
}

static int _unfold_IP_range(char* ip_range, char***ip_list, int size)
{
	int i=0,count=0, ret=0;
	int range_digits[5];
	memset(range_digits,0,sizeof(range_digits));
	ret=sscanf(ip_range,"%d.%d.%d.%d-%d",&range_digits[0],&range_digits[1],&range_digits[2],&range_digits[3],&range_digits[4]);
	if(ret!=4&&ret!=5)
	{
		return 0;
	}
	if(ret==4&&range_digits[4]==0)
	{
		range_digits[4]=range_digits[3];
	}
	for(i=0;i<5;i++)
	{
		if(range_digits[i]<0||range_digits[i]>255)
		{
			return 0;
		}
	}
	count=range_digits[4]-range_digits[3]+1;
	*ip_list=(char**)realloc(*ip_list, sizeof(char*)*(size+count));
	for(i=0;i<count;i++)
	{
		(*ip_list)[size+i]=(char*)malloc(64);
		snprintf((*ip_list)[size+i],64,"%d.%d.%d.%d",range_digits[0],range_digits[1],range_digits[2],range_digits[3]+i);
	}
	return count;
}

static int unfold_IP_range(const char* ip_range, char***ip_list)
{
	char *token=NULL,*sub_token=NULL,*saveptr;
	char *buffer=(char*)calloc(sizeof(char),strlen(ip_range)+1);
	int count=0;
	strcpy(buffer,ip_range);
	for (token = buffer; ; token= NULL)
	{
		sub_token= strtok_r(token,";", &saveptr);
		if (sub_token == NULL)
			break;
		count+=_unfold_IP_range(sub_token, ip_list,count);	
	}
	free(buffer);
	return count;
}

static int decode_one_specific_group_ip(const char *iplist, struct dst_ipaddr_group *dstaddr)
{
	int i, ipaddr_num;
	char **balance_iplist=NULL;
	
	ipaddr_num = unfold_IP_range(iplist, &balance_iplist);
	if(ipaddr_num == 0)
	{
		free(dstaddr);
		return -1;
	}
	dstaddr->dstaddr_num = ipaddr_num;
	dstaddr->dstaddrs = (u_int32_t *)calloc(1, sizeof(u_int32_t)*ipaddr_num);
	for(i=0; i<ipaddr_num; i++)
	{
		if(inet_pton(AF_INET, balance_iplist[i], (void *)&dstaddr->dstaddrs[i]) != 1)
		{
			free(dstaddr->dstaddrs);
			return -1;
		}
		free(balance_iplist[i]);
	}
	free(balance_iplist);
	return 0;
}

static int32_t load_and_init_server_group(char *dst_ipaddr, struct dst_ipaddr_group *dstaddrs, void *runtime_log)
{
	u_int32_t intval;

	intval = strlen(dst_ipaddr);
	if(dst_ipaddr[intval-1] != ';')
	{
		dst_ipaddr[intval] = ';';
		dst_ipaddr[intval+1] = '\0';
	}
	if(decode_one_specific_group_ip(dst_ipaddr, dstaddrs) < 0)
	{
		return -2;
	}
	return 0;
}

static void conhash_delay_destroy_timer_cb(int fd, short kind, void *userp)
{
	struct time_event *delay_event=(struct time_event *)userp;

	conhash_instance_free((struct consistent_hash *)delay_event->data);
	free(delay_event);
}

static void load_balance_common_timer_start(struct event *time_event)
{
	struct timeval tv;

	tv.tv_sec = 2;
	tv.tv_usec = 0;
	evtimer_add(time_event, &tv);
}

static void conhash_handle_delay_destroy(struct event_base *evbase, struct consistent_hash *conhash)
{
	struct time_event *delay_event;

	delay_event = (struct time_event *)malloc(sizeof(struct time_event));
	delay_event->data = conhash;
	evtimer_assign(&delay_event->timer_event, evbase, conhash_delay_destroy_timer_cb, delay_event);
	load_balance_common_timer_start(&delay_event->timer_event);
}

static void conhash_insert_dest_host(struct dst_host_cnn_balance *balance)
{
	struct conhash_bucket bucket;
	struct consistent_hash *tmphash, *newhash=NULL;
	enum CONHASH_ERRCODE code;
	
	bucket.bucket_id = balance->dstip;
	bucket.point_num = DEFAULT_HOST_CAPACITY * LOAD_BALANC_VIRT_TIMES;;
	bucket.tag = NULL;

	newhash = conhash_instance_copy(balance->param->conhash);
	code = conhash_insert_bucket(newhash, &bucket);
	assert(code == CONHASH_OK);
	
	tmphash = balance->param->conhash;
	balance->param->conhash = newhash;
	conhash_handle_delay_destroy(balance->param->evbase, tmphash);
}

static void conhash_remove_dest_host(struct dst_host_cnn_balance *balance)
{
	struct consistent_hash *tmphash, *newhash=NULL;
	enum CONHASH_ERRCODE code;
	
	newhash = conhash_instance_copy(balance->param->conhash);
	code = conhash_remove_bucket(newhash, balance->dstip, NULL);
	assert(code == CONHASH_OK || code==CONHASH_BUCKET_NOT_FOUND);
	
	tmphash = balance->param->conhash;
	balance->param->conhash = newhash;
	conhash_handle_delay_destroy(balance->param->evbase, tmphash);
}

static void client_bufferevent_error_cb(struct bufferevent *bev, short event, void *arg)  
{
	struct dst_host_cnn_balance *balance = (struct dst_host_cnn_balance *)arg;
	const char *errtype;
	
	if(event & BEV_EVENT_CONNECTED) 
	{  
		errtype = "connected";
		balance->connection_status = TCP_STATUS_CONNECTED;
		balance->param->connected_hosts += 1; //仅有管理线程一个会进行链接探测
		if(balance->connect_failed)
		{
			balance->connect_failed = false;
			balance->param->failed_hosts -= 1;
		}
		conhash_insert_dest_host(balance);
		assert(balance->param->connected_hosts > 0);
		assert(balance->param->failed_hosts >= 0);
		drsclient_set_sockopt_keepalive(bufferevent_getfd(bev), 10, 5, 2);
	}
	else
	{
		if (event & BEV_EVENT_TIMEOUT) {  
			errtype = "Timed out";  
		}
		else if (event & BEV_EVENT_EOF) {  
			errtype = "disconnected";  
		}
		else if (event & BEV_EVENT_ERROR) {
			errtype = "some other error"; 
		}
		else {
			errtype = "unkonwn error";
		}
		bufferevent_free(bev);
		balance->bev = NULL;

		if(balance->connection_status == TCP_STATUS_CONNECTED)
		{
			balance->param->connected_hosts -= 1;
		}
		if(!balance->connect_failed)
		{
			balance->connect_failed = true;
			balance->param->failed_hosts += 1;
		}
		balance->connection_status = TCP_STATUS_DISCONNECT;
	
		load_balance_common_timer_start(&balance->timer_detect);
		conhash_remove_dest_host(balance);

		assert(balance->param->connected_hosts >= 0);
		assert(balance->param->failed_hosts > 0);
	}
	
	MESA_HANDLE_RUNTIME_LOGV2(balance->param->runtime_log, RLOG_LV_INFO, "connection event: %s, addr: %s", errtype, balance->srvaddr);
}

int do_bufferevent_connection(struct doris_http_parameter *param, struct dst_host_cnn_balance *balance, struct sockaddr *server_addr)
{
	if(balance->connection_status==TCP_STATUS_CONNECTING)
	{
		return 0;
	}

	if(NULL == (balance->bev = bufferevent_socket_new(param->evbase, -1, BEV_OPT_CLOSE_ON_FREE)))
	{
		assert(0);return -1;
	}
	
	if(bufferevent_socket_connect(balance->bev, server_addr, sizeof(struct sockaddr_in)))
	{
		bufferevent_free(balance->bev);
		balance->bev = NULL;
		MESA_HANDLE_RUNTIME_LOGV2(param->runtime_log, RLOG_LV_FATAL, "bufferevent_socket_connect error: %s", strerror(errno));
		assert(0);return -2;
	}
	balance->connection_status = TCP_STATUS_CONNECTING;
	bufferevent_setcb(balance->bev, NULL, NULL, client_bufferevent_error_cb, balance);
	bufferevent_setwatermark(balance->bev, EV_WRITE, 100*1024*1024UL, 0);
	bufferevent_enable(balance->bev, EV_READ|EV_WRITE|EV_PERSIST);
	return 0;
}

void param_connection_detect_timer_cb(int fd, short kind, void *userp)
{
	struct dst_host_cnn_balance *balance = (struct dst_host_cnn_balance *)userp;

	if(do_bufferevent_connection(balance->param, balance, (struct sockaddr*)&balance->addr))
	{
		assert(0);
	}
}

static int32_t doris_launch_group_connection(struct doris_http_parameter *param, struct event_base* evbase)
{
	char ipaddr[64];

	param->balance = (struct dst_host_cnn_balance *)calloc(1, sizeof(struct dst_host_cnn_balance)*param->ipgroup.dstaddr_num);
	
	for(u_int32_t i=0; i<param->ipgroup.dstaddr_num; i++) //组内
	{
		param->balance[i].dstip = param->ipgroup.dstaddrs[i];
		param->balance[i].addr.sin_family = AF_INET;
		param->balance[i].addr.sin_port = htons(param->manage_port);
		param->balance[i].addr.sin_addr.s_addr = param->ipgroup.dstaddrs[i];
		param->balance[i].connection_status = TCP_STATUS_IDLE;
		param->balance[i].param = param;
	
		inet_ntop(AF_INET, &param->ipgroup.dstaddrs[i], ipaddr, 64);
		snprintf(param->balance[i].srvaddr, 64, "%s:%u", ipaddr, param->server_port);

		evtimer_assign(&param->balance[i].timer_detect, evbase, param_connection_detect_timer_cb, &param->balance[i]);
	
		if(do_bufferevent_connection(param, &param->balance[i], (struct sockaddr *)&param->balance[i].addr))
		{
			MESA_HANDLE_RUNTIME_LOGV2(param->runtime_log, RLOG_LV_FATAL, "do_bufferevent_connection error: %s.", strerror(errno));
			return -1;
		}
	}
	return 0;
}

void doris_http_parameter_destroy(struct doris_http_parameter *param)
{
	for(u_int32_t i=0; i<param->ipgroup.dstaddr_num; i++) //组内
	{
		if(evtimer_pending(&param->balance[i].timer_detect, NULL))
		{
			evtimer_del(&param->balance[i].timer_detect);
		}
		if(param->balance[i].bev != NULL)
		{
			bufferevent_free(param->balance[i].bev);
		}
	}
	conhash_instance_free(param->conhash);
	free(param->ipgroup.dstaddrs);
	free(param);
}

struct doris_http_parameter *doris_http_parameter_new(const char* profile_path, const char* section, struct event_base* evbase, void *runtime_log)
{
	struct doris_http_parameter *param;
	u_int32_t intval;
	char dst_ipaddr[8192];

	param = (struct doris_http_parameter *)calloc(1, sizeof(struct doris_http_parameter));
	param->runtime_log = runtime_log;
	param->evbase = evbase;

	//multi curl
	MESA_load_profile_uint_def(profile_path, section, "max_connection_per_host", &intval, 1);
	param->maximum_host_cnns = intval;
	MESA_load_profile_uint_def(profile_path, section, "max_cnnt_pipeline_num", &intval, 20);
	param->maximum_pipelines = intval;
	MESA_load_profile_uint_def(profile_path, section, "max_curl_transfer_timeout_s", &intval, 0);
	param->transfer_timeout = intval;
	MESA_load_profile_int_def(profile_path, section, "https_connection_on", &param->ssl_connection, 0);
	MESA_load_profile_uint_def(profile_path, section, "max_curl_session_num", &param->max_http_sessions, 30);

	//Server
	if(MESA_load_profile_uint_def(profile_path, section, "http_server_listen_port", &param->server_port, 9898) < 0)
	{
		free(param);
		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Load config %s [%s] http_server_listen_port not found.", profile_path, section);
		return NULL;
	}
	MESA_load_profile_uint_def(profile_path, section, "http_server_manage_port", &param->manage_port, 2233);

	if(MESA_load_profile_string_nodef(profile_path, section, "http_server_ip_list", dst_ipaddr, 8192) < 0)
	{
		free(param);
		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Load config %s [%s] http_server_ip_list not found.", profile_path, section);
		return NULL;
	}
	
	if(load_and_init_server_group(dst_ipaddr, &param->ipgroup, runtime_log))
	{
		MESA_HANDLE_RUNTIME_LOGV2(runtime_log, RLOG_LV_FATAL, "Decode %s [%s] http_server_ip_list format error: %s", profile_path, section, dst_ipaddr);
		assert(0);return NULL;
	}
	param->conhash = conhash_instance_new(NULL, 0);;
	if(doris_launch_group_connection(param, evbase))
	{
		assert(0);return NULL;
	}
	return param;
}

void doris_http_instance_destroy(struct doris_http_instance *instance)
{
	map<u_int32_t, doris_curl_multihd*>::iterator iter;
	struct doris_curl_multihd *multihd;
	CURLMsg *msg;
	int msgs_left;
	struct doris_http_ctx *ctx;
	CURL *easy;

	for(iter=instance->server_hosts->begin(); iter!=instance->server_hosts->end(); )
	{
		multihd = iter->second;

		while((msg = curl_multi_info_read(multihd->multi_hd, &msgs_left)))
		{
			easy = msg->easy_handle;
			curl_easy_getinfo(easy, CURLINFO_PRIVATE, &ctx);
			curl_multi_remove_handle(multihd->multi_hd, easy);
			curl_easy_cleanup(easy);
			ctx->curl = NULL;
			ctx->transfering = 0;
			ctx->res = CURLE_ABORTED_BY_CALLBACK;
			ctx->res_code = 0;
			ctx->cb.transfer_done_cb(ctx->res, 0, ctx->error, ctx->cb.userp);
		}
		curl_multi_cleanup(multihd->multi_hd);
		
		instance->server_hosts->erase(iter++);
	}
	free(instance);
}

struct doris_http_instance *doris_http_instance_new(struct doris_http_parameter *param, struct event_base* evbase, void *runtimelog)
{
	struct doris_http_instance *instance;
	struct doris_curl_multihd *multihd;

	instance = (struct doris_http_instance *)calloc(1, sizeof(struct doris_http_instance));
	instance->runtime_log = runtimelog;	
	instance->evbase = evbase;
	instance->param = param;
	
	//为每一个minio IP都创建一个multi_handle
	instance->server_hosts = new map<u_int32_t, doris_curl_multihd*>;
	for(u_int32_t i=0; i<param->ipgroup.dstaddr_num; i++)
	{
		multihd = doris_initialize_multihd_for_host(instance, &param->balance[i]);
		instance->server_hosts->insert(make_pair(param->ipgroup.dstaddrs[i], multihd));
	}
	return instance;
}