#include "StdAfx.h"
#ifndef ACL_PREPARE_COMPILE

#include "stdlib/acl_define.h"
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>

#ifdef ACL_BCB_COMPILER
#pragma hdrstop
#endif
#include "stdlib/acl_stdlib.h"
#include "net/acl_net.h"
#include "net/acl_sane_inet.h"
#include "aio/acl_aio.h"

#endif

#include "../event/events_define.h" /* just for ACL_EVENTS_STYLE_IOCP define */
#include "aio.h"

#define	WRITE_SAFE_ENABLE(x, callback) do {  \
	if (((x)->flag & ACL_AIO_FLAG_ISWR) == 0) {  \
		(x)->flag |= ACL_AIO_FLAG_ISWR;  \
		acl_event_enable_write((x)->aio->event, (x)->stream,  \
			(x)->timeout, callback, (x));  \
	}  \
} while (0)

#define WRITE_SAFE_DIABLE(x) do {  \
	if (((x)->flag & ACL_AIO_FLAG_ISWR) != 0) {  \
		(x)->flag &= ~ACL_AIO_FLAG_ISWR;  \
		acl_event_disable_write((x)->aio->event, (x)->stream);  \
	}  \
} while (0)

#ifdef ACL_WINDOWS
static void __connect_notify_callback(int event_type, ACL_EVENT *event,
	ACL_VSTREAM *stream, void *context);

static void ConnectTimer(int event_type acl_unused,
	ACL_EVENT *event acl_unused, void *ctx)
{
	ACL_ASTREAM *conn = (ACL_ASTREAM*) ctx;

	if (conn->aio->event_mode != ACL_EVENT_WMSG) {
		acl_msg_fatal("event_mode(%d) != ACL_EVENT_WMSG(%d)",
			conn->aio->event_mode, ACL_EVENT_WMSG);
	}
	__connect_notify_callback(ACL_EVENT_RW_TIMEOUT, NULL, NULL, ctx);
}
#endif

static void __connect_notify_callback(int event_type, ACL_EVENT *event,
	ACL_VSTREAM *stream acl_unused, void *context)
{
	const char *myname = "__connect_notify_callback";
	ACL_ASTREAM *conn = (ACL_ASTREAM *) context;
	int   ret, err;
	socklen_t errlen;

	WRITE_SAFE_DIABLE(conn);

	/* жǷǳʱ· */
	if ((event_type & ACL_EVENT_RW_TIMEOUT) != 0) {
		if (aio_timeout_callback(conn) < 0) {
			acl_aio_iocp_close(conn);
		} else if (conn->flag & ACL_AIO_FLAG_IOCP_CLOSE) {
			/* IOӳٹر״̬ΪдIOѾɹɣ
			 * ҪIOӳٹرչ
			 */
			acl_aio_iocp_close(conn);
		} else {
			acl_event_enable_write(event, conn->stream,
				conn->timeout, __connect_notify_callback,
				conn);
		}
		return;
	}

#ifdef ACL_WINDOWS
	/* ǻ win32 Ϣ¼Ҫȡ֮ǰõĳʱʱ */
	if (conn->aio->event_mode == ACL_EVENT_WMSG) {
		acl_aio_cancel_timer(conn->aio, ConnectTimer, conn);
	}
#endif

	if ((event_type & ACL_EVENT_XCPT) != 0) {
		acl_aio_iocp_close(conn);
		return;
	}

	errlen = sizeof(err);
	ret = getsockopt(ACL_VSTREAM_SOCK(acl_aio_vstream(conn)),
			SOL_SOCKET, SO_ERROR, (char *) &err, &errlen);
	if (ret >= 0) {
		acl_set_error(err);
	}

#if defined(ACL_SUNOS5)
	/*
	 * Solaris 2.4's socket emulation doesn't allow you
	 * to determine the error from a failed non-blocking
	 * connect and just returns EPIPE.  Create a fake
	 * error message for connect.   -- fenner@parc.xerox.com
	 */
	if (ret < 0 && errno == EPIPE) {
		acl_set_error(ACL_ENOTCONN);
	}
#endif

	if (errno == 0 || errno == ACL_EISCONN) {
		event_type = ACL_EVENT_CONNECT;
	} else if ((event_type & ACL_EVENT_CONNECT) == 0) {
		event_type |= ACL_EVENT_XCPT;
	}

	if ((event_type & ACL_EVENT_XCPT) != 0) {
		conn->flag |= ACL_AIO_FLAG_DEAD;
		acl_aio_iocp_close(conn);
		return;
	}

	if ((event_type & ACL_EVENT_CONNECT) == 0) {
		acl_msg_fatal("%s: unknown event: %d", myname, event_type);
	}

	/* ü1Էֹ connect_fn ڲ˹رչ̣connect_fn
	 * ͨ-1ڻصغر
	 */
	conn->nrefer++;

	if (conn->connect_handles) {
		ACL_ITER iter;
		ACL_FIFO connect_handles;

		/* 뽫صӻصһһһ,
		 * Ϊ ACL_AIO ڻصпܷǶףֹظ
		 */

		acl_fifo_init(&connect_handles);
		acl_foreach_reverse(iter, conn->connect_handles) {
			AIO_CONNECT_HOOK *handle = (AIO_CONNECT_HOOK*) iter.data;
			if (handle->disable) {
				continue;
			}
			acl_fifo_push(&connect_handles, handle);
		}

		while (1) {
			AIO_CONNECT_HOOK *handle = acl_fifo_pop(&connect_handles);
			if (handle == NULL) {
				break;
			}
			ret = handle->callback(conn, handle->ctx);
			if (ret == 0) {
				continue;
			}

			conn->nrefer--;
			if (ret < 0 || conn->flag  & ACL_AIO_FLAG_IOCP_CLOSE) {
				acl_aio_iocp_close(conn);
			}
			return;
		}
	}

	conn->nrefer--;

	if (ret < 0) {
		acl_aio_iocp_close(conn);
	} else if ((conn->flag  & ACL_AIO_FLAG_IOCP_CLOSE)) {
		/* ֮ǰѾIOӳٹرձ־λ
		 * ٴIOӳٹرչ
		 */
		acl_aio_iocp_close(conn);
	}
}

ACL_ASTREAM *acl_aio_connect(ACL_AIO *aio, const char *addr, int timeout)
{
	return acl_aio_connect2(aio, addr, NULL, timeout);
}

ACL_ASTREAM *acl_aio_connect2(ACL_AIO *aio, const char *addr,
	const char *local, int timeout)
{
	const char *myname = "acl_aio_connect";
	ACL_ASTREAM *conn;
	ACL_VSTREAM *cstream;

	if (aio == NULL || addr == NULL || *addr == 0) {
		acl_msg_fatal("%s: input invalid", myname);
	}

#ifdef ACL_EVENTS_STYLE_IOCP
	/* XXX: Windows ƽ̨ݲְ֧󶨱 IP. */
	(void) local;

	if (aio->event_mode == ACL_EVENT_KERNEL) {
		ACL_SOCKET connfd = WSASocket(AF_INET, SOCK_STREAM,
			IPPROTO_TCP, 0, 0, WSA_FLAG_OVERLAPPED);

		cstream = acl_vstream_fdopen(connfd, ACL_VSTREAM_FLAG_RW,
				aio->rbuf_size, timeout, ACL_VSTREAM_TYPE_SOCK);
		acl_assert(cstream);
		acl_vstream_set_peer(cstream, addr);
	} else {
		cstream = acl_vstream_connect(addr, ACL_NON_BLOCKING,
				0, 0, aio->rbuf_size);
	}
#else
	if (local != NULL && (*local == '@' || *local == '#')) {
		ACL_VSTRING *buf = acl_vstring_alloc(128);
		acl_vstring_strcat(buf, addr);
		acl_vstring_strcat(buf, local);
		cstream = acl_vstream_connect(acl_vstring_str(buf), ACL_NON_BLOCKING,
			0, 0, aio->rbuf_size);
		acl_vstring_free(buf);
	} else {
		cstream = acl_vstream_connect(addr, ACL_NON_BLOCKING,
			0, 0, aio->rbuf_size);
	}
#endif

	if (cstream == NULL) {
		acl_msg_error("%s: connect addr(%s) error", myname, addr);
		return NULL;
	}

	cstream->flag |= ACL_VSTREAM_FLAG_CONNECTING;

	conn = acl_aio_open(aio, cstream);
	if (conn == NULL) {
		acl_msg_fatal("%s: open connection error", myname);
	}

#ifdef ACL_WINDOWS
	if (timeout > 0 && aio->event_mode == ACL_EVENT_WMSG) {
		acl_aio_request_timer(aio, ConnectTimer, conn,
			timeout * 1000000, 0);
	}
#endif
	conn->error = acl_last_error();
	acl_aio_ctl(conn, ACL_AIO_CTL_TIMEOUT, timeout, ACL_AIO_CTL_END);

	WRITE_SAFE_ENABLE(conn, __connect_notify_callback);
	return conn;
}

/*--------------------------------------------------------------------------*/

struct ACL_ASTREAM_CTX {
	ACL_SOCKADDR ns_addr;
	ACL_SOCKADDR serv_addr;
	ACL_ASTREAM *conn;
	int status;
	void *ctx;
};

typedef struct {
	ACL_AIO *aio;
	ACL_SOCKADDR ns_addr;
	ACL_SOCKADDR serv_addr;
	ACL_ASTREAM *conn;
	ACL_AIO_CONNECT_ADDR_FN callback;
	void *context;
	int   port;
	int   timeout;
	ACL_ARGV *ip_list;
	int   ip_next;
} RESOLVE_CTX;

static void resolve_ctx_free(RESOLVE_CTX *ctx)
{
	acl_argv_free(ctx->ip_list);
	acl_myfree(ctx);
}

static int connect_callback(ACL_ASTREAM *conn, void *context);
static int connect_timeout(ACL_ASTREAM *conn, void *context);
static ACL_ASTREAM *try_connect_one(RESOLVE_CTX *ctx);

static int connect_failed(ACL_ASTREAM *conn acl_unused, void *context)
{
	RESOLVE_CTX *ctx = (RESOLVE_CTX *) context;
	ACL_ASTREAM_CTX conn_ctx;

	/* Ϊ connect_callback  connect_timeout ˹رջصԵ 
	 * ʱһʧµģӳʱ¡
	 */

	/*  DNS  IP ַһ IP ַ */
	if (try_connect_one(ctx) != NULL) {
		/*  -1 رյǰʱӶµȴӳɹ */
		return -1;
	}

	acl_set_error(ACL_ECONNREFUSED);

	memset(&conn_ctx, 0, sizeof(conn_ctx));
	conn_ctx.conn = NULL;
	conn_ctx.ctx  = ctx->context;
	memcpy(&conn_ctx.ns_addr, &ctx->ns_addr, sizeof(ACL_SOCKADDR));
	memcpy(&conn_ctx.serv_addr, &ctx->serv_addr, sizeof(ACL_SOCKADDR));
	conn_ctx.status = ACL_ASTREAM_STATUS_CONNECT_ERROR;

	ctx->callback(&conn_ctx);
	resolve_ctx_free(ctx);
	return -1;
}

static int connect_timeout(ACL_ASTREAM *conn, void *context)
{
	RESOLVE_CTX *ctx = (RESOLVE_CTX *) context;
	ACL_ASTREAM_CTX conn_ctx;

	 /*  acl aio ƣʱصҷ -1 ʱע
	  * رջصŻᱻãͨڴ˴첽עĹ
	  * ջصӶֹ connect_failed ٱá
	  */
	acl_aio_del_close_hook(conn, connect_failed, context);

	/*  DNS  IP ַһ IP ַ */
	if (try_connect_one(ctx) != NULL) {
		/*  -1 رյǰʱӶµȴӳɹ */
		return -1;
	}

	acl_set_error(ACL_ETIMEDOUT);

	memset(&conn_ctx, 0, sizeof(conn_ctx));
	conn_ctx.conn = NULL;
	conn_ctx.ctx  = ctx->context;
	memcpy(&conn_ctx.ns_addr, &ctx->ns_addr, sizeof(ACL_SOCKADDR));
	memcpy(&conn_ctx.serv_addr, &ctx->serv_addr, sizeof(ACL_SOCKADDR));
	conn_ctx.status = ACL_ASTREAM_STATUS_CONNECT_TIMEOUT;

	ctx->callback(&conn_ctx);
	resolve_ctx_free(ctx);
	return -1;
}

static int connect_callback(ACL_ASTREAM *conn, void *context)
{
	RESOLVE_CTX *ctx = (RESOLVE_CTX *) context;
	ACL_ASTREAM_CTX conn_ctx;

	/* ʱעĻص IO ʱرʱ
	 * ֻصӦעĻص
	 */
	acl_aio_del_connect_hook(conn, connect_callback, context);
	acl_aio_del_timeo_hook(conn, connect_timeout, context);
	acl_aio_del_close_hook(conn, connect_failed, context);

	acl_set_error(0);

	memset(&conn_ctx, 0, sizeof(conn_ctx));
	conn_ctx.conn = conn;
	conn_ctx.ctx  = ctx->context;
	memcpy(&conn_ctx.ns_addr, &ctx->ns_addr, sizeof(ACL_SOCKADDR));
	memcpy(&conn_ctx.serv_addr, &ctx->serv_addr, sizeof(ACL_SOCKADDR));
	conn_ctx.status = ACL_ASTREAM_STATUS_OK;

	ctx->callback(&conn_ctx);
	resolve_ctx_free(ctx);
	return 0;
}

static ACL_ASTREAM *try_connect_one(RESOLVE_CTX *ctx)
{
	int n = acl_argv_size(ctx->ip_list);

	while (ctx->ip_next < n) {
		char  addr[128];
		const char *ip = acl_argv_index(ctx->ip_list,  ctx->ip_next);
		ACL_SOCKADDR sa;

		ctx->ip_next++;
		acl_assert(ip && *ip);
		snprintf(addr, sizeof(addr), "%s|%d", ip, ctx->port);
		if (acl_sane_pton(addr, (struct sockaddr *)&sa) == 0) {
			continue;
		}

		memcpy(&ctx->serv_addr, &sa, sizeof(sa));
		ctx->conn = acl_aio_connect(ctx->aio, addr, ctx->timeout);
		if (ctx->conn == NULL) {
			continue;
		}

		acl_aio_add_connect_hook(ctx->conn, connect_callback, ctx);
		acl_aio_add_timeo_hook(ctx->conn, connect_timeout, ctx);
		acl_aio_add_close_hook(ctx->conn, connect_failed, ctx);
		return ctx->conn;
	}

	return NULL;
}

static void dns_lookup_callback(ACL_DNS_DB *db, void *context, int errnum,
	const ACL_RFC1035_MESSAGE *msg)
{
	RESOLVE_CTX *ctx  = (RESOLVE_CTX *) context;
	ACL_ITER     iter;
	ACL_ASTREAM_CTX conn_ctx;

	(void) msg;
	memset(&conn_ctx, 0, sizeof(conn_ctx));
	conn_ctx.ctx = ctx->context;

	if (db == NULL) {
		acl_set_error(errnum);
		conn_ctx.status = ACL_ASTREAM_STATUS_NS_ERROR;
		ctx->callback(&conn_ctx);
		resolve_ctx_free(ctx);
		return;
	}

	memcpy(&conn_ctx.ns_addr, &db->ns_addr, sizeof(conn_ctx.ns_addr));
	memcpy(&ctx->ns_addr, &db->ns_addr, sizeof(ctx->ns_addr));

	acl_foreach(iter, db) {
		const ACL_HOST_INFO *info = (const ACL_HOST_INFO *) iter.data;
		acl_argv_add(ctx->ip_list, info->ip, NULL);
	}

	if (acl_argv_size(ctx->ip_list) <= 0) {
		acl_set_error(errnum);
		conn_ctx.status = ACL_ASTREAM_STATUS_NS_ERROR;
		ctx->callback(&conn_ctx);
		resolve_ctx_free(ctx);
		return;
	}

	if (try_connect_one(ctx) == NULL) {
		acl_set_error(ACL_ECONNREFUSED);
		conn_ctx.status = ACL_ASTREAM_STATUS_CONNECT_ERROR;
		ctx->callback(&conn_ctx);
		resolve_ctx_free(ctx);
	}
}

int acl_aio_connect_addr(ACL_AIO *aio, const char *addr, int timeout,
	 ACL_AIO_CONNECT_ADDR_FN callback, void *context)
{
	return acl_aio_connect_addr2(aio, addr, NULL, timeout, callback, context);
}

int acl_aio_connect_addr2(ACL_AIO *aio, const char *addr,
	const char *local, int timeout,
	ACL_AIO_CONNECT_ADDR_FN callback, void *context)
{
	char buf[128], *ptr;
	int  port;
	RESOLVE_CTX *ctx;

	ACL_SAFE_STRNCPY(buf, addr, sizeof(buf));
	ptr = strrchr(buf, '|');
	if (ptr == NULL) {
		ptr = strrchr(buf, ':');
	}
	if (ptr == NULL || *(ptr + 1) == 0) {
		acl_msg_error("%s(%d), %s: invalid addr=%s",
			__FILE__, __LINE__, __FUNCTION__, addr);
		return -1;
	}
	*ptr++ = 0;
	port = atoi(ptr);
	if (port <= 0 || port > 65535) {
		acl_msg_error("%s(%d), %s: invalid port=%d, addr=%s",
			__FILE__, __LINE__, __FUNCTION__, port, addr);
		return -1;
	}

	ctx = (RESOLVE_CTX *) acl_mycalloc(1, sizeof(RESOLVE_CTX));
	ctx->aio      = aio;
	ctx->callback = callback;
	ctx->context  = context;
	ctx->port     = port;
	ctx->timeout  = timeout;
	ctx->ip_list  = acl_argv_alloc(5);

	if (acl_is_ip(buf) || acl_valid_unix(buf)) {
		ACL_SOCKADDR sa;
		ACL_ASTREAM *conn;

		if (acl_sane_pton(addr, (struct sockaddr *)&sa) == 0) {
			acl_msg_error("%s(%d): invalid addr=%s",
				__FUNCTION__, __LINE__, addr);
			resolve_ctx_free(ctx);
			return -1;
		}

		conn = acl_aio_connect2(aio, addr, local, timeout);
		if (conn == NULL) {
			resolve_ctx_free(ctx);
			return -1;
		}

		memcpy(&ctx->serv_addr, &sa, sizeof(ACL_SOCKADDR));

		acl_aio_add_connect_hook(conn, connect_callback, ctx);
		acl_aio_add_timeo_hook(conn, connect_timeout, ctx);
		acl_aio_add_close_hook(conn, connect_failed, ctx);

		return 0;
	} else if (aio->dns == NULL) {
		acl_msg_error("%s(%d), %s: call acl_aio_set_dns first",
			__FILE__, __LINE__, __FUNCTION__);
		return -1;
	} else {
		/* XXX: Not support binding local IP and interface now. */
		acl_dns_lookup(aio->dns, buf, dns_lookup_callback, ctx);
		return 0;
	}
}

int acl_astream_get_status(const ACL_ASTREAM_CTX *ctx)
{
	return ctx ? ctx->status : ACL_ASTREAM_STATUS_INVALID;
}

const ACL_SOCKADDR *acl_astream_get_ns_addr(const ACL_ASTREAM_CTX *ctx)
{
	return ctx ? &ctx->ns_addr : NULL;
}

const ACL_SOCKADDR *acl_astream_get_serv_addr(const ACL_ASTREAM_CTX *ctx)
{
	return ctx ? &ctx->serv_addr : NULL;
}

ACL_ASTREAM *acl_astream_get_conn(const ACL_ASTREAM_CTX *ctx)
{
	return ctx ? ctx->conn : NULL;
}

void *acl_astream_get_ctx(const ACL_ASTREAM_CTX *ctx)
{
	return ctx ? ctx->ctx : NULL;
}
