tango-tfe/cache/src/tango_cache_redis.cpp

#include <sys/types.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 <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>
#include <cjson/cJSON.h>

#include "tango_cache_transfer.h"
#include "tango_cache_tools.h"
#include "tango_cache_redis.h"

#define PARSE_JSON_RET_ERROR -1
#define PARSE_JSON_RET_TIMEOUT  0
#define PARSE_JSON_RET_SUCC  1

#define CACHE_REDIS_CONNECT_IDLE 0
#define CACHE_REDIS_CONNECTING   1
#define CACHE_REDIS_CONNECTED    2
#define CACHE_REDIS_DISCONNECTED 3

struct http_hdr_name
{
	const char *json_name;
	const char *http_name;
};
struct http_hdr_name g_http_hdr_name[HDR_CONTENT_NUM]=
{
	{"content-type", "Content-Type: "},
	{"content-encoding", "Content-Encoding: "},
	{"content-disposition", "Content-Disposition: "},
	{"content-md5", "Content-MD5: "}
};

//һ<><D2BB>mainip<69><70><EFBFBD>ӳɹ<D3B3><C9B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><D0BB><EFBFBD><EFBFBD><EFBFBD>
static void main_redis_asyn_connect_cb(const struct redisAsyncContext *ac, int status)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)redisAsyncGetConnectionData(ac);
	
	if(status == REDIS_OK)
	{
		evtimer_del(&instance->timer_redis);
		if(instance->redis_connecting == CACHE_REDIS_CONNECTED)
		{
			redisAsyncDisconnect(instance->redis_ac);
		}
		sprintf(instance->current_redisip, "%s", instance->param->redis.mainip);
		instance->redis_ac = (struct redisAsyncContext *)ac;
		instance->redis_connecting = CACHE_REDIS_CONNECTED;
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis connect %s:%u success.", 
			instance->param->redis.mainip, instance->param->redis.port);
	}
	else
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis connect %s:%u failed: %s.", 
			instance->param->redis.mainip, instance->param->redis.port, ac->errstr);
	}
}

static void redis_asyn_disconnect_cb(const struct redisAsyncContext *ac, int status)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)redisAsyncGetConnectionData(ac);
	
	if(status == REDIS_OK)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis disconnect %s:%u success.", 
			instance->current_redisip, instance->param->redis.port);
	}
	else
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis disconnect %s:%u failed: %s.", 
			instance->current_redisip, instance->param->redis.port, ac->errstr);
	}
	instance->redis_connecting = CACHE_REDIS_DISCONNECTED;
	
	if(!strcmp(instance->current_redisip, instance->param->redis.mainip))
	{
		main_redis_check_timer_start(instance);
	}
}

void main_redis_check_timer_cb(evutil_socket_t fd, short what, void *arg)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)arg;
	redisAsyncContext *redis_ac;
	struct timeval tv;
	
	redis_ac = redisAsyncConnect(instance->param->redis.mainip, instance->param->redis.port);
	if(redis_ac == NULL)
	{
		return ;
	}
	redisLibeventAttach(redis_ac, instance->evbase);
	redisAsyncSetConnectionData(redis_ac, instance);
	redisAsyncSetConnectCallback(redis_ac, main_redis_asyn_connect_cb);
	redisAsyncSetDisconnectCallback(redis_ac, redis_asyn_disconnect_cb);

	tv.tv_sec = 60;
	tv.tv_usec = 0;
	evtimer_add(&instance->timer_redis, &tv);
}

void main_redis_check_timer_start(struct tango_cache_instance *instance)
{
	struct timeval tv;
	
	tv.tv_sec = 60;
	tv.tv_usec = 0;
	evtimer_assign(&instance->timer_redis, instance->evbase, main_redis_check_timer_cb, instance);
	evtimer_add(&instance->timer_redis, &tv);
}

static void redis_asyn_connect_cb(const struct redisAsyncContext *ac, int status)
{
	struct tango_cache_instance *instance = (struct tango_cache_instance *)redisAsyncGetConnectionData(ac);

	if(status == REDIS_OK)
	{
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis connect %s:%u success.", 
			instance->current_redisip, instance->param->redis.port);
		instance->redis_connecting = CACHE_REDIS_CONNECTED;
	}
	else
	{
		instance->redis_connecting = CACHE_REDIS_CONNECT_IDLE;
		MESA_HANDLE_RUNTIME_LOGV2(instance->runtime_log, RLOG_LV_INFO, "Redis connect %s:%u failed: %s.", 
			instance->current_redisip, instance->param->redis.port, ac->errstr);
		if(!strcmp(instance->current_redisip, instance->param->redis.mainip))
		{
			main_redis_check_timer_start(instance);
		}
	}
}

int redis_asyn_connect_init(struct tango_cache_instance *instance, const char *redisip)
{
	sprintf(instance->current_redisip, "%s", redisip); //mainip<69>õ<EFBFBD>ʱ<EFBFBD><CAB1>ʹ<EFBFBD><CAB9>mainip

	instance->redis_ac = redisAsyncConnect(instance->current_redisip, instance->param->redis.port);
	if(instance->redis_ac == NULL)
	{
		return -1;
	}
	instance->redis_connecting = CACHE_REDIS_CONNECTING;
	redisLibeventAttach(instance->redis_ac, instance->evbase);
	redisAsyncSetConnectionData(instance->redis_ac, instance);
	redisAsyncSetConnectCallback(instance->redis_ac, redis_asyn_connect_cb);
	redisAsyncSetDisconnectCallback(instance->redis_ac, redis_asyn_disconnect_cb);
	return 0;
}

int wiredlb_redis_asyn_connect(struct tango_cache_instance *instance)
{
	struct WLB_consumer_t cons_array[64];
	int i, cons_num;

	cons_num = wiredLB_list(instance->param->redis.wiredlb, 64, cons_array);
	for(i=0; i<cons_num; i++)
	{
		if(strcmp(instance->param->redis.mainip, cons_array[i].ip_addr))
		{
			if(0==redis_asyn_connect_init(instance, cons_array[i].ip_addr))
			{
				break;
			}
		}
	}
	if(i == cons_num)
	{
		return -1;
	}
	return 0;
}

static int parse_minio_events_json(struct tango_cache_ctx *ctx, const char *jcontent)
{
	cJSON *root, *pobject = NULL, *ptarget, *plastMod, *pexpires;
	int ret = PARSE_JSON_RET_ERROR;
	char usertag[2048];
	size_t datalen;

	//Records[0]->s3->object->key...userMetaData->metas...
	if(NULL == (root=cJSON_Parse(jcontent)))
	{
		goto out_json;
	}
	if(NULL==(pobject=cJSON_GetObjectItem(root, "Records")) || pobject->type!=cJSON_Array)
	{
		goto out_json;
	}
	if(NULL == (pobject=cJSON_GetArrayItem(pobject, 0))) //<2F><>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ԫ<EFBFBD>أ<EFBFBD>һ<EFBFBD><D2BB>ֻ<EFBFBD><D6BB>һ<EFBFBD><D2BB>
	{
		goto out_json;
	}
	if(NULL == (pobject=cJSON_GetObjectItem(pobject, "s3")) || pobject->type!=cJSON_Object)
	{
		goto out_json;
	}
	if(NULL == (pobject=cJSON_GetObjectItem(pobject, "object")) || pobject->type!=cJSON_Object)
	{
		goto out_json;
	}
	//<2F><>ȡ<EFBFBD><C8A1><EFBFBD><EFBFBD>
	if(NULL == (ptarget=cJSON_GetObjectItem(pobject, "size")) || ptarget->type!=cJSON_Number)
	{
		goto out_json;
	}
	ctx->get.result.tlength = ptarget->valueint; //TODO: <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>4GB<47><42>ô<EFBFBD>죿
	if(NULL == (ptarget=cJSON_GetObjectItem(pobject, "userMetadata")) || ptarget->type!=cJSON_Object)
	{
		goto out_json;
	}
	if(NULL==(plastMod=cJSON_GetObjectItem(ptarget, "X-Amz-Meta-Lm")) || NULL==(pexpires=cJSON_GetObjectItem(ptarget, "expires")))
	{
		goto out_json;
	}
	ctx->get.need_hdrs = RESPONSE_HDR_ALL;
	ctx->get.last_modify = atol(plastMod->valuestring);
	ctx->get.expires = expires_hdr2timestamp(pexpires->valuestring, strlen(pexpires->valuestring));
	if(!check_expires_fresh_header(ctx))
	{
		ret = PARSE_JSON_RET_TIMEOUT;
		goto out_json;
	}
	
	if(NULL!=(plastMod=cJSON_GetObjectItem(ptarget, "X-Amz-Meta-User")))
	{
		if((datalen = Base64_DecodeBlock((unsigned char*)plastMod->valuestring, strlen(plastMod->valuestring), (unsigned char*)usertag, 2048))>0)
		{
			easy_string_savedata(&ctx->get.response_tag, usertag, datalen);
		}
	}
	for(int i=0; i<HDR_CONTENT_NUM; i++)
	{
		if(NULL != (plastMod=cJSON_GetObjectItem(ptarget, g_http_hdr_name[i].json_name)))
		{
			easy_string_savedata(&ctx->response, g_http_hdr_name[i].http_name, strlen(g_http_hdr_name[i].http_name));
			easy_string_savedata(&ctx->response, plastMod->valuestring, strlen(plastMod->valuestring));
			easy_string_savedata(&ctx->response, "\r\n", strlen("\r\n"));
		}
	}	
	cJSON_Delete(root); 
	return PARSE_JSON_RET_SUCC;
	
out_json:
	cJSON_Delete(root); 
	return ret;
}

static void redis_hget_command_cb(struct redisAsyncContext *ac, void *vreply, void *privdata)
{
	redisReply *reply = (redisReply *)vreply;
	struct tango_cache_ctx *ctx  = (struct tango_cache_ctx *)privdata;
	int ret;

	if(reply == NULL || reply->type!=REDIS_REPLY_STRING)
	{
		if(reply!=NULL && reply->type == REDIS_REPLY_NIL)
		{
			tango_cache_set_fail_state(ctx, CACHE_CACHE_MISS);
			ctx->get.result.type = RESULT_TYPE_MISS;
			promise_success(ctx->promise, &ctx->get.result);
			promise_finish(ctx->promise);
		}
		else
		{
			tango_cache_set_fail_state(ctx, CACHE_ERR_REDIS_JSON);
			if(reply!=NULL && reply->type==REDIS_REPLY_ERROR)
			{
				promise_failed(ctx->promise, FUTURE_ERROR_CANCEL, reply->str);
			}
			else
			{
				promise_failed(ctx->promise, FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
			}
		}
		tango_cache_ctx_destroy(ctx);
		return;
	}

	ret = parse_minio_events_json(ctx, reply->str);
	switch(ret)
	{
		case PARSE_JSON_RET_ERROR:
			tango_cache_set_fail_state(ctx, CACHE_ERR_REDIS_JSON);
			promise_failed(ctx->promise, FUTURE_ERROR_CANCEL, tango_cache_get_errstring(ctx));
			tango_cache_ctx_destroy(ctx);
			break;
		case PARSE_JSON_RET_TIMEOUT:
			if(ctx->get.state == GET_STATE_DELETE)
			{
				ctx->get.state = GET_STATE_END;
				cache_delete_minio_object(ctx);
			}
			break;
		case PARSE_JSON_RET_SUCC:
			fetch_header_over_biz(ctx);
			ctx->get.result.type = RESULT_TYPE_END;
			promise_success(ctx->promise, &ctx->get.result);
			promise_finish(ctx->promise);
			tango_cache_ctx_destroy(ctx);
			break;
		default: assert(0);break;
	}
}

int tango_cache_head_redis(struct tango_cache_ctx *ctx)
{
	int ret = -1;

	ctx->instance->statistic.get_recv_num += 1;
	switch(ctx->instance->redis_connecting)
	{
		case CACHE_REDIS_CONNECTED:
			ret = redisAsyncCommand(ctx->instance->redis_ac, redis_hget_command_cb, ctx, "HGET %s %s/%s", 
				ctx->instance->param->redis_key, ctx->instance->param->bucketname, ctx->object_key);
			if(ret != REDIS_OK)
			{
				ctx->instance->redis_connecting = CACHE_REDIS_CONNECT_IDLE;
				if(!strcmp(ctx->instance->current_redisip, ctx->instance->param->redis.mainip))
				{
					main_redis_check_timer_start(ctx->instance);
				}
				tango_cache_set_fail_state(ctx, CACHE_ERR_REDIS_CONNECT);
				tango_cache_ctx_destroy(ctx);
			}
			break;
		case CACHE_REDIS_DISCONNECTED:
		case CACHE_REDIS_CONNECT_IDLE:
			wiredlb_redis_asyn_connect(ctx->instance);
		case CACHE_REDIS_CONNECTING:
			tango_cache_set_fail_state(ctx, CACHE_ERR_REDIS_CONNECT);
			tango_cache_ctx_destroy(ctx);
			break;
		default: assert(0);break;
	}
	return ret;
}