stellar-stellar-2022/src/plugin_manager/plugin_manager.cpp

#include <assert.h>
#include <errno.h>

#include "uthash/uthash.h"

#include "session_manager.h"
#include "plugin_manager_module.h"

/******************************************************************************
 * CallBack Runtime (For Per Session)
 ******************************************************************************/

struct callback_runtime
{
	int skip;
	void *cb_args;

	enum session_event_type event;
	fn_session_event_callback *event_cb;
	fn_session_error_callback *error_cb;
};

struct session_plugin_ctx
{
	int callback_index;
	int callback_num;
	struct callback_runtime *callbacks;
};

/******************************************************************************
 * CallBack Static (For Per Plugin Manager)
 ******************************************************************************/

struct callback_static
{
	enum session_event_type event;
	fn_session_event_callback *event_cb;
	fn_session_error_callback *error_cb;
};

struct plugin_manager_eventcb
{
	char session_name[MAX_SESSION_NAME_LENGTH]; // key

	int callback_num;				   // val size
	struct callback_static *callbacks; // val: dynamic array

	UT_hash_handle hh;
};

/******************************************************************************
 * Struct plugin_manager
 ******************************************************************************/

struct plugin_manager
{
	int used_module_num;
	int used_config_num;
	int used_evencb_num; // only plugin register eventcb numbers

	struct plugin_manager_module *modules[MAX_PLUGIN_NUM];
	struct plugin_manager_config *configs[MAX_PLUGIN_NUM];
	struct plugin_manager_eventcb *evcb_htable;
};

/******************************************************************************
 * Create/Destory plugin ctx (per session)
 ******************************************************************************/

static struct session_plugin_ctx *plugin_manager_create_plugin_ctx(struct plugin_manager *plug_mgr, const char *session_name)
{
	if (session_name == NULL || strlen(session_name) == 0)
	{
		plugin_manager_log(ERROR, "invalid parameter, session name is empty");
		return NULL;
	}

	struct plugin_manager_eventcb *elem;
	HASH_FIND_STR(plug_mgr->evcb_htable, session_name, elem);
	if (elem == NULL)
	{
		plugin_manager_log(ERROR, "can't find event callback for session name '%s'", session_name);
		return NULL;
	}
	else
	{
		struct session_plugin_ctx *plug_ctx = safe_alloc(struct session_plugin_ctx, 1);
		plug_ctx->callback_num = elem->callback_num;
		plug_ctx->callbacks = safe_alloc(struct callback_runtime, plug_ctx->callback_num);

		for (int i = 0; i < plug_ctx->callback_num; i++)
		{
			plug_ctx->callbacks[i].skip = 0;
			plug_ctx->callbacks[i].event = elem->callbacks[i].event;
			plug_ctx->callbacks[i].event_cb = elem->callbacks[i].event_cb;
			plug_ctx->callbacks[i].error_cb = elem->callbacks[i].error_cb;
			plug_ctx->callbacks[i].cb_args = NULL;
		}

		return plug_ctx;
	}
}

static void plugin_manager_destory_plugin_ctx(struct session_plugin_ctx *plug_ctx)
{
	if (plug_ctx)
	{
		safe_free(plug_ctx->callbacks);
		safe_free(plug_ctx);
	}
}

/******************************************************************************
 * Tools for managing plugins
 ******************************************************************************/

static int plugin_manager_parse_plugins(struct plugin_manager *plug_mgr, const char *file)
{
	char line_buffer[4096] = {0};

	if (strlen(file) <= 0)
	{
		plugin_manager_log(ERROR, "Invalid parameter, plugin config file name cannot be empty");
		return -1;
	}

	FILE *fp = fopen(file, "r");
	if (fp == NULL)
	{
		plugin_manager_log(ERROR, "can't open %s, %s", file, strerror(errno));
		return -1;
	}

	while (fgets(line_buffer, sizeof(line_buffer), fp))
	{
		if ('#' == line_buffer[0] || '\n' == line_buffer[0])
		{
			memset(line_buffer, 0, sizeof(line_buffer));
			continue;
		}
		line_buffer[strcspn(line_buffer, "\r\n")] = 0;

		if (plug_mgr->used_config_num >= MAX_PLUGIN_NUM)
		{
			plugin_manager_log(ERROR, "the number of registered plugins exceeds the limit and cannot exceed %d", MAX_PLUGIN_NUM);
			goto err;
		}

		struct plugin_manager_config *config = plugin_mangager_config_create();
		if (plugin_mangager_config_parse(config, line_buffer) == -1)
		{
			plugin_mangager_config_destory(config);
			goto err;
		}

		plug_mgr->configs[plug_mgr->used_config_num] = config;
		plug_mgr->used_config_num++;
		memset(line_buffer, 0, sizeof(line_buffer));
	}

	fclose(fp);

	return 0;

err:

	if (fp)
	{
		fclose(fp);
		fp = NULL;
	}

	return -1;
}

static int plugin_manager_open_plugins(struct plugin_manager *plug_mgr)
{
	for (int i = 0; i < plug_mgr->used_config_num; i++)
	{
		struct plugin_manager_config *config = plug_mgr->configs[i];
		struct plugin_manager_module *module = plugin_manager_module_open(config);
		if (module == NULL)
		{
			return -1;
		}
		plugin_manager_module_dump(module, config);

		plug_mgr->modules[plug_mgr->used_module_num] = module;
		plug_mgr->used_module_num++;
	}

	return 0;
}

static int plugin_manager_register_plugins(struct plugin_manager *plug_mgr)
{
	for (int i = 0; i < plug_mgr->used_module_num; i++)
	{
		struct plugin_manager_module *module = plug_mgr->modules[i];
		if (plugin_manager_module_register(plug_mgr, module) == -1)
		{
			return -1;
		}
		plug_mgr->used_evencb_num++;
	}

	return 0;
}

static int plugin_manager_init_plugins(struct plugin_manager *plug_mgr)
{
	for (int i = 0; i < plug_mgr->used_module_num; i++)
	{
		struct plugin_manager_module *module = plug_mgr->modules[i];
		if (plugin_manager_module_init(module) == -1)
		{
			return -1;
		}
		double percentage = ((double)(i + 1)) / ((double)plug_mgr->used_module_num) * ((double)100);
		plugin_manager_log(INFO, "Plugin initialization progress: [%.2f%]", percentage);
	}

	return 0;
}

static void plugin_manager_exit_plugins(struct plugin_manager *plug_mgr)
{
	if (plug_mgr && plug_mgr->used_module_num)
	{
		for (int i = 0; i < plug_mgr->used_module_num; i++)
		{
			struct plugin_manager_module *module = plug_mgr->modules[i];
			plugin_manager_module_exit(module);
		}
	}
}

static void plugin_manager_close_plugins(struct plugin_manager *plug_mgr)
{
	if (plug_mgr && plug_mgr->used_module_num)
	{
		for (int i = 0; i < plug_mgr->used_module_num; i++)
		{
			struct plugin_manager_module *module = plug_mgr->modules[i];
			plugin_manager_module_close(module);
		}
		plug_mgr->used_module_num = 0;
	}
}

// deparse for destory parse
static void plugin_manager_deparse_plugins(struct plugin_manager *plug_mgr)
{
	if (plug_mgr && plug_mgr->used_config_num)
	{
		for (int i = 0; i < plug_mgr->used_config_num; i++)
		{
			struct plugin_manager_config *config = plug_mgr->configs[i];
			plugin_mangager_config_destory(config);
		}
		plug_mgr->used_config_num = 0;
	}
}

/******************************************************************************
 * Public API for managing plugins
 ******************************************************************************/

int plugin_manager_load(struct plugin_manager *plug_mgr, const char *file)
{
	if (plugin_manager_parse_plugins(plug_mgr, file) == -1)
	{
		return -1;
	}

	if (plugin_manager_open_plugins(plug_mgr) == -1)
	{
		return -1;
	}

	if (plugin_manager_register_plugins(plug_mgr) == -1)
	{
		return -1;
	}

	if (plugin_manager_init_plugins(plug_mgr) == -1)
	{
		return -1;
	}

	return 0;
}

void plugin_manager_unload(struct plugin_manager *plug_mgr)
{
	plugin_manager_exit_plugins(plug_mgr);
	plugin_manager_close_plugins(plug_mgr);
	plugin_manager_deparse_plugins(plug_mgr);
}

struct plugin_manager *plugin_manager_create()
{
	struct plugin_manager *plug_mgr = safe_alloc(struct plugin_manager, 1);

	plug_mgr->used_module_num = 0;
	plug_mgr->used_config_num = 0;
	plug_mgr->used_evencb_num = 0;

	plug_mgr->evcb_htable = NULL;

	return plug_mgr;
}

void plugin_manager_destory(struct plugin_manager *plug_mgr)
{
	if (plug_mgr)
	{
		if (plug_mgr->evcb_htable)
		{
			struct plugin_manager_eventcb *elem;
			struct plugin_manager_eventcb *tmp;
			HASH_ITER(hh, plug_mgr->evcb_htable, elem, tmp)
			{
				HASH_DEL(plug_mgr->evcb_htable, elem);

				safe_free(elem->callbacks);
				safe_free(elem);
			}

			plug_mgr->evcb_htable = NULL;
			plug_mgr->used_evencb_num = 0;
		}

		plugin_manager_unload(plug_mgr);
		safe_free(plug_mgr);
	}
}

int plugin_manager_register(struct plugin_manager *plug_mgr, const char *session_name, enum session_event_type event, fn_session_event_callback *event_cb, fn_session_error_callback *error_cb)
{
	if (strlen(session_name) <= 0)
	{
		plugin_manager_log(ERROR, "invalid parameter, session name is empty");
		return -1;
	}

	if (strlen(session_name) > MAX_SESSION_NAME_LENGTH)
	{
		plugin_manager_log(ERROR, "invalid parameter, session name '%s' is too long and exceeds '%d' bytes", session_name, MAX_SESSION_NAME_LENGTH);
		return -1;
	}

	if (event_cb == NULL)
	{
		plugin_manager_log(ERROR, "invalid parameter, the event callback corresponding to the session name '%s' is null", session_name);
		return -1;
	}

	if (error_cb == NULL)
	{
		plugin_manager_log(ERROR, "invalid parameter, the error callback corresponding to the session name '%s' is null", session_name);
		return -1;
	}

	struct plugin_manager_eventcb *elem;
	HASH_FIND_STR(plug_mgr->evcb_htable, session_name, elem);
	// session_name exists, add a new cb to the end of the callbacks dynamic array
	if (elem)
	{
		elem->callbacks = (struct callback_static *)realloc(elem->callbacks, (elem->callback_num + 1) * sizeof(struct callback_static));

		elem->callbacks[elem->callback_num].event = event;
		elem->callbacks[elem->callback_num].event_cb = event_cb;
		elem->callbacks[elem->callback_num].error_cb = error_cb;

		elem->callback_num++;
	}
	// session_name does not exist, allocate a new node elem, and add elem to the hash table
	else
	{
		elem = safe_alloc(struct plugin_manager_eventcb, 1);
		memcpy(elem->session_name, session_name, strlen(session_name));

		elem->callbacks = (struct callback_static *)realloc(elem->callbacks, (elem->callback_num + 1) * sizeof(struct callback_static));

		elem->callbacks[elem->callback_num].event = event;
		elem->callbacks[elem->callback_num].event_cb = event_cb;
		elem->callbacks[elem->callback_num].error_cb = error_cb;

		elem->callback_num++;

		HASH_ADD_STR(plug_mgr->evcb_htable, session_name, elem);
	}

	return 0;
}

void plugin_manager_dispatch(struct plugin_manager *plug_mgr, struct stellar_event *event)
{
	const struct stellar_session *seesion = stellar_event_get_session(event);
	struct session_plugin_ctx *plug_ctx = stellar_event_get_plugin_ctx(event);
	enum session_event_type event_type = stellar_event_get_type(event);
	const char *session_name = stellar_event_get_session_name(event);
	struct stellar_packet *packet = stellar_event_get_packet(event);
	uint16_t payload_len = stellar_event_get_payload_length(event);
	const char *payload = stellar_event_get_payload(event);

	assert(seesion);
	assert(session_name);

	char event_str_buffer[1024] = {0};
	session_event_type_int2str(event_type, event_str_buffer, 1024);

	// the same session may trigger multi times opening events
	if (event_type & SESSION_EVENT_OPENING)
	{
		if (plug_ctx == NULL)
		{
			plug_ctx = plugin_manager_create_plugin_ctx(plug_mgr, session_name);
			if (plug_ctx == NULL)
			{
				plugin_manager_log(ERROR, "can't create runtime plugin ctx for session '%s', Please check whether the callback is registered in the current session");
				return;
			}
			stellar_event_set_plugin_ctx(event, plug_ctx);
		}
	}

	if (plug_ctx)
	{
		for (int i = 0; i < plug_ctx->callback_num; i++)
		{
			struct callback_runtime *runtime = &plug_ctx->callbacks[i];
			if (runtime->skip == 1)
			{
				plugin_manager_log(DEBUG, "dispatch, skip event_cb: %p, session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
				continue;
			}

			if (runtime->event & event_type)
			{
				plug_ctx->callback_index = i;
				plugin_manager_log(DEBUG, "dispatch, run event_cb: %p, session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
				runtime->event_cb(seesion, event_type, packet, payload, payload_len, &runtime->cb_args);
			}
		}
	}
	else
	{
		plugin_manager_log(ERROR, "session '%s' runtime plugin ctx is null when running event callback", session_name);
		abort();
	}

	if (event_type & SESSION_EVENT_CLOSING)
	{
		plugin_manager_destory_plugin_ctx(plug_ctx);
		stellar_event_set_plugin_ctx(event, NULL);
	}
}

/******************************************************************************
 * Public API For Plugin
 ******************************************************************************/

void pm_session_dettach_me(const struct stellar_session *session)
{
	struct session_plugin_ctx *plugin_ctx = stellar_session_get_plugin_ctx(session);
	assert(plugin_ctx);
	struct callback_runtime *runtime_me = &plugin_ctx->callbacks[plugin_ctx->callback_index];

	/*
	 * Just set the skip flag and don't call this event callback next.
	 * The plugin is closed before calling pm_session_dettach_me.
	 */
	runtime_me->skip = 1;
	plugin_manager_log(DEBUG, "%p dettach me, disable event_cb: %p, session: %s", runtime_me->event_cb, runtime_me->event_cb, stellar_session_get_name(session));
}

void pm_session_dettach_others(const struct stellar_session *session)
{
	struct session_plugin_ctx *plugin_ctx = stellar_session_get_plugin_ctx(session);
	assert(plugin_ctx);
	struct callback_runtime *runtime_me = &plugin_ctx->callbacks[plugin_ctx->callback_index];

	for (int i = 0; i < plugin_ctx->callback_num; i++)
	{
		if (i != plugin_ctx->callback_index)
		{
			struct callback_runtime *runtime_other = &plugin_ctx->callbacks[i];
			runtime_other->skip = 1;
			plugin_manager_log(DEBUG, "%p dettach others, run error_cb: %p, session: %s", runtime_me->event_cb, runtime_other->error_cb, stellar_session_get_name(session));
			runtime_other->error_cb(session, ERROR_EVENT_DETTACH, &runtime_other->cb_args);
		}
	}
}

/******************************************************************************
 * Util For Gtest
 ******************************************************************************/

void *pm_session_get_plugin_pme(const struct stellar_session *session)
{
	struct session_plugin_ctx *plugin_ctx = stellar_session_get_plugin_ctx(session);
	assert(plugin_ctx);
	struct callback_runtime *runtime_me = &plugin_ctx->callbacks[plugin_ctx->callback_index];

	return runtime_me->cb_args;
}

/******************************************************************************
 * Suppport LUA plugins
 ******************************************************************************/

// TODO