#include "lib_acl.h"
#include "conn_cache.h"

/* Ӧĳ IP:PORT ֵӳؽṹͶ */

typedef struct CONN_POOL {
	CONN_CACHE *conn_cache;	/* ĳӳػ */
	ACL_AIO *aio;		/* ӳ첽IO */
	ACL_FIFO conns;		/* ӳеӶ(CONN) */
	char key[256];		/* ӳضӦĴ洢ӳػеĴ洢 */
} CONN_POOL;

/* ӳеĳӶͶ */

struct CONN {
	CONN_POOL *conn_pool;	/* ĵӳض */
	ACL_ASTREAM *stream;	/* ӵ첽 */
	void (*free_fn)(ACL_ASTREAM *stream, void*); /* ͷŸʱĻص */
	void *ctx;		/* free_fn Բ֮һ */
	ACL_FIFO_INFO *info;	/* Ӷ洢ӳ(conn_pool)ĶеĶָ */
};

/* ͷӶռõڴ棬ر */

static void conn_free(CONN *conn)
{
	CONN_POOL *conns = conn->conn_pool;

	/* ȴӳضɾ */
	if (conn->info)
		acl_fifo_delete_info(&conns->conns, conn->info);
	/* ûԶص */
	if (conn->free_fn)
		conn->free_fn(conn->stream, conn->ctx);
	/* ͷڴռ */
	acl_myfree(conn);
}

/* ͷӶ󲢹ر */

static void conn_close(CONN *conn)
{
	/* Ƚֹ첽Ķ */
	acl_aio_disable_read(conn->stream);

	/* 첽رո, Ȼ첽Զ read_close_callback */
	acl_aio_iocp_close(conn->stream);
}

/* ӳضͷʱô˻ص */

static void conn_pool_free(CONN_POOL *conns)
{
	CONN *conn;

	/* ҪӳеӶͷ */

	while ((conn = acl_fifo_pop(&conns->conns)) != NULL) {
		if (conn->stream) {
			if (conn->free_fn)
				conn->free_fn(conn->stream, conn->ctx);
			acl_aio_clean_hooks(conn->stream);
			acl_aio_iocp_close(conn->stream);
		}
		acl_myfree(conn);
	}
	acl_myfree(conns);
}

/* ͷӳضʱص */

static void conn_pool_free_timer(int event_type acl_unused, void *context)
{
	CONN_POOL *conns = (CONN_POOL*) context;

	conn_pool_free(conns);
}

/* ͷӳصĶʱ, ֮ԲöʱͷӳضΪ
 * ʹͷŹ̲ĵݹ鴦бǰͷ
 */

static void set_conn_pool_free_timer(CONN_POOL *conns)
{
	acl_aio_request_timer(conns->aio, conn_pool_free_timer, conns, 1, 1);
}

static void conn_pool_stat_timer(int event_type acl_unused, void *context)
{
	const char *myname = "conn_pool_stat_timer";
	CONN_CACHE *cache = (CONN_CACHE*) context;

	/* only for test */
	if (0)
	acl_msg_info("%s(%d): nset: %d, nget: %d, nclose: %d, inter: %d",
		myname, __LINE__, cache->nset, cache->nget,
		cache->nclose, cache->nset - cache->nget - cache->nclose);
}

static void set_conn_pool_stat_timer(CONN_CACHE *cache)
{
	/* öʱ */
	acl_aio_request_timer(cache->aio, conn_pool_stat_timer, cache, 2, 1);
}

CONN_CACHE *conn_cache_create(ACL_AIO *aio, int conn_limit)
{
	const char *myname = "conn_cache_create";
	CONN_CACHE *cache = (CONN_CACHE*) acl_mycalloc(1, sizeof(CONN_CACHE));

	cache->aio = aio;
	cache->conn_limit = conn_limit;
	cache->cache = acl_htable_create(1024, 0);
	acl_msg_info("%s(%d): ok, conn_limit: %d", myname, __LINE__, conn_limit);

	/* ӳػ״̬ϢĶʱ */
	set_conn_pool_stat_timer(cache);
	return (cache);
}

/* ɶʱĻص */

static int read_callback(ACL_ASTREAM *stream acl_unused, void *ctx acl_unused,
	char *data acl_unused, int dlen acl_unused)
{
	const char *myname = "read_callback";

	acl_msg_info("%s(%d), %s: can read connection from server, dlen(%d), data(%s)",
		__FILE__, __LINE__, myname, dlen, data);

	/* ΪΪӣӦݿɶݿɶΪ
	 * ޷֪δЩݶҪرո
	 */

	/*  -1 Ӷرջص */
	return (-1);
}

/* رʱĻص */

static int read_close_callback(ACL_ASTREAM *stream acl_unused, void *ctx)
{
	CONN *conn = (CONN*) ctx;
	CONN_POOL *conns = conn->conn_pool;

	/* ͷŸӶڴռ䣬رոӣ
	 * رչ첽Զر
	 */
	conn_free(conn);

	/* ӳΪͷŸӳ */
	if (acl_fifo_size(&conns->conns) == 0) {
		acl_htable_delete(conns->conn_cache->cache, conns->key, NULL);
		set_conn_pool_free_timer(conns);
	}

	conns->conn_cache->nclose++;

	/*  acl_aio_iocp_close  */
	return (-1);
}

/* ʱʱĻص */

static int read_timeout_callback(ACL_ASTREAM *stream acl_unused, void *ctx acl_unused)
{
	/*  -1 Ӷرջص */
	return (-1);
}
	
void conn_cache_push_stream(CONN_CACHE *cache, ACL_ASTREAM *stream,
	int timeout, void (*free_fn)(ACL_ASTREAM*, void*), void *ctx)
{
	const char *key = ACL_VSTREAM_PEER(acl_aio_vstream(stream));
	CONN_POOL *conns;
	CONN *conn;

#if 0
	acl_aio_clean_hooks(stream);
#endif

	/* 鿴KEYӳؾǷڣã򴴽µ */

	conns = (CONN_POOL*) acl_htable_find(cache->cache, key);
	if (conns == NULL) {
		conns = (CONN_POOL*) acl_mymalloc(sizeof(CONN_POOL));
		conns->conn_cache = cache;
		conns->aio = acl_aio_handle(stream);
		acl_fifo_init(&conns->conns);
		ACL_SAFE_STRNCPY(conns->key, key, sizeof(conns->key));
		acl_htable_enter(cache->cache, key, conns);
	}

#if 0
	/* ӳеƣͷɵӶ */
	if (acl_fifo_size(&conns->conns) >= cache->conn_limit) {
		conn = acl_fifo_pop(&conns->conns);
		if (conn) {
			conn->info = NULL;
			conn_close(conn);
		}
	}
#endif

	cache->nset++;

	/* µ첽ӻ */
	conn = (CONN*) acl_mymalloc(sizeof(CONN));
	conn->stream = stream;
	conn->ctx = ctx;
	conn->free_fn = free_fn;

	/* ӳ */
	conn->info = acl_fifo_push(&conns->conns, conn);
	conn->conn_pool = conns;

	/* øĻص */
	acl_aio_ctl(stream,
		ACL_AIO_CTL_READ_HOOK_ADD, read_callback, conn,
		ACL_AIO_CTL_CLOSE_HOOK_ADD, read_close_callback, conn,
		ACL_AIO_CTL_TIMEO_HOOK_ADD, read_timeout_callback, conn,
		ACL_AIO_CTL_TIMEOUT, timeout,
		ACL_AIO_CTL_END);
	/* ʼ */
	acl_aio_read(stream);
}

CONN *conn_cache_get_conn(CONN_CACHE *cache, const char *key)
{
	CONN_POOL *conns;
	CONN *conn;

	/* Ȳ鿴KEYӳضǷڣ򷵻NULL */

	conns = (CONN_POOL*) acl_htable_find(cache->cache, key);
	if (conns == NULL) {
		return (NULL);
	}

	/* ӸKEYӳȡһӣȡΪNULLͷŸӳض */

	conn = acl_fifo_pop(&conns->conns);
	if (conn == NULL) {
		/* ȴӳػɾ */
		acl_htable_delete(cache->cache, conns->key, NULL);
		/* ͷſյӳضĶʱ */
		set_conn_pool_free_timer(conns);
		return (NULL);
	}

	/* ȡ֮ǰõĻص */
#if 1
	acl_aio_del_read_hook(conn->stream, read_callback, conn);
	acl_aio_del_close_hook(conn->stream, read_close_callback, conn);
	acl_aio_del_timeo_hook(conn->stream, read_timeout_callback, conn);
#else
	acl_aio_clean_hooks(conn->stream);
#endif

	if (conn->free_fn)
		conn->free_fn(conn->stream, conn->ctx);
	/* ȡ */
	acl_aio_disable_read(conn->stream);

	cache->nget++;
	return (conn);
}

ACL_ASTREAM *conn_cache_get_stream(CONN_CACHE *cache, const char *key, void **ctx_pptr)
{
	CONN *conn;
	ACL_ASTREAM *stream;

	conn = conn_cache_get_conn(cache, key);
	if (conn == NULL) {
		if (ctx_pptr)
			*ctx_pptr = NULL;
		return (NULL);
	}

	if (ctx_pptr)
		*ctx_pptr = conn->ctx;
	stream = conn->stream;
	acl_myfree(conn);  /* ΪѾȡԿͷ CONN  */
	return (stream);
}

void conn_cache_delete_key(CONN_CACHE *cache, const char *key)
{
	CONN_POOL *conns;
	ACL_ITER iter;

	conns = (CONN_POOL*) acl_htable_find(cache->cache, key);
	if (conns == NULL)
		return;

	/* ӳеһһر */
	acl_foreach(iter, &conns->conns) {
		CONN *conn = (CONN*) iter.data;
		if (conn->stream) {
			acl_aio_iocp_close(conn->stream);
			conn->stream = NULL;
		}
	}

	/* ӳشӳػɾ */
	acl_htable_delete(cache->cache, conns->key, NULL);
	/* ͷųضĶʱ */
	set_conn_pool_free_timer(conns);
}

void conn_cache_delete_conn(CONN *conn)
{
	/* ͷŲرոӶ */
	conn_close(conn);
}

void conn_cache_delete_stream(CONN_CACHE *cache, ACL_ASTREAM *stream)
{
	const char *key = ACL_VSTREAM_PEER(acl_aio_vstream(stream));
	CONN_POOL *conns;
	CONN *conn;
	ACL_ITER iter;

	conns = acl_htable_find(cache->cache, key);
	if (conns == NULL)
		return;
	/* ӳضе */
	acl_foreach(iter, &conns->conns) {
		conn = (CONN*) iter.data;
		if (conn->stream == stream) {
			/* ͷŲرոӶ */
			conn_close(conn);
			break;
		}
	}
}
