tango-tfe/plugin/protocol/http/src/http_entry.cpp

#include <MESA/MESA_list_queue.h>
#include <tfe_stream.h>
#include <tfe_utils.h>
#include <tfe_plugin.h>
#include <tfe_stream.h>
#include <http_parser.h>
#include <malloc.h>

#include <http_common.h>
#include <http_half.h>
#include <assert.h>
#include <event.h>

struct http_plugin __g_http_plugin;
struct http_plugin * g_http_plugin = &__g_http_plugin;

int http_plugin_init(struct tfe_proxy * proxy)
{
	return 0;
}

void http_plugin_deinit(struct tfe_proxy * proxy)
{
	return;
}

int http_connection_entry_open(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_conn_dir dir, void ** pme)
{
	struct http_connection_private * ht_conn = ALLOC(
	struct http_connection_private, 1);
	TAILQ_INIT(&ht_conn->hs_private_list);
	TAILQ_INIT(&ht_conn->hs_private_orphan_list);
	ht_conn->stream = stream;
	*pme = (void *) ht_conn;
	return 0;
}

struct user_event_dispatch_closure
{
	const struct tfe_stream * stream;
	const struct tfe_http_session * session;
	unsigned int thread_id;
};

static int __user_event_dispatch(struct http_half_private * hf_private,
	enum tfe_http_event ev, const unsigned char * data, size_t len, void * user)
{
	struct user_event_dispatch_closure * __closure = (struct user_event_dispatch_closure *) user;
	struct http_frame_session_ctx * ht_frame = hf_private->session->ht_frame;
	//todo:
	http_frame_raise_event(ht_frame, __closure->stream, (struct tfe_http_session *) __closure->session,
		ev, data, len, __closure->thread_id);
	return 0;
}

static int __write_http_half(struct http_half_private * hf_private,
	const struct tfe_stream * stream, enum tfe_conn_dir dir)
{
	/* Construct, and write response immediately */
	hf_private_construct(hf_private);
	size_t __to_write_len = evbuffer_get_length(hf_private->evbuf_raw);
	unsigned char * __to_write = evbuffer_pullup(hf_private->evbuf_raw, __to_write_len);

	/* Write the data to stream, UPSTREAM is the incoming direction for response */
	int ret = 0;
	if (hf_private->write_ctx)
	{
		ret = tfe_stream_write_frag(hf_private->write_ctx, __to_write, __to_write_len);
	}
	else
	{
		ret = tfe_stream_write(stream, dir, __to_write, __to_write_len);
	}

	if (unlikely(ret < 0)) { assert(0); }
	evbuffer_drain(hf_private->evbuf_raw, __to_write_len);
	return ret;
}

static int __write_http_half_to_line(const struct tfe_stream * stream,
	enum tfe_conn_dir dir, struct http_half_private * hf_private)
{
	int ret = 0;

	if (hf_private->is_setup_by_stream)
	{
		/* By stream, first time to write http request/response,
		 * and has been call body_begin, construct the header */
		if (hf_private->evbuf_body != NULL && hf_private->write_ctx == NULL)
		{
//			printf("frag write start, stream = %p, hf_private = %p\n", stream, hf_private);

			/* Still need to send data(not call body_end()), need to write frag start
			 * Otherwise, means the body is ready, send out directly without frag */
			if (hf_private->message_status < STATUS_COMPLETE)
			{
				hf_private->write_ctx = tfe_stream_write_frag_start(stream, dir);
				assert(hf_private->write_ctx != NULL);
			}

			ret = __write_http_half(hf_private, stream, dir);
			if (unlikely(ret < 0)) return ret;
		}
	}
	else
	{
		/* Not stream, need to be complete, then construct all data and send out */
		if (hf_private->message_status == STATUS_COMPLETE)
		{
			ret = __write_http_half(hf_private, stream, dir);
			if (unlikely(ret < 0)) return ret;
		}
	}

	return 0;
}

int __on_request_handle_user_req_or_resp(const tfe_stream * stream, struct http_session_private * hs_private,
	struct http_half_private * hf_private_req_in, bool & need_to_close_the_session)
{
	struct http_connection_private * hc_private = hs_private->hc_private;
	int ret = 0;

	/* Cannot setup user request and user response simultaneously */
	assert(!(hs_private->hf_private_req_user != NULL && hs_private->hf_private_resp_user != NULL));

	/* User's construct request */
	if (hs_private->hf_private_req_user != NULL)
	{
		struct http_half_private * hf_private_req_user = hs_private->hf_private_req_user;
		ret = __write_http_half_to_line(stream, CONN_DIR_UPSTREAM, hf_private_req_user);
		if (unlikely(ret < 0))
		{
			TFE_STREAM_LOG_ERROR(stream, "Failed to write HTTP request setup by user".);
			return ret;
		}

		if (hf_private_req_user->message_status == STATUS_COMPLETE)
		{
			hf_private_destory(hf_private_req_user);
			hs_private->hf_private_resp_user = NULL;
		}

		assert(hf_private_req_in->stream_action == ACTION_DEFER_DATA
		|| hf_private_req_in->stream_action == ACTION_DROP_DATA);

		hf_private_req_in->stream_action = ACTION_DROP_DATA;
	}

		/* User's construct response, send before real response arrived. */
	else if (hs_private->hf_private_resp_user != NULL)
	{
		struct http_half_private * hf_private_resp_user = hs_private->hf_private_resp_user;
		ret = __write_http_half_to_line(stream, CONN_DIR_DOWNSTREAM, hf_private_resp_user);
		if (unlikely(ret < 0))
		{
			TFE_STREAM_LOG_ERROR(stream, "Failed to write HTTP request setup by user".);
			goto __errout;
		}

		if (hf_private_resp_user->message_status == STATUS_COMPLETE)
		{
			hf_private_destory(hf_private_resp_user);
			hs_private->hf_private_resp_user = NULL;
			need_to_close_the_session = true;
		}
		else
		{
			TAILQ_REMOVE(&hc_private->hs_private_list, hs_private, next);
			TAILQ_INSERT_TAIL(&hc_private->hs_private_orphan_list, hs_private, next);
		}

		assert(hf_private_req_in->stream_action == ACTION_DEFER_DATA
		||  hf_private_req_in->stream_action == ACTION_DROP_DATA);
		hf_private_req_in->stream_action = ACTION_DROP_DATA;
	}

	return 0;

__errout:
	return -1;
}

enum tfe_stream_action __http_connection_entry_on_request(const struct tfe_stream * stream,
	struct http_connection_private * hc_private, unsigned int thread_id, const unsigned char * data, size_t len)
{
	struct http_session_private * hs_private = TAILQ_LAST(&hc_private->hs_private_list, hs_private_list);
	struct http_half_private * hf_private_req_in = to_hf_request_private(hs_private);

	bool __need_to_close_the_session = false;
	int ret = 0;

	enum tfe_stream_action __action = ACTION_FORWARD_DATA;
	size_t __action_args = 0;

	/* There is no available in session list,
	 * that indicate all HTTP request has corresponding response,
	 * or the last request is finished, we need to create a new session. */
	if (hs_private == NULL || hf_private_req_in->message_status == STATUS_COMPLETE)
	{
		/* HTTP Request and Session */
		hf_private_req_in = hf_private_create(TFE_HTTP_REQUEST, 1, 0);
		hs_private = hs_private_create(hc_private, hf_private_req_in, NULL);
		hf_private_set_session(hf_private_req_in, hs_private);

		/* Closure, catch stream, session and thread_id */
		struct user_event_dispatch_closure * __closure = ALLOC(
		struct user_event_dispatch_closure, 1);
		__closure->thread_id = thread_id;
		__closure->stream = stream;
		__closure->session = to_hs_public(hs_private);

		/* Set callback, this callback used to raise business event */
		hf_private_set_callback(hf_private_req_in, __user_event_dispatch, __closure, free);

		/* Call business plugin */
		hs_private->ht_frame = http_frame_raise_session_begin(stream, &hs_private->hs_public, thread_id);
		if (hs_private->ht_frame == NULL)
		{
			TFE_STREAM_LOG_ERROR(stream, "Failed at raising session begin event. ");
			goto __errout;
		}

		TAILQ_INSERT_TAIL(&hc_private->hs_private_list, hs_private, next);
	}

	/* The session is suspended, and to resume */
	if (hs_private->suspend_tag_effective)
	{
		hf_private_req_in->event_cb(hf_private_req_in, hs_private->suspend_event, NULL, 0,
			hf_private_req_in->event_cb_user);

		hs_private->suspend_event = (enum tfe_http_event) 0;
		hs_private->suspend_tag_effective = false;
		hs_private->suspend_tag_user = false;

		if (__on_request_handle_user_req_or_resp(stream, hs_private,
			hf_private_req_in, __need_to_close_the_session) < 0)
		{
			goto __errout;
		}

		/* Ignore parse the content which is nullptr. */
		goto __out;
	}

	/* Parse the content, the data which in defered state has been ignored. */
	ret = hf_private_parse(hf_private_req_in, data, len);

	/* Need more data, no boundary touched */
	if (ret == 0)
	{
		if (hf_private_req_in->stream_action == ACTION_DROP_DATA ||
			hf_private_req_in->stream_action == ACTION_FORWARD_DATA)
		{
			hf_private_req_in->parse_cursor = 0;
		}

		return hf_private_req_in->stream_action;
	}

	/* Suspend */
	if (hs_private->suspend_tag_user)
	{
		assert(!hs_private->suspend_tag_effective);
		hs_private->suspend_tag_effective = true;

		/* Suspend TCP stream */
		tfe_stream_suspend(stream, CONN_DIR_DOWNSTREAM);
		return ACTION_DEFER_DATA;
	}

	/* Some kind of error happened, write log and detach the stream */
	if (ret == -1)
	{
		TFE_STREAM_LOG_ERROR(stream, "Failed at parsing stream as HTTP: %u, %s, %s",
			hf_private_req_in->parse_errno, http_errno_name(hf_private_req_in->parse_errno),
			http_errno_description(hf_private_req_in->parse_errno));

		goto __errout;
	}

	/* Touch a boundary, such as the end of HTTP headers, bodys, et al. */
	__action_args = hf_private_req_in->parse_cursor;
	hf_private_req_in->parse_cursor = 0;

	if (__on_request_handle_user_req_or_resp(stream, hs_private,
		hf_private_req_in, __need_to_close_the_session) < 0)
	{
		goto __errout;
	}

	if (hf_private_req_in->stream_action == ACTION_FORWARD_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_FOWARD_BYTES, &__action_args, sizeof(__action_args));
		__action = ACTION_FORWARD_DATA;
	}

	if (hf_private_req_in->stream_action == ACTION_DROP_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_DROP_BYTES, &__action_args, sizeof(__action_args));
		__action = ACTION_DROP_DATA;
	}

__out:
	/* There is nothing for this session, close the session */
	if (__need_to_close_the_session)
	{
		http_frame_raise_session_end(hs_private->ht_frame, stream, &hs_private->hs_public, thread_id);
		TAILQ_REMOVE(&hc_private->hs_private_list, hs_private, next);
		hs_private_destory(hs_private);
	}

	return __action;

__errout:
	tfe_stream_detach(stream);
	return ACTION_FORWARD_DATA;
}

enum tfe_stream_action __http_connection_entry_on_response(const struct tfe_stream * stream,
	struct http_connection_private * hc_private, unsigned int thread_id, const unsigned char * data, size_t len)
{
	struct http_session_private * hs_private = TAILQ_FIRST(&hc_private->hs_private_list);
	struct http_half_private * hf_private_resp_in;
	struct http_half_private * hf_private_resp_user;
	enum tfe_stream_action __stream_action;

	int ret = 0;
	size_t __action_byptes;

	/* Standalone response, it means missing something or malformed http protocol */
	if (hs_private == NULL)
	{
		TFE_STREAM_LOG_ERROR(stream, "Standlone HTTP response emerged. Malformed HTTP Protocol, detached. ");
		goto __errout;
	}

	hf_private_resp_in = to_hf_response_private(hs_private);
	/* First time parse http response */
	if (hf_private_resp_in == NULL)
	{
		/* HTTP Version */
		short resp_major = to_hf_request_private(hs_private)->major;
		short resp_minor = to_hf_request_private(hs_private)->minor;

		/* Response */
		hf_private_resp_in = hf_private_create(TFE_HTTP_RESPONSE, resp_major, resp_minor);
		hs_private_hf_private_set(hs_private, hf_private_resp_in, TFE_HTTP_RESPONSE);
		hf_private_set_session(hf_private_resp_in, hs_private);

		/* Closure, catch stream, session and thread_id */
		struct user_event_dispatch_closure * __closure = ALLOC(
		struct user_event_dispatch_closure, 1);
		__closure->thread_id = thread_id;
		__closure->stream = stream;
		__closure->session = to_hs_public(hs_private);

		/* Set callback, this callback used to raise business event */
		hf_private_set_callback(hf_private_resp_in, __user_event_dispatch, __closure, free);

		/* Setup user's action */
		if (hs_private->hf_private_resp_user != NULL)
		{
			hf_private_resp_in->is_user_stream_action_set = true;
			hf_private_resp_in->user_stream_action = ACTION_DROP_DATA;
		}
	}

	/* Parse the content, the data which in defered state has been ignored. */
	ret = hf_private_parse(hf_private_resp_in, data, len);

	/* Need more data, no boundary touched */
	if (ret == 0)
	{
		if (hf_private_resp_in->stream_action == ACTION_DROP_DATA ||
			hf_private_resp_in->stream_action == ACTION_FORWARD_DATA)
		{
			hf_private_resp_in->parse_cursor = 0;
		}

		return hf_private_resp_in->stream_action;
	}

	/* Some kind of error happened, write log and detach the stream */
	if (ret == -1)
	{
		TFE_STREAM_LOG_ERROR(stream, "Failed at parsing HTTP response: %u, %s, %s",
			hf_private_request->parse_errno, http_errno_name(hf_private_request->parse_errno),
			http_errno_description(hf_private_request->parse_errno));

		goto __errout;
	}

	hf_private_resp_user = hs_private->hf_private_resp_user;
	if (hf_private_resp_user != NULL)
	{
		ret = __write_http_half_to_line(stream, CONN_DIR_DOWNSTREAM, hf_private_resp_user);
		if (unlikely(ret < 0))
		{
			TFE_STREAM_LOG_ERROR(stream, "Failed to write HTTP response setup by user".);
			goto __errout;
		}

		if (hf_private_resp_user->message_status == STATUS_COMPLETE)
		{
			hf_private_destory(hf_private_resp_user);
			hs_private->hf_private_resp_user = NULL;
		}

		if (hf_private_resp_in->stream_action == ACTION_DEFER_DATA)
		{
			hf_private_resp_in->stream_action = ACTION_DROP_DATA;
		}
	}

	__stream_action = hf_private_resp_in->stream_action;
	__action_byptes = hf_private_resp_in->parse_cursor;
	hf_private_resp_in->parse_cursor = 0;

	if (hf_private_resp_in->message_status == STATUS_COMPLETE)
	{
		/* Still sending user's response, should not destroy the session,
		 * move the session to orphan list, then we can handle the next session's response */
		if (hs_private->hf_private_resp_user != NULL)
		{
			TAILQ_REMOVE(&hc_private->hs_private_list, hs_private, next);
			TAILQ_INSERT_TAIL(&hc_private->hs_private_orphan_list, hs_private, next);
		}

			/* Nothing to do, everything is over, destroy the session */
		else
		{
			http_frame_raise_session_end(hs_private->ht_frame, stream, &hs_private->hs_public, thread_id);
			TAILQ_REMOVE(&hc_private->hs_private_list, hs_private, next);
			hs_private_destory(hs_private);
		}
	}

	if (__stream_action == ACTION_FORWARD_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_FOWARD_BYTES, &__action_byptes, sizeof(__action_byptes));
		return ACTION_FORWARD_DATA;
	}

	if (__stream_action == ACTION_DROP_DATA)
	{
		tfe_stream_action_set_opt(stream, ACTION_OPT_DROP_BYTES, &__action_byptes, sizeof(__action_byptes));
		return ACTION_DROP_DATA;
	}

	goto __errout;

__errout:
	tfe_stream_detach(stream);
	return ACTION_FORWARD_DATA;
}

int __http_connection_identify(const struct tfe_stream * stream,
	struct http_connection_private * ht_conn, const unsigned char * data, size_t len)
{
	struct http_half_private * hf_private = hf_private_create(TFE_HTTP_REQUEST, 1, 0);
	int ret = hf_private_parse(hf_private, data, len);
	hf_private_destory(hf_private);
	return ret;
}

#define HTTP_INDENTIFY_LENGTH            4

enum tfe_stream_action http_connection_entry_data(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_conn_dir dir, const unsigned char * data, size_t len, void ** pme)
{
	struct http_connection_private * ht_conn = (struct http_connection_private *) (*pme);

	if (ht_conn->is_preempted == 0)
	{
		/* If the server push response before client send request, this must not be HTTP, detach the stream */
		if (dir == CONN_DIR_UPSTREAM)
			goto __detach;

		/* Protocol Identification, we need 8 bytes at least to tell it is HTTP or not */
		if (len < HTTP_INDENTIFY_LENGTH) goto __detach;

		/* Now, we want to identify this stream */
		int ret = __http_connection_identify(stream, ht_conn, data, len);
		if (ret < 0) goto __detach;

		/* This is HTTP, try to preempt the stream
		 * It may be failed because other plugin preempted before us */
		ret = tfe_stream_preempt(stream);
		if (ret != 0) goto __detach;
		ht_conn->is_preempted = 1;
	}

	/* This stream has been preempt, this plugin try to parse it */
	return (dir == CONN_DIR_DOWNSTREAM) ? __http_connection_entry_on_request(stream, ht_conn, thread_id,
		data, len) : __http_connection_entry_on_response(stream, ht_conn, thread_id, data, len);

__detach:
	tfe_stream_detach(stream);
	return ACTION_FORWARD_DATA;
}

void http_connection_entry_close(const struct tfe_stream * stream, unsigned int thread_id,
	enum tfe_stream_close_reason reason, void ** pme)
{
	struct http_connection_private * ht_conn = (struct http_connection_private *) (*pme);
	*pme = NULL;

	/* Delete all live sessions */
	while (true)
	{
		struct http_session_private * hs_private_iter = TAILQ_FIRST(&ht_conn->hs_private_list);
		if (hs_private_iter == NULL) break;

		TAILQ_REMOVE(&ht_conn->hs_private_list, hs_private_iter, next);
		hs_private_destory(hs_private_iter);
	}

	/* Clear session counter, and free ht_conn structure */
	ht_conn->session_id_counter = 0;
	free(ht_conn);
}

static struct tfe_plugin __http_plugin_info =
	{
		.symbol = "HTTP",
		.type = TFE_PLUGIN_TYPE_PROTOCOL,
		.on_init = http_plugin_init,
		.on_deinit = http_plugin_deinit,
		.on_open = http_connection_entry_open,
		.on_data = http_connection_entry_data,
		.on_close = http_connection_entry_close
	};

TFE_PLUGIN_REGISTER(HTTP, __http_plugin_info
)