tango-tfe/plugin/business/pangu-http/pangu_web_cache.cpp

#include "pangu_web_cache.h"
#include <tango_cache_pending.h>
#include <tango_cache_client.h>

#include <tfe_proxy.h>
#include <tfe_http.h>
#include <tfe_utils.h>


#include <event2/event.h>
#include <event2/buffer.h>


struct cache_handle
{
	unsigned int thread_count;
	struct tango_cache_instance **clients;
	
};
struct cache_update_context
{
	struct tango_cache_ctx * write_ctx;
};
struct cache_handle* create_web_cache_handle(const char* profile_path, const char* section, void *logger)
{
	struct cache_handle* handle=ALLOC(struct cache_handle, 1);
	handle->thread_count=tfe_proxy_get_thread_count();
	handle->clients=ALLOC(struct tango_cache_instance *, handle->thread_count);
	int i=0;
	for(i=0; i<handle->thread_count; i++)
	{
		handle->clients[i]=tango_cache_instance_new(tfe_proxy_get_evbase(i), profile_path, section, logger);
	}
	return handle;
}

static char* read_http1_hdr(const char* hdr, const char* field_name)
{	
	const char *p=NULL, *q=NULL;
	char* value=NULL;
	p=strcasestr(hdr, field_name);
	if(p==NULL)
	{
		return NULL;
	}
	p=strstr(p, ":");
	if(p==NULL)
	{
		return NULL;
	}
	p++;
	q=strcasestr(p, "\r\n");
	if(q==NULL)
	{
		return NULL;
	}
	value=(char*) calloc(sizeof(char), (q-p+1));
	memcpy(value, p, q-p);
	return value;
}
void cache_query_free_meta(struct cached_meta* meta)
{
	FREE(&meta->content_length);
	FREE(&meta->content_type);
	FREE(&meta);
	return;
}
int cache_query_result_is_header(future_result_t * result)
{
	struct tango_cache_result* cache_result=tango_cache_read_result(result);
	if(cache_result->type==RESULT_TYPE_HEADER)
	{
		return 1;
	}
	else
	{
		return 0;
	}
}
int cache_query_result_is_body(future_result_t * result)
{
	struct tango_cache_result* cache_result=tango_cache_read_result(result);
	if(cache_result->type==RESULT_TYPE_BODY)
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

struct cached_meta* cache_query_result_get_header(future_result_t * result)
{
	struct tango_cache_result* cache_result=tango_cache_read_result(result);
	struct cached_meta* meta;
	if(cache_result->type!=RESULT_TYPE_HEADER)
	{
		return NULL;
	}
	meta= ALLOC(struct cached_meta, 1);
	meta->content_length=read_http1_hdr((const char*)cache_result->data_frag, "content-length");
	meta->content_type=read_http1_hdr((const char*)cache_result->data_frag, "content-type");
	return meta;
}
void cache_query_result_append_data(struct evbuffer* buf, future_result_t * result)
{
	struct tango_cache_result* cache_result=tango_cache_read_result(result);
	assert(cache_result->type==RESULT_TYPE_BODY);
	evbuffer_add(buf, cache_result->data_frag, cache_result->size);
	return;
}

enum cache_query_status async_web_cache_query(struct cache_handle* handle, unsigned int thread_id,
										const struct tfe_http_half * request, struct future* f)
{
	struct request_freshness req_fresshness;
	enum cache_pending_action get_action;
	int ret=0;
	get_action=tfe_cache_get_pending(request, &req_fresshness);
	switch(get_action)
	{
		case UNDEFINED:
		case FORBIDDEN:		
			return WEB_CACHE_NOT_APPLICABLE;
		case VERIFY:
		case ALLOWED:
			break;
		default:
			assert(0);
			return WEB_CACHE_NOT_APPLICABLE;
	}
	
	struct tango_cache_meta meta;
	memset(&meta, 0, sizeof(meta));
	meta.url=request->req_spec.url;
	memcpy(&(meta.put), &req_fresshness, sizeof(meta.put));
	ret=tango_cache_fetch_object(handle->clients[thread_id], f, &meta);
	assert(ret==0);
	return WEB_CACHE_QUERING;
}

struct cache_update_context* web_cache_update_start(struct cache_handle* handle, unsigned int thread_id, 
															const struct tfe_http_session * session)
{
	struct cache_update_context* update_ctx=NULL;
	struct response_freshness resp_freshness;
	enum cache_pending_action put_action;
	struct tango_cache_ctx *write_ctx=NULL;
	char buffer[TFE_STRING_MAX];
	const char* value=NULL;
	int i=0;
	put_action=tfe_cache_put_pending(session->resp, &resp_freshness);
	if(put_action!=ALLOWED)
	{
		return NULL;
	}

	struct tango_cache_meta meta;
	memset(&meta, 0, sizeof(meta));
	meta.url=session->req->req_spec.url;
	i=0;
	snprintf(buffer, sizeof(buffer), "content-type:%s",session->resp->resp_spec.content_length);
	meta.std_hdr[i]=buffer;
	i++;
	memcpy(&meta.put, &resp_freshness, sizeof(resp_freshness));
	write_ctx=tango_cache_update_start(handle->clients[thread_id], NULL, &meta);
	if(write_ctx==NULL)
	{
		return NULL;
	}
	
	update_ctx=ALLOC(struct cache_update_context, 1);
	update_ctx->write_ctx=write_ctx;
	return update_ctx;

}
void web_cache_update(struct cache_update_context* ctx, const unsigned char * body_frag, size_t frag_size)
{
	tango_cache_update_frag_data(ctx->write_ctx, (const char*)body_frag, frag_size);
	return;
}
void web_cache_update_end(struct cache_update_context* ctx)
{
	tango_cache_update_end(ctx->write_ctx);
	free(ctx);
	return;
}