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>

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);
	*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;

	http_frame_raise_event(ht_frame, __closure->stream, __closure->session,
		ev, data, len, __closure->thread_id); return 0;
}

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_request = to_hf_request_private(hs_private);

	/* tfe_hexdump(stderr, __FUNCTION__, data, (unsigned int)len); */
	int ret = 0;
	size_t __action_byptes;

	/* 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_request->message_status == STATUS_COMPLETE)
	{
		/* HTTP Request and Session */
		hf_private_request = hf_private_create(TFE_HTTP_REQUEST, 1, 0);
		hs_private = hs_private_create(hc_private, hf_private_request, NULL);

		/* 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_request, __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);
	}

	/* proceed parse content for last request */
	ret = hf_private_parse(hf_private_request, data, len);

	/* Need more data, no boundary touched */
	if (ret == 0)
	{
		return hf_private_request->stream_action;
	}

	/* 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_request->parse_errno, http_errno_name(hf_private_request->parse_errno),
			http_errno_description(hf_private_request->parse_errno));

		goto __errout;
	}

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

	if (hf_private_request->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 (hf_private_request->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;
}

enum tfe_stream_action __http_connection_entry_on_response(const struct tfe_stream * stream,
	struct http_connection_private * ht_conn, unsigned int thread_id, const unsigned char * data, size_t len)
{
	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_RESPONSE, 1, 0);
	int ret = hf_private_parse(hf_private, data, len);
	hf_private_destory(hf_private);
	return ret;
}

#define HTTP_INDENTIFY_LENGTH            8

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)
		{
			static const unsigned int __defer_bytes = HTTP_INDENTIFY_LENGTH;
			tfe_stream_action_set_opt(stream, ACTION_OPT_DEFER_BYTES, (void *) &__defer_bytes, sizeof(__defer_bytes));
			return ACTION_DEFER_DATA;
		}

		/* 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);
	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)