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

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

#include "uthash/uthash.h"

#include "sdk/include/plugin.h"
#include "session_manager.h"
#include "plugin_manager_module.h"

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

enum plugin_status
{
	PLUGIN_STATUS_NORMAL = 0x00,
	PLUGIN_STATUS_DETTACH_ME = 0x01,
	PLUGIN_STATUS_TAKEN_OVER = 0x02, // Noitce: this is taken over not take over
};

struct callback_runtime
{
	void *cb_args;
	plugin_event_callback *event_cb;

	enum session_event_type event;
	enum plugin_status status;
	int is_be_called;
};

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;
	plugin_event_callback *event_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].is_be_called = 0;
			plug_ctx->callbacks[i].status = PLUGIN_STATUS_NORMAL;
			plug_ctx->callbacks[i].event = elem->callbacks[i].event;
			plug_ctx->callbacks[i].event_cb = elem->callbacks[i].event_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, plugin_event_callback *event_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;
	}

	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->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->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);
	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", session_name);
				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->status == PLUGIN_STATUS_DETTACH_ME)
			{
				plugin_manager_log(DEBUG, "dispatch, skip event_cb: %p, plugin status: 'dettach me', session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
				continue;
			}
			else if (runtime->status == PLUGIN_STATUS_TAKEN_OVER)
			{
				if ((event_type & SESSION_EVENT_CLOSING) && (runtime->event & SESSION_EVENT_CLOSING) && runtime->is_be_called)
				{
					plug_ctx->callback_index = i;
					plugin_manager_log(DEBUG, "dispatch, run event_cb: %p, plugin status: 'taken over', session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
					runtime->event_cb(seesion, SESSION_EVENT_CLOSING, payload, payload_len, &runtime->cb_args);
					continue;
				}
				else
				{
					plugin_manager_log(DEBUG, "dispatch, skip event_cb: %p, plugin status: 'taken over', session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
				}
			}
			else if (runtime->status == PLUGIN_STATUS_NORMAL)
			{
				if (runtime->event & event_type)
				{
					plug_ctx->callback_index = i;
					plugin_manager_log(DEBUG, "dispatch, run event_cb: %p, plugin status: 'normal', session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
					runtime->event_cb(seesion, event_type, payload, payload_len, &runtime->cb_args);
					runtime->is_be_called = 1;
				}
				else
				{
					plugin_manager_log(DEBUG, "dispatch, skip event_cb: %p, plugin status: 'normal', session: %s, event: (%d, %s)", runtime->event_cb, session_name, event_type, event_str_buffer);
				}
			}
		}
	}
	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
 ******************************************************************************/

/*
 * pm_session_dettach_me just sets the flag to disable this plugin and no longer call this event callback.
 * Before calling pm_session_dettach_me, the current plugin must release related resources for the current session.
 */
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];

	runtime_me->status = PLUGIN_STATUS_DETTACH_ME;
	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));
}

/*
 * The current plugin(cb2) takes over the current session, the pm_session_take_over setting flag disables other plugins,
 * and the current session does not call other plugins except for the SESSION_EVENT_CLOSING event.
 *
 *                               +-----+   +-----+   +-----+   +-----+
 * Plugin runtime callback list: | cb1 |-->| cb2 |-->| cb3 |-->| cb4 |
 *                               +-----+   +-----+   +-----+   +-----+
 *                                           /|\
 *                                            |
 *                         plugin cb2 run pm_session_take_over
 *
 * A plugin(cb1/cb3/cb4) that is taken over, if the plugin was called before being taken over and has a registered SESSION_EVENT_CLOSING event,
 * it will be called again when the SESSION_EVENT_CLOSING event comes. Otherwise, the plugin will not be called.
 */
void pm_session_take_over(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->status = PLUGIN_STATUS_TAKEN_OVER;
			plugin_manager_log(DEBUG, "%p take over, disable event_cb: %p, session: %s", runtime_me->event_cb, runtime_other->event_cb, stellar_session_get_name(session));
		}
	}
}

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

void *pm_session_get_plugin_ctx(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