tango-tfe/platform/src/tcp_stream.cpp

#include <netinet/in.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <assert.h>

#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/bufferevent_ssl.h>
#include <event2/buffer.h>
#include <event2/thread.h>
#include <event2/dns.h>

#include <openssl/ssl.h>
#include <openssl/err.h>
#include <openssl/rand.h>
#include <openssl/x509.h>
#include <openssl/x509v3.h>

#include <tfe_stream.h>
#include <tfe_utils.h>
#include <tfe_future.h>
#include <tfe_plugin.h>

#include <platform.h>
#include <ssl_stream.h>
#include <tcp_stream.h>
#include <proxy.h>
#include <netinet/tcp.h>

#ifndef TFE_CONFIG_OUTPUT_LIMIT_DEFAULT
#define TFE_CONFIG_OUTPUT_LIMIT_DEFAULT                (1024 * 1024)
#endif

/* forward declaration of libevent callbacks */
static void __stream_bev_readcb(struct bufferevent *, void *);
static void __stream_bev_writecb(struct bufferevent *, void *);
static void __stream_bev_eventcb(struct bufferevent *, short, void *);

/* ====================================================================================================================
 * HELPER FUNCTIONS
 * ===================================================================================================================*/

static inline struct tfe_stream_private * to_stream_private(const struct tfe_stream * stream)
{
	return container_of(stream, struct tfe_stream_private, head);
}

static inline struct tfe_conn_private * __this_conn(struct tfe_stream_private * _stream, enum tfe_conn_dir dir)
{
	return ((dir == CONN_DIR_DOWNSTREAM) ? (_stream->conn_downstream) : (_stream->conn_upstream));
}

static inline struct tfe_conn_private * __peer_conn(struct tfe_stream_private * _stream, enum tfe_conn_dir dir)
{
	return ((dir == CONN_DIR_DOWNSTREAM) ? (_stream->conn_upstream) : (_stream->conn_downstream));
}

static inline enum tfe_conn_dir __bev_dir(struct tfe_stream_private * _stream, struct bufferevent * bev)
{
	if (_stream->conn_downstream && bev == _stream->conn_downstream->bev)
	{
		return CONN_DIR_DOWNSTREAM;
	}

	if (_stream->conn_upstream && bev == _stream->conn_upstream->bev)
	{
		return CONN_DIR_UPSTREAM;
	}

	assert(0);
}

static inline bool __is_ssl(struct tfe_stream_private * _stream)
{
	return (_stream->session_type == STREAM_PROTO_SSL);
}

static void call_plugin_close(struct tfe_stream_private * _stream)
{
	unsigned int plugin_id_iter = 0;
	unsigned int plugin_id = 0;

	for (const struct tfe_plugin * p_info_iter = tfe_plugin_iterate(&plugin_id_iter);
		 p_info_iter != NULL; p_info_iter = tfe_plugin_iterate(&plugin_id_iter))
	{
		_stream->calling_idx = plugin_id;
		struct plugin_ctx * plug_ctx = &_stream->plugin_ctxs[plugin_id];

		/* TODO: do not use pme to determinate we call on_open or not ever. */
		if (p_info_iter->on_close && plug_ctx->pme != NULL)
		{
			p_info_iter->on_close(&_stream->head, _stream->thread_ref->thread_id,
				REASON_PASSIVE_CLOSED, &(plug_ctx->pme));
		}

		plugin_id++;
	}
}

/* ====================================================================================================================
 * INTERFACE
 * ===================================================================================================================*/

void tfe_stream_detach(const struct tfe_stream * stream)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);
	int plug_id = _stream->calling_idx;
	_stream->plugin_ctxs[plug_id].state = PLUG_STATE_DETACHED;
}

int tfe_stream_preempt(const struct tfe_stream * stream)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);
	int plug_id = _stream->calling_idx;
	for (unsigned int i = 0; i < _stream->nr_plugin_ctxs; i++)
	{
		if (_stream->plugin_ctxs[i].state == PLUG_STATE_PREEPTION)
		{
			return -1;
		}
	}
	_stream->plugin_ctxs[plug_id].state = PLUG_STATE_PREEPTION;
	return 0;
}

void tfe_stream_suspend(const struct tfe_stream * stream, enum tfe_conn_dir by)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);
	assert(_stream != NULL);
	assert(_stream->conn_upstream != NULL && _stream->conn_downstream != NULL);
	assert(_stream->conn_upstream->bev != NULL);
	assert(_stream->conn_downstream->bev != NULL);

	/* stream cannot be suspended twice or more */
	assert(!_stream->is_suspended);
	_stream->is_suspended = true;
	_stream->suspended_by = by;

	/* disable all events */
	int ret = 0;
	ret = bufferevent_disable(_stream->conn_upstream->bev, EV_READ | EV_WRITE);
	assert(ret == 0);

	ret = bufferevent_disable(_stream->conn_downstream->bev, EV_READ | EV_WRITE);
	assert(ret == 0);

	(void) ret;
}

void tfe_stream_resume(const struct tfe_stream * stream)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);
	assert(_stream->is_suspended);

	bufferevent_enable(_stream->conn_upstream->bev, EV_READ | EV_WRITE);
	bufferevent_enable(_stream->conn_downstream->bev, EV_READ | EV_WRITE);

	if (_stream->suspended_by == CONN_DIR_DOWNSTREAM)
	{
		bufferevent_trigger(_stream->conn_downstream->bev, EV_READ, BEV_OPT_DEFER_CALLBACKS);
	}
	else
	{
		bufferevent_trigger(_stream->conn_upstream->bev, EV_READ, BEV_OPT_DEFER_CALLBACKS);
	}

	_stream->is_suspended = false;
	_stream->suspended_by = CONN_DIR_DOWNSTREAM;
}

struct tfe_stream_write_ctx * tfe_stream_write_frag_start(const struct tfe_stream * stream, enum tfe_conn_dir dir)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);
	struct tfe_stream_write_ctx ** ref_write_ctx = NULL;
	struct tfe_conn_private * this_conn = NULL;
	struct tfe_conn_private * peer_conn = NULL;

	if (dir == CONN_DIR_DOWNSTREAM)
	{
		this_conn = _stream->conn_downstream;
		peer_conn = _stream->conn_upstream;
		ref_write_ctx = &_stream->w_ctx_downstream;
	}
	else
	{
		this_conn = _stream->conn_upstream;
		peer_conn = _stream->conn_downstream;
		ref_write_ctx = &_stream->w_ctx_upstream;
	}

	if (*ref_write_ctx != NULL)
		return NULL;

	this_conn->on_writing = 1;
	*ref_write_ctx = ALLOC(struct tfe_stream_write_ctx, 1);
	(*ref_write_ctx)->_stream = _stream;
	(*ref_write_ctx)->dir = dir;

	bufferevent_disable(peer_conn->bev, EV_READ);
	return *ref_write_ctx;
}

int tfe_stream_write_frag(struct tfe_stream_write_ctx * w_ctx, const unsigned char * data, size_t size)
{
	struct tfe_conn_private * this_conn = __this_conn(w_ctx->_stream, w_ctx->dir);
	int ret = 0;
	if (this_conn != NULL)
	{
		ret = bufferevent_write(this_conn->bev, data, size);
	}
	else
	{
		return -EPIPE;
	}

	return ret;
}

void tfe_stream_write_frag_end(struct tfe_stream_write_ctx * w_ctx)
{
	struct tfe_conn_private * this_conn = __this_conn(w_ctx->_stream, w_ctx->dir);
	struct tfe_conn_private * peer_conn = __peer_conn(w_ctx->_stream, w_ctx->dir);

	if (this_conn != NULL)
	{
		this_conn->on_writing = 0;
		bufferevent_enable(peer_conn->bev, EV_READ);
	}

	if (w_ctx->_stream != NULL)
	{
		if (w_ctx->dir == CONN_DIR_DOWNSTREAM)
		{
			assert(w_ctx->_stream->w_ctx_downstream == w_ctx);
			w_ctx->_stream->w_ctx_downstream = NULL;
		}
		else
		{
			assert(w_ctx->_stream->w_ctx_upstream == w_ctx);
			w_ctx->_stream->w_ctx_upstream = NULL;
		}
	}

	free(w_ctx);
}

int tfe_stream_write(const struct tfe_stream * stream, enum tfe_conn_dir dir, const unsigned char * data, size_t size)
{
	int ret = 0;
	struct tfe_stream_write_ctx * wctx = tfe_stream_write_frag_start(stream, dir);
	ret = tfe_stream_write_frag(wctx, data, size);
	tfe_stream_write_frag_end(wctx);
	return ret;
}

static tfe_conn_private * __conn_private_create_by_bev(struct tfe_stream_private * stream, struct bufferevent * bev)
{
	struct tfe_conn_private * __conn_private = ALLOC(struct tfe_conn_private, 1);
	__conn_private->bev = bev;
	__conn_private->fd = bufferevent_getfd(bev);

	bufferevent_setcb(__conn_private->bev, __stream_bev_readcb, __stream_bev_writecb, __stream_bev_eventcb, stream);
	bufferevent_enable(__conn_private->bev, EV_READ | EV_WRITE);

	return __conn_private;
}

int tfe_stream_action_set_opt(const struct tfe_stream * stream, enum tfe_stream_action_opt type,
	void * value, size_t size)
{
	struct tfe_stream_private * _stream = to_stream_private(stream);

#define __SAVE_PARAM(what) do {                                        \
    if(size != sizeof(__typeof__(what))) { return -EINVAL; }        \
    else { what = *(__typeof(what) *)value;    } } while(0)            \

	switch (type)
	{
		case ACTION_OPT_FOWARD_BYTES: __SAVE_PARAM(_stream->forward_bytes);
			break;
		case ACTION_OPT_DEFER_BYTES: __SAVE_PARAM(_stream->defer_bytes);
			break;
		case ACTION_OPT_DEFER_TIME_TV: __SAVE_PARAM(_stream->defer_timeval);
			break;
		case ACTION_OPT_DROP_BYTES: __SAVE_PARAM(_stream->drop_bytes);
			break;
	}

#undef __SAVE_PARAM
	return 0;
}

/* ====================================================================================================================
 * CONNECTION STRUCTURE AND OPERATION FUCTIONS
 * ===================================================================================================================*/

evutil_socket_t __conn_private_release_fd(struct tfe_conn_private * conn)
{
	evutil_socket_t __to_release_fd = conn->fd;
	conn->fd = 0;
	return __to_release_fd;
}

static void __conn_private_destory(struct tfe_conn_private * conn)
{
	bufferevent_disable(conn->bev, EV_READ | EV_WRITE);
	bufferevent_free(conn->bev);

	if (conn->fd > 0) evutil_closesocket(conn->fd);
	free(conn);

	TFE_PROXY_STAT_INCREASE(STAT_FD_INSTANT_CLOSE, 1);
}

static void __conn_private_destory_with_ssl(struct event_base * ev_base,
	struct tfe_conn_private * conn, struct ssl_stream * ssl_stream)
{
	if (ssl_stream == NULL) return __conn_private_destory(conn);
	evutil_socket_t __to_closed_fd = __conn_private_release_fd(conn);
	ssl_stream_free_and_close_fd(ssl_stream, ev_base, __to_closed_fd);
	return __conn_private_destory(conn);
}

static void __stream_bev_downstream_statcb(struct evbuffer * buffer, const struct evbuffer_cb_info * info, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
}

static void __stream_bev_upstream_statcb(struct evbuffer * buffer, const struct evbuffer_cb_info * info, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
}

static void __stream_bev_passthrough_readcb(struct bufferevent * bev, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	struct tfe_conn_private * peer_conn = __peer_conn(_stream, __bev_dir(_stream, bev));

	struct evbuffer * __input_buffer = bufferevent_get_input(bev);
	if (peer_conn == NULL)
	{
		evbuffer_drain(__input_buffer, evbuffer_get_length(__input_buffer));
		return;
	}

	struct evbuffer * __output_buffer = bufferevent_get_output(peer_conn->bev);
	evbuffer_add_buffer(__output_buffer, __input_buffer);
}

static void __stream_bev_passthrough_writecb(struct bufferevent * bev, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	struct tfe_conn_private ** ref_this_conn{};
	struct tfe_conn_private ** ref_peer_conn{};

	if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
	{
		ref_this_conn = &_stream->conn_upstream;
		ref_peer_conn = &_stream->conn_downstream;
	}

	if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
	{
		ref_this_conn = &_stream->conn_downstream;
		ref_peer_conn = &_stream->conn_upstream;
	}

	struct evbuffer * __output_buffer = bufferevent_get_output(bev);
	assert(__output_buffer != NULL);

	if (*ref_peer_conn == NULL && evbuffer_get_length(__output_buffer) == 0)
	{
		__conn_private_destory(*ref_this_conn);
		*ref_this_conn = NULL;
	}

	if (*ref_peer_conn == NULL && *ref_this_conn == NULL)
	{
		call_plugin_close(_stream);
		tfe_stream_destory(_stream);
	}

	return;
}

static void __stream_bev_passthrough_eventcb(struct bufferevent * bev, short events, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	struct tfe_conn_private ** ref_this_conn{};
	struct tfe_conn_private ** ref_peer_conn{};

	if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
	{
		ref_this_conn = &_stream->conn_upstream;
		ref_peer_conn = &_stream->conn_downstream;
	}

	if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
	{
		ref_this_conn = &_stream->conn_downstream;
		ref_peer_conn = &_stream->conn_upstream;
	}

	if (events & BEV_EVENT_ERROR || events & BEV_EVENT_EOF)
	{
		if (evbuffer_get_length(bufferevent_get_input(bev)))
		{
			__stream_bev_passthrough_readcb(bev, arg);
		}

		goto __close_connection;
	}

	return;

__close_connection:
	if (*ref_peer_conn != NULL)
	{
		struct bufferevent * __peer_bev = (*ref_peer_conn)->bev;
		struct evbuffer * __peer_output_buffer = bufferevent_get_output(__peer_bev);

		if (evbuffer_get_length(__peer_output_buffer) == 0)
		{
			__conn_private_destory(*ref_peer_conn);
			*ref_peer_conn = NULL;
		}
	}

	if (*ref_this_conn != NULL)
	{
		__conn_private_destory(*ref_this_conn);
		*ref_this_conn = NULL;
	}

	if (*ref_this_conn == NULL && *ref_peer_conn == NULL)
	{
		call_plugin_close(_stream);
		tfe_stream_destory(_stream);
	}

	return;
}

/*
 * Callback for read events on the up- and downstream connection bufferevents.
 * Called when there is data ready in the input evbuffer.
 */

static void __stream_bev_readcb(struct bufferevent * bev, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	enum tfe_conn_dir dir = __bev_dir(_stream, bev);
	struct tfe_conn_private * peer_conn = __peer_conn(_stream, dir);

	/* Peer connection is terminated, drain all data.
	 * This connection will be destoryed in __event_cb */
	struct evbuffer * inbuf = bufferevent_get_input(bev);
	if (peer_conn == NULL)
	{
		evbuffer_drain(inbuf, evbuffer_get_length(inbuf));
		return;
	}

	struct evbuffer * outbuf = bufferevent_get_output(peer_conn->bev);
	assert(inbuf != NULL && outbuf != NULL);

	enum tfe_stream_action action_tmp = ACTION_FORWARD_DATA;
	enum tfe_stream_action action_final = ACTION_FORWARD_DATA;

	size_t drain_size = 0;
	size_t contigous_len = evbuffer_get_length(inbuf);
	unsigned char * contiguous_data = evbuffer_pullup(inbuf, contigous_len);

	_stream->defer_bytes = 0;
	_stream->drop_bytes = 0;
	_stream->forward_bytes = 0;

	unsigned int plugin_id_iter = 0;
	unsigned int plugin_id = 0;

	for (const struct tfe_plugin * p_info_iter = tfe_plugin_iterate(&plugin_id_iter);
		 p_info_iter != NULL; p_info_iter = tfe_plugin_iterate(&plugin_id_iter))
	{
		_stream->calling_idx = plugin_id;
		struct plugin_ctx * plug_ctx = &_stream->plugin_ctxs[plugin_id];

		if (p_info_iter->on_open != NULL && plug_ctx->is_plugin_opened == 0)
		{
			p_info_iter->on_open(&_stream->head, _stream->thread_ref->thread_id, dir, &(plug_ctx->pme));
			plug_ctx->is_plugin_opened = 1;
		}

		if (p_info_iter->on_data != NULL)
		{
			action_tmp = p_info_iter->on_data(&_stream->head, _stream->thread_ref->thread_id,
				dir, contiguous_data, contigous_len, &(plug_ctx->pme));
		}

		if (plug_ctx->state == PLUG_STATE_PREEPTION)
		{
			action_final = action_tmp;
		}

		plugin_id++;
	}

	switch (action_final)
	{
		case ACTION_FORWARD_DATA:
			if (_stream->forward_bytes > 0)
			{
				evbuffer_remove_buffer(inbuf, outbuf, _stream->forward_bytes);
			}
			else
			{
				evbuffer_add_buffer(outbuf, inbuf);
			}
			break;

		case ACTION_DROP_DATA:
			if (_stream->drop_bytes > 0)
			{
				drain_size = _stream->drop_bytes;
			}
			else
			{
				drain_size = evbuffer_get_length(inbuf);
			}
			evbuffer_drain(inbuf, drain_size);
			break;

		case ACTION_DEFER_DATA:
			if (_stream->defer_bytes > 0)
			{
				bufferevent_setwatermark(bev, EV_WRITE, _stream->defer_bytes, 0);
			}
			break;
		default: assert(0);
			break;
	}

#if 0
    if (evbuffer_get_length(outbuf) >= TFE_CONFIG_OUTPUT_LIMIT_DEFAULT)
    {
        bufferevent_setwatermark(peer_conn->bev, EV_WRITE, TFE_CONFIG_OUTPUT_LIMIT_DEFAULT / 2, TFE_CONFIG_OUTPUT_LIMIT_DEFAULT);
        bufferevent_disable(bev, EV_READ);
    }
#endif
}

/*
 * Callback for write events on the up- and downstream connection bufferevents.
 * Called when either all data from the output evbuffer has been written,
 * or if the outbuf is only half full again after having been full.
 */
static void __stream_bev_writecb(struct bufferevent * bev, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	struct event_base * ev_base = bufferevent_get_base(bev);

	struct tfe_conn_private ** ref_this_conn{};
	struct tfe_conn_private ** ref_peer_conn{};
	struct ssl_stream ** ref_this_ssl_stream{};

	const char * __str_dir = NULL;
	if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
	{
		ref_this_conn = &_stream->conn_upstream;
		ref_this_ssl_stream = &_stream->ssl_upstream;
		ref_peer_conn = &_stream->conn_downstream;
	}

	if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
	{
		ref_this_conn = &_stream->conn_downstream;
		ref_this_ssl_stream = &_stream->ssl_downstream;
		ref_peer_conn = &_stream->conn_upstream;
	}

	struct evbuffer * __output_buffer = bufferevent_get_output(bev);
	assert(__output_buffer != NULL);

	if (*ref_peer_conn == NULL /* Peer connection is closed */
		&& (*ref_this_conn)->on_writing == 0 /* No body is prepare to write data, eg. No body call stream_write */
		&& evbuffer_get_length(__output_buffer) == 0) /* Nothing is in send queue */
	{
		TFE_PROXY_STAT_INCREASE(STAT_FD_CLOSE_BY_EVENT_WRITE, 1);

		__conn_private_destory_with_ssl(ev_base, *ref_this_conn, *ref_this_ssl_stream);
		*ref_this_conn = NULL;
		*ref_this_ssl_stream = NULL;
	}

	if (*ref_peer_conn == NULL && *ref_this_conn == NULL)
	{
		call_plugin_close(_stream);
		tfe_stream_destory(_stream);
	}
}

/*
 * Callback for meta events on the up- and downstream connection bufferevents.
 * Called when EOF has been reached, a connection has been made, and on errors.
 */
static void __stream_bev_eventcb(struct bufferevent * bev, short events, void * arg)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) arg;
	struct event_base * ev_base = bufferevent_get_base(bev);

	struct tfe_conn_private ** ref_this_conn{};
	struct tfe_conn_private ** ref_peer_conn{};
	struct ssl_stream ** ref_this_ssl_stream{};
	struct ssl_stream ** ref_peer_ssl_stream{};

	const char * __str_dir = NULL;
	if (__bev_dir(_stream, bev) == CONN_DIR_UPSTREAM)
	{
		ref_this_conn = &_stream->conn_upstream;
		ref_peer_conn = &_stream->conn_downstream;
		ref_this_ssl_stream = &_stream->ssl_upstream;
		ref_peer_ssl_stream = &_stream->ssl_downstream;
	}

	if (__bev_dir(_stream, bev) == CONN_DIR_DOWNSTREAM)
	{
		ref_this_conn = &_stream->conn_downstream;
		ref_peer_conn = &_stream->conn_upstream;
		ref_this_ssl_stream = &_stream->ssl_downstream;
		ref_peer_ssl_stream = &_stream->ssl_upstream;
	}

	if (events & BEV_EVENT_ERROR || events & BEV_EVENT_EOF)
	{
		if (evbuffer_get_length(bufferevent_get_input(bev)))
		{
			__stream_bev_readcb(bev, arg);
		}

		if (events & BEV_EVENT_ERROR) { TFE_PROXY_STAT_INCREASE(STAT_FD_CLOSE_BY_EVENT_ERROR, 1); }
		if (events & BEV_EVENT_EOF) { TFE_PROXY_STAT_INCREASE(STAT_FD_CLOSE_BY_EVENT_EOF, 1); }

		goto __close_connection;
	}

	return;

__close_connection:
	if (*ref_peer_conn != NULL)
	{
		struct bufferevent * __peer_bev = (*ref_peer_conn)->bev;
		struct evbuffer * __peer_output_buffer = bufferevent_get_output(__peer_bev);

		if (evbuffer_get_length(__peer_output_buffer) == 0 && (*ref_peer_conn)->on_writing == 0)
		{
			__conn_private_destory_with_ssl(ev_base, *ref_peer_conn, *ref_peer_ssl_stream);
			*ref_peer_conn = NULL;
			*ref_peer_ssl_stream = NULL;
		}
	}

	if (*ref_this_conn != NULL)
	{
		__conn_private_destory_with_ssl(ev_base, *ref_this_conn, *ref_this_ssl_stream);
		*ref_this_conn = NULL;
		*ref_this_ssl_stream = NULL;
	}

	if (*ref_this_conn == NULL && *ref_peer_conn == NULL)
	{
		goto __call_plugin_close;
	}

	return;

__call_plugin_close:
	call_plugin_close(_stream);
	tfe_stream_destory(_stream);
}

static tfe_conn_private * __conn_private_create_by_fd(struct tfe_stream_private * stream, evutil_socket_t fd)
{
	struct tfe_conn_private * __conn_private = ALLOC(struct tfe_conn_private, 1);
	struct event_base * __ev_base = stream->thread_ref->evbase;

	__conn_private->_stream_ref = stream;
	__conn_private->bev = bufferevent_socket_new(__ev_base, fd, BEV_OPT_DEFER_CALLBACKS);
	__conn_private->fd = fd;

	if (!__conn_private->bev)
	{
		TFE_LOG_ERROR(__STREAM_LOGGER(stream), "Failed at creating bufferevent for fd %d", fd);
		goto __errout;
	}

	if (stream->passthough)
	{
		bufferevent_setcb(__conn_private->bev, __stream_bev_passthrough_readcb,
			__stream_bev_passthrough_writecb, __stream_bev_passthrough_eventcb, stream);
	}
	else
	{
		bufferevent_setcb(__conn_private->bev, __stream_bev_readcb,
			__stream_bev_writecb, __stream_bev_eventcb, stream);
	}

	return __conn_private;

__errout:
	if (__conn_private != NULL) free(__conn_private);
	return NULL;
}

void __conn_private_enable(struct tfe_conn_private * conn_private)
{
	assert(conn_private != NULL && conn_private->bev != NULL);
	bufferevent_enable(conn_private->bev, EV_READ | EV_WRITE);
}

void ssl_downstream_create_on_success(future_result_t * result, void * user)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
	struct ssl_stream * downstream = ssl_downstream_create_result_release_stream(result);
	struct bufferevent * bev = ssl_downstream_create_result_release_bev(result);

	_stream->conn_downstream = __conn_private_create_by_bev(_stream, bev);
	_stream->ssl_downstream = downstream;

	future_destroy(_stream->future_downstream_create);
	_stream->future_downstream_create = NULL;
	_stream->defer_fd_downstream = 0;

	assert(_stream->conn_downstream != NULL && _stream->conn_upstream != NULL);
	__conn_private_enable(_stream->conn_downstream);
	__conn_private_enable(_stream->conn_upstream);

	return;
}

void ssl_downstream_create_on_fail(enum e_future_error err, const char * what, void * user)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
	assert(_stream != NULL && _stream->session_type == STREAM_PROTO_SSL);

	TFE_STREAM_LOG_ERROR(_stream, "Failed to create SSL downstream, close the connection : %s. ", what);

	/* There is nothing we can do because upstream has been handshake,
	 * Close the stream */
	tfe_stream_destory(_stream);
}

void ssl_upstream_create_on_success(future_result_t * result, void * user)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
	struct event_base * ev_base = _stream->thread_ref->evbase;

	struct ssl_stream * upstream = ssl_upstream_create_result_release_stream(result);
	struct bufferevent * bev = ssl_upstream_create_result_release_bev(result);
	assert(upstream != NULL && bev != NULL);

	/* Create connection ctx by bev */
	_stream->conn_upstream = __conn_private_create_by_bev(_stream, bev);
	_stream->ssl_upstream = upstream;

	assert(_stream->conn_upstream != NULL);
	assert(_stream->ssl_upstream != NULL);

	future_destroy(_stream->future_upstream_create);
	_stream->future_upstream_create = NULL;
	_stream->defer_fd_upstream = 0;

	/* Next, create downstream */
	_stream->future_downstream_create = future_create("ssl_down", ssl_downstream_create_on_success,
		ssl_downstream_create_on_fail, _stream);

	ssl_async_downstream_create(_stream->future_downstream_create, _stream->ssl_mgr,
		_stream->ssl_upstream, _stream->defer_fd_downstream, _stream->keyring_id, ev_base);
}

void ssl_upstream_create_on_fail(enum e_future_error err, const char * what, void * user)
{
	struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
	assert(_stream != NULL && _stream->session_type == STREAM_PROTO_SSL);

	TFE_STREAM_LOG_ERROR(_stream, "Failed to create SSL upstream, pass-through the connection : %s. ", what);

	_stream->conn_downstream = __conn_private_create_by_fd(_stream, _stream->defer_fd_downstream);
	_stream->conn_upstream = __conn_private_create_by_fd(_stream, _stream->defer_fd_downstream);

	assert(_stream->conn_downstream != NULL);
	assert(_stream->conn_upstream != NULL);

	_stream->defer_fd_downstream = 0;
	_stream->defer_fd_upstream = 0;
	__conn_private_enable(_stream->conn_downstream);
	__conn_private_enable(_stream->conn_upstream);
}

struct tfe_stream * tfe_stream_create(struct tfe_proxy * pxy, struct tfe_thread_ctx * thread_ctx)
{
	struct tfe_stream_private * _stream = ALLOC(struct tfe_stream_private, 1);
	TFE_PROXY_STAT_INCREASE(STAT_STREAM_CREATE, 1);

	_stream->thread_ref = thread_ctx;
	_stream->proxy_ref = pxy;
	_stream->stream_logger = pxy->logger;

	unsigned int total_plugin_count = tfe_plugin_total_counts();
	_stream->plugin_ctxs = ALLOC(struct plugin_ctx, total_plugin_count);
	return &_stream->head;
}

void __stream_access_log_write(struct tfe_stream_private * stream)
{
	const char * str_passthrough = stream->passthough ? "PASSTHROUGH" : "-";

	MESA_handle_runtime_log(stream->stream_logger, RLOG_LV_INFO, "access",
		"%d %d %d %s %s", stream->log_fd_downstream, stream->log_fd_upstream, stream->keyring_id,
		stream->str_stream_addr, str_passthrough);
}

void tfe_stream_destory(struct tfe_stream_private * stream)
{
	struct tfe_thread_ctx * thread = stream->thread_ref;
	struct event_base * ev_base = thread->evbase;

	TFE_PROXY_STAT_INCREASE(STAT_STREAM_DESTROY, 1);
	__stream_access_log_write(stream);

	if (stream->str_stream_addr)
	{
		free(stream->str_stream_addr);
	}

	if (__is_ssl(stream) && stream->ssl_upstream)
	{
		evutil_socket_t __to_closed_fd = __conn_private_release_fd(stream->conn_upstream);
		ssl_stream_free_and_close_fd(stream->ssl_upstream, ev_base, __to_closed_fd);
	}

	if (__is_ssl(stream) && stream->ssl_downstream)
	{
		evutil_socket_t __to_closed_fd = __conn_private_release_fd(stream->conn_downstream);
		ssl_stream_free_and_close_fd(stream->ssl_downstream, ev_base, __to_closed_fd);
	}

	if (stream->conn_upstream)
	{
		__conn_private_destory(stream->conn_upstream);
	}

	if (stream->conn_downstream)
	{
		__conn_private_destory(stream->conn_downstream);
	}

	if (stream->defer_fd_downstream)
	{
		evutil_closesocket(stream->defer_fd_downstream);
	}

	if (stream->defer_fd_upstream)
	{
		evutil_closesocket(stream->defer_fd_upstream);
	}

	if (stream->future_downstream_create)
	{
		future_destroy(stream->future_downstream_create);
	}

	if (stream->future_upstream_create)
	{
		future_destroy(stream->future_upstream_create);
	}

	stream->proxy_ref = NULL;
	free(stream);
	thread->load--;
}

static struct tfe_stream_addr * __stream_addr_create_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downstream)
{
	struct tfe_stream_addr * __stream_addr = NULL;

	struct sockaddr_storage sk_src_storage{};
	struct sockaddr * sk_src_ptr = (struct sockaddr *) &sk_src_storage;
	socklen_t sk_src_len = sizeof(sk_src_storage);

	struct sockaddr_storage sk_dst_storage{};
	struct sockaddr * sk_dst_ptr = (struct sockaddr *) &sk_dst_storage;
	socklen_t sk_dst_len = sizeof(sk_dst_storage);

	int ret = getsockname(fd_downstream, sk_src_ptr, &sk_src_len);
	if (unlikely(ret < 0))
	{
		TFE_STREAM_LOG_ERROR(stream, "Failed at calling getsockaddr() for fd %d : %s", fd_downstream, strerror(errno));
		goto __errout;
	}

	ret = getpeername(fd_downstream, sk_dst_ptr, &sk_dst_len);
	if (unlikely(ret < 0))
	{
		TFE_STREAM_LOG_ERROR(stream, "Failed at calling getpeername() for fd %d : %s", fd_downstream, strerror(errno));
		goto __errout;
	}

	assert(sk_src_ptr->sa_family == sk_dst_ptr->sa_family);
	if (sk_src_ptr->sa_family == AF_INET)
	{
		__stream_addr = (struct tfe_stream_addr *) malloc(
			sizeof(struct tfe_stream_addr) + sizeof(struct tfe_stream_addr_tuple4_v4));

		struct tfe_stream_addr_ipv4 * st_addr_v4 = __stream_addr->ipv4;
		struct sockaddr_in * sk_v4_src_ptr = (struct sockaddr_in *) sk_src_ptr;
		struct sockaddr_in * sk_v4_dst_ptr = (struct sockaddr_in *) sk_dst_ptr;

		__stream_addr->addrtype = TFE_ADDR_STREAM_TUPLE4_V4;
		__stream_addr->addrlen = sizeof(struct tfe_stream_addr_tuple4_v4);

		st_addr_v4->saddr.s_addr = sk_v4_src_ptr->sin_addr.s_addr;
		st_addr_v4->source = sk_v4_src_ptr->sin_port;
		st_addr_v4->daddr.s_addr = sk_v4_dst_ptr->sin_addr.s_addr;
		st_addr_v4->dest = sk_v4_dst_ptr->sin_port;
	}
	else if (sk_src_ptr->sa_family == AF_INET6)
	{
		assert(0);
	}
	else
	{
		TFE_STREAM_LOG_ERROR(stream, "Invalid sockaddr family for fd %d: sa_family is %d.",
			fd_downstream, sk_src_ptr->sa_family);
		goto __errout;
	}

	return __stream_addr;

__errout:
	if (__stream_addr != NULL) free(__stream_addr);
	return NULL;
}

void __stream_fd_option_setup(struct tfe_stream_private * _stream, evutil_socket_t fd)
{
	struct tfe_stream * stream = &_stream->head;
	struct tfe_proxy_tcp_options * tcp_options = &_stream->proxy_ref->tcp_options;

	/* Make it non-blocking */
	evutil_make_socket_nonblocking(fd);

	/* Recv Buffer */
	if (tcp_options->sz_rcv_buffer >= 0)
	{
		if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (const void *) &tcp_options->sz_rcv_buffer, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_RCVBUF, %d) failed, ignored: %s",
				tcp_options->sz_rcv_buffer, strerror(errno));
		}
	}

	/* Send Buffer */
	if (tcp_options->sz_snd_buffer >= 0)
	{
		if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (const void *) &tcp_options->sz_snd_buffer, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_SNDBUF, %d) failed, ignored: %s",
				tcp_options->sz_snd_buffer, strerror(errno));
		}
	}

	/* Keep-alive */
	if (tcp_options->so_keepalive > 0)
	{
		if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (const void *) &tcp_options->so_keepalive, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_KEEPALIVE, %d) failed, ignored: %s",
				tcp_options->so_keepalive, strerror(errno));
		}
	}

	if (tcp_options->tcp_keepcnt > 0)
	{
		if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, (const void *) &tcp_options->tcp_keepcnt, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPCNT, %d) failed, ignored: %s",
				tcp_options->tcp_keepcnt, strerror(errno));
		}
	}

	if (tcp_options->tcp_keepintvl > 0)
	{
		if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, (const void *) &tcp_options->tcp_keepintvl, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPINTVL, %d) failed, ignored: %s",
				tcp_options->tcp_keepintvl, strerror(errno));
		}
	}

	if (tcp_options->tcp_keepidle > 0)
	{
		if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, (const void *) &tcp_options->tcp_keepidle, sizeof(int)) == -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPIDLE, %d) failed, ignored: %s",
				tcp_options->tcp_keepidle, strerror(errno));
		}
	}

	if (tcp_options->tcp_user_timeout > 0)
	{
		if (setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (const void *) &tcp_options->tcp_user_timeout, sizeof(int))
			== -1)
		{
			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_USER_TIMEOUT, %d) failed, ignored: %s",
				tcp_options->tcp_user_timeout, strerror(errno));
		}
	}

	return;
}

int tfe_stream_init_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downstream, evutil_socket_t fd_upstream)
{
	struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);
	struct event_base * ev_base = _stream->thread_ref->evbase;

	_stream->defer_fd_downstream = fd_downstream;
	_stream->defer_fd_upstream = fd_upstream;
	_stream->log_fd_downstream = fd_downstream;
	_stream->log_fd_upstream = fd_upstream;

	__stream_fd_option_setup(_stream, fd_downstream);
	__stream_fd_option_setup(_stream, fd_upstream);

	_stream->head.addr = tfe_stream_addr_create_by_fd(fd_downstream, CONN_DIR_DOWNSTREAM);
	if (unlikely(_stream->head.addr == NULL))
	{
		TFE_LOG_ERROR(_stream->stream_logger, "Failed to create address from fd %d, %d, terminate fds.",
			fd_downstream, fd_upstream);
		goto __errout;
	}

	_stream->str_stream_addr = tfe_stream_addr_to_str(_stream->head.addr);
	if (_stream->session_type == STREAM_PROTO_PLAIN)
	{
		_stream->conn_downstream = __conn_private_create_by_fd(_stream, fd_downstream);
		_stream->conn_upstream = __conn_private_create_by_fd(_stream, fd_upstream);

		/* Defer FD has been transfer to conn_downstream/conn_upstream */
		_stream->defer_fd_downstream = 0;
		_stream->defer_fd_upstream = 0;

		if (unlikely(_stream->conn_downstream == NULL || _stream->conn_upstream == NULL))
		{
			goto __errout;
		}

		assert(_stream->conn_downstream != NULL);
		assert(_stream->conn_upstream != NULL);

		__conn_private_enable(_stream->conn_downstream);
		__conn_private_enable(_stream->conn_upstream);

		TFE_PROXY_STAT_INCREASE(STAT_STREAM_TCP_PLAIN, 1);
	}

	if (_stream->session_type == STREAM_PROTO_SSL)
	{
		_stream->ssl_mgr = _stream->proxy_ref->ssl_mgr_handler;

		_stream->future_upstream_create = future_create("ssl_up",
			ssl_upstream_create_on_success, ssl_upstream_create_on_fail, (void *) _stream);

		/* Defer setup conn_downstream & conn_upstream in async callbacks. */
		ssl_async_upstream_create(_stream->future_upstream_create,
			_stream->ssl_mgr, fd_upstream, fd_downstream, ev_base);

		TFE_PROXY_STAT_INCREASE(STAT_STREAM_TCP_SSL, 1);
	}

	return 0;

__errout:
	return -1;
}

int tfe_stream_option_set(struct tfe_stream * stream, enum tfe_stream_option opt, const void * arg, size_t sz_arg)
{
	struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);

	if (opt == TFE_STREAM_OPT_SESSION_TYPE)
	{
		assert(sz_arg == sizeof(enum tfe_stream_proto));
		_stream->session_type = *(enum tfe_stream_proto *) arg;
	}
	else if (opt == TFE_STREAM_OPT_PASSTHROUGH)
	{
		assert(sz_arg == sizeof(bool));
		_stream->passthough = *(bool *) arg;
	}
	else if (opt == TFE_STREAM_OPT_KEYRING_ID)
	{
		assert(sz_arg == sizeof(unsigned int));
		_stream->keyring_id = *(unsigned int *) arg;
	}

	return 0;
}

void tfe_stream_write_access_log(const struct tfe_stream * stream, int level, const char * fmt, ...)
{
	va_list arg_ptr;
	va_start(arg_ptr, fmt);
	struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);

	/* Format input content */
	char * __tmp_buffer;
	vasprintf(&__tmp_buffer, fmt, arg_ptr);

	/* Log content with stream tag */
	MESA_handle_runtime_log(_stream->stream_logger, level, "access", "%s %s", _stream->str_stream_addr, __tmp_buffer);
	free(__tmp_buffer);
}