/*************************************************************************
    > File Name: lua_binding_functions.c
    > Author:
    > Created Time: 2024-08
    > Encoding : UTF-8
 ************************************************************************/

/*************************************************************************
 * 声明并定义所有需要在lua状态机中绑定的函数
 * version
 * [ v0.1 ]
 * 08-14
 * 1. 实现函数
 *    新增插件注册函数
 *    int lua_plugin_manage_regist
 *    新增会话相关函数
 *    int lua_session_get_id
 *    int lua_session_set_id
 *    int lua_session_get_type
 *    int lua_session_set_type
 *    新增message相关函数
 *    int lua_mq_create_topic
 *    int lua_mq_get_topic_id
 *    int lua_mq_update_topic
 *    int lua_mq_destory_topic
 *    int lua_mq_subscribe_topic
 *    int lua_mq_topic_is_active
 *    int lua_mq_publish_message
 *    int lua_mq_ignore_message
 *    int lua_mq_unignore_message
 * 
 * 08-29
 * 修改create_topic逻辑, 允许创建一个空的topic
 ************************************************************************/
#include "lua_plugin_manage_internal.h"

#include "stellar/session.h"
#include "stellar/session_mq.h"

int global_max_plugin_id = 0;

/* ***** ***** ***** ***** ***** ***** */
int lua_plugin_manage_regist(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TTABLE || lua_type(state, -2) != LUA_TFUNCTION ||
        lua_type(state, -3) != LUA_TFUNCTION || lua_type(state, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 取出处理第四个参数 */
    lua_getfield(state, -1, LUA_PLUGIN_ENV_DEFAULT_KEY); /* stack 4, table中取出对应结构的指针 */
    if (lua_type(state, -1) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }
    struct lua_model *plugin_env = (struct lua_model *)lua_topointer(state, -1);
    lua_pop(state, 2);
    // debug_lua_state_stack(state, 0, "here");
    // printf("env pointer is %p\n", plugin_env);

    /* 取出处理第三个参数 */
    int ctx_free_id = luaL_ref(state, LUA_REGISTRYINDEX); /* stack 3 */
    if (ctx_free_id == LUA_REFNIL)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 取出处理第二个参数 */
    int ctx_new_id = luaL_ref(state, LUA_REGISTRYINDEX); /* stack 2 */
    if (ctx_new_id == LUA_REFNIL)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, ctx_free_id);
        lua_settop(state, 0);
        return 0;
    }

    /* 取出处理第一个参数 */
    struct stellar *st = (struct stellar *)lua_topointer(state, -1); /* stack 1 */
    if (!st)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, ctx_new_id);
        luaL_unref(state, LUA_REGISTRYINDEX, ctx_free_id);
        lua_settop(state, 0);
        return 0;
    }
    lua_pop(state, 1);

    /* 在stellar中注册, 获取注册id */
    int plugin_id = stellar_session_plugin_register(st, lpm_ctx_new_func, lpm_ctx_free_func, (void *)plugin_env);
#ifdef LUAPLUGIN_BASIC_UNITTEST
    LOGDEBUG("now regist new plugin, plugin id is %d, %d, %d\n", plugin_id, ctx_new_id, ctx_free_id);
#endif

    /* TODO: 如果运行完全符合预期的话, 理论上仅有thread 0在此处需要插入新的插件, 且不应该有错误
     * 对于其他线程这里应该直接检查ref id是否一致即可, 按理说不应该再插入新插件
     * 后续可以修改为根据线程号执行不同的处理流程
     */
    /* 如果在其他线程中已经完成过注册 */
    struct lua_plugin *search_plugin = NULL;
    while ((search_plugin = utarray_next(plugin_env->plugin_array, search_plugin)))
    {
        if (search_plugin->plugin_id == plugin_id)
        {
            /* 初始化过程中已经进行过加载 */
            if (search_plugin->ctx_new_ref != ctx_new_id || search_plugin->ctx_free_ref != ctx_free_id)
            {
                LOGERROR("regist plugin, same id with different function ref");
                LOGERROR("plugin id %d, registed %d, %d, new ref %d, %d", plugin_id,
                         search_plugin->ctx_new_ref, search_plugin->ctx_free_ref,
                         ctx_new_id, ctx_free_id);
                lua_settop(state, 0);
                return 0;
            }
            lua_settop(state, 0);
            lua_pushinteger(state, plugin_id);
            return 1;
        }
    }
    /* 统计记录一下当前最大的plugin_id编号 */
    if (plugin_id > global_max_plugin_id)
        global_max_plugin_id = plugin_id;

    /* 将注册完成的新插件插入到队列中 */
    struct lua_plugin new_plugin;
    memset(&new_plugin, 0, sizeof(new_plugin));
    new_plugin.plugin_id = plugin_id;
    new_plugin.ctx_new_ref = ctx_new_id;
    new_plugin.ctx_free_ref = ctx_free_id;
    utarray_push_back(plugin_env->plugin_array, &new_plugin);
    plugin_env->plugin_count += 1;
    global_schema->plugin_count += 1;

    lua_settop(state, 0);
    lua_pushinteger(state, plugin_id);

    return 1;
}

/* ***** ***** ***** ***** ***** ***** */
#if 0
int lua_session_get_id(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 1)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_pop(state, 1);

    lua_pushinteger(state, session_get_id(sess));
    return 1;
}

int lua_session_set_id(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 2)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    int setid = lua_tointeger(state, -1);
    lua_pop(state, 1);
    struct session *sess = (struct session *)lua_topointer(state, -1);
    lua_pop(state, 1);

    session_set_id(sess, setid);
    return 0;
}

int lua_session_set_type(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 2)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    int settype = lua_tointeger(state, -1);
    lua_pop(state, 1);
    struct session *sess = (struct session *)lua_topointer(state, -1);
    lua_pop(state, 1);

    session_set_type(sess, settype);
    return 0;
}
#endif

int lua_session_get_type(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 1)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_pop(state, 1);

    lua_pushinteger(state, (int)session_get_type(sess));
    return 1;
}

/* ***** ***** ***** ***** ***** ***** */
/*
 * TODO: 未完整考虑线程安全问题, 例如
 * 多个线程同时注册一个topic, 是否需要做处理等。
 */
static UT_icd lua_plugin_mq_icd = {sizeof(struct lua_plugin_mq), NULL, NULL, NULL};

int lua_mq_create_topic(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (((lua_type(state, -1) != LUA_TTABLE || lua_type(state, -2) != LUA_TFUNCTION) &&
         (lua_type(state, -1) != LUA_TNIL || lua_type(state, -2) != LUA_TNIL)) ||
        lua_type(state, -3) != LUA_TSTRING || lua_type(state, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 创建对该table的引用, 防止该table被回收 */
    int private_ref = 0;
    if (lua_type(state, -1) != LUA_TNIL)
    {
        private_ref = luaL_ref(state, LUA_REGISTRYINDEX); /* stack top function */
        if (private_ref == LUA_REFNIL)
        {
            lua_settop(state, 0);
            return 0;
        }
    }

    /* 导出free message的调用函数 */
    int free_ref = 0;
    if (lua_type(state, -1) != LUA_TFUNCTION)
    {
        free_ref = luaL_ref(state, LUA_REGISTRYINDEX);
        if (free_ref == LUA_REFNIL)
        {
            luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
            lua_settop(state, 0);
            return 0;
        }
    }

    /* 出栈队列名称 */
    char *name = strdup((char *)lua_tostring(state, -1));
    lua_pop(state, 1);

    /* 出栈传入的stellar */
    struct stellar *st = (struct stellar *)lua_topointer(state, -1);
    if (!st)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, free_ref);
        luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
        if (name)
            free(name);
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    /* 插入新的元素 */
    int topic_id = -1;
    if (private_ref && free_ref)
    {
        struct lua_message_mq new_mq;
        memset(&new_mq, 0, sizeof(new_mq));
        utarray_push_back(global_schema->message_mq_array, &new_mq);
        /* 从队列尾部取出最后一个元素 */
        struct lua_message_mq *mq = utarray_eltptr(global_schema->message_mq_array, (utarray_len(global_schema->message_mq_array) - 1));
        mq->freemessage_ref = free_ref;
        mq->mq_private_ref = private_ref;
        /* 调用stellar中mq的topic创建函数 */
        /* BugFix: 仔细看了代码, 在stellar中没有对name做拷贝处理, 这里name不能释放 */
        topic_id = stellar_session_mq_create_topic(st, (const char *)name, lpm_message_free_func, mq);

        /* 创建topic失败, 还原创建topic之前的状态 */
        if (topic_id < 0)
        {
            utarray_pop_back(global_schema->message_mq_array);
            luaL_unref(state, LUA_REGISTRYINDEX, free_ref);
            luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
            if (name)
                free(name);
            return 0;
        }
        global_schema->mq_count += 1;

        lua_rawgeti(state, LUA_REGISTRYINDEX, mq->mq_private_ref);
#if 0
    /* 不检查了, 直接覆盖 */
    /* 在传入的table中检查是否存在与mq_private_env同名的元素 */
    lua_getfield(state, -1, LUA_MQ_ENV_DEFAULT_KEY); /* stack top nil */
    if (lua_type(state, -1) != LUA_TNIL) {
        lua_settop(state, 0);
        return 0;
    }
    lua_pop(state, 1); /* stack top table */
#endif
        /* 在该table中加入新元素 */
        lua_pushlightuserdata(state, (void *)mq);        /* stack top new_mq */
        lua_setfield(state, -2, LUA_MQ_ENV_DEFAULT_KEY); /* stack top table */
        mq->topic_id = topic_id;
    }
    else
    {
        /* 只是创建一个新的空topic */
        topic_id = stellar_session_mq_create_topic(st, (const char *)name, NULL, NULL);
        if (topic_id < 0)
        {
            if (name)
                free(name);
            return 0;
        }
    }

    lua_settop(state, 0);
    lua_pushinteger(state, topic_id);

    return 1;
}

int lua_mq_get_topic_id(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 2)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TSTRING || lua_type(state, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 出栈队列名称 */
    char *name = strdup(lua_tostring(state, -1));
    lua_pop(state, 1);

    /* 出栈传入的stellar */
    struct stellar *st = (struct stellar *)lua_topointer(state, -1);
    if (!st)
    {
        if (name)
            free(name);
        return 0;
    }
    lua_pop(state, 1);

    int topic_id = stellar_session_mq_get_topic_id(st, (const char *)name);
    if (name)
        free(name);

    lua_settop(state, 0);
    lua_pushinteger(state, topic_id);
    return 1;
}

int lua_mq_update_topic(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TTABLE || lua_type(state, -2) != LUA_TFUNCTION ||
        lua_type(state, -3) != LUA_TNUMBER || lua_type(state, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 创建对该table的引用, 防止该table被回收 */
    int private_ref = luaL_ref(state, LUA_REGISTRYINDEX); /* stack top function */
    if (private_ref == LUA_REFNIL)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 导出free message的调用函数 */
    int free_ref = luaL_ref(state, LUA_REGISTRYINDEX);
    if (free_ref == LUA_REFNIL)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
        lua_settop(state, 0);
        return 0;
    }

    /* topic_id */
    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    /* 出栈传入的stellar */
    struct stellar *st = (struct stellar *)lua_topointer(state, -1);
    if (!st)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, free_ref);
        luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    struct lua_message_mq *mq = search_message_mq_by_id(topic_id);
    if (!mq || mq->topic_id != topic_id)
    {
        /* 如果topic不是lua创建的, 需要加入管理 */
        struct lua_message_mq new_mq;
        memset(&new_mq, 0, sizeof(new_mq));
        utarray_push_back(global_schema->message_mq_array, &new_mq);
        mq = utarray_eltptr(global_schema->message_mq_array, (utarray_len(global_schema->message_mq_array) - 1));

        if (stellar_session_mq_update_topic(st, topic_id, lpm_message_free_func, mq))
        {
            utarray_pop_back(global_schema->message_mq_array);
            luaL_unref(state, LUA_REGISTRYINDEX, free_ref);
            luaL_unref(state, LUA_REGISTRYINDEX, private_ref);
            return 0;
        }
#ifdef LUAPLUGIN_BASIC_UNITTEST
        LOGDEBUG("update topic, id %d %d %d", topic_id, private_ref, free_ref);
#endif

        mq->freemessage_ref = free_ref;
        mq->mq_private_ref = private_ref;
        mq->topic_id = topic_id;
        lua_pushboolean(state, 1);
        return 1;
    }
    else
    {
        /* 本身是由lua创建的 */
        /* 更新private_ref */
        lua_rawgeti(state, LUA_REGISTRYINDEX, private_ref);
        lua_pushlightuserdata(state, (void *)mq);        /* stack top new_mq */
        lua_setfield(state, -2, LUA_MQ_ENV_DEFAULT_KEY); /* stack top table */
        // luaL_unref(state, LUA_REGISTRYINDEX, mq->mq_private_ref);
        mq->mq_private_ref = private_ref;
        /* 更新free function ref */
        // luaL_unref(state, LUA_REGISTRYINDEX, mq->freemessage_ref);
        mq->freemessage_ref = free_ref;

#ifdef LUAPLUGIN_BASIC_UNITTEST
        LOGDEBUG("update topic, id %d %d %d", topic_id, private_ref, free_ref);
#endif
    }

    lua_settop(state, 0);
    lua_pushboolean(state, 1);

    return 1;
}

int lua_mq_destory_topic(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 2)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* topic_id */
    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    /* 出栈传入的stellar */
    struct stellar *st = (struct stellar *)lua_topointer(state, -1);
    if (!st)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    /* 优先调用C函数进行卸载 */
    if (stellar_session_mq_destroy_topic(st, topic_id) < 0)
    {
        return 0;
    }

    /* 不方便删除, 将id置为-1, 确保匹配不到 */
    struct lua_message_mq *mq = search_message_mq_by_id(topic_id);
    if (mq)
    {
        mq->topic_id = -1;
        luaL_unref(state, LUA_REGISTRYINDEX, mq->mq_private_ref);
        luaL_unref(state, LUA_REGISTRYINDEX, mq->freemessage_ref);
        mq->freemessage_ref = 0;
        mq->mq_private_ref = 0;
    }

    lua_pushboolean(state, 1);
    return 1;
}

int lua_mq_subscribe_topic(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TFUNCTION ||
        lua_type(state, -3) != LUA_TNUMBER || lua_type(state, -4) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 读取参数 */
    int plugin_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    int on_message_ref = luaL_ref(state, LUA_REGISTRYINDEX);
    if (on_message_ref == LUA_REFNIL)
    {
        lua_settop(state, 0);
        return 0;
    }

    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    struct stellar *st = (struct stellar *)lua_topointer(state, -1);
    if (!st)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_pop(state, 1);

    lua_settop(state, 0);
    if (stellar_session_mq_subscribe(st, topic_id, lpm_on_session_msg_func, plugin_id))
    {

        /* 订阅失败, 返回false */
        lua_pushboolean(state, 0);
    }
    else
    {
        struct lua_plugin *plugin = search_plugin_by_id(plugin_id);
        if (plugin)
        {
            if (!plugin->sub_topic_array)
            {
                /* 该插件尚未注册任何topic, 注册第一个topic时将创建接收的topic列表 */
                utarray_new(plugin->sub_topic_array, &lua_plugin_mq_icd);
            }
            else
            {
                /* 如果该插件中之前已经订阅过该消息, 更新message函数 */
                struct lua_plugin_mq *mq = NULL;
                while ((mq = utarray_next(plugin->sub_topic_array, mq)))
                {
                    if (mq->topic_id == topic_id)
                    {
                        luaL_unref(state, LUA_REGISTRYINDEX, mq->onmessage_ref);
                        mq->onmessage_ref = on_message_ref;
                        lua_pushboolean(state, 1);
                        return 1;
                    }
                }
            }

            struct lua_plugin_mq new_mq;
            memset(&new_mq, 0, sizeof(new_mq));
            new_mq.topic_id = topic_id;
            new_mq.onmessage_ref = on_message_ref;
            utarray_push_back(plugin->sub_topic_array, &new_mq);
        }
        /* 订阅成功, 返回true */
        lua_pushboolean(state, 1);
    }

    return 1;
}

int lua_mq_publish_message(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 3)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TTABLE || lua_type(state, -2) != LUA_TNUMBER ||
        lua_type(state, -3) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 倒序依次获取参数 */
    int mess_ref = luaL_ref(state, LUA_REGISTRYINDEX);
    if (mess_ref == LUA_REFNIL)
    {
        lua_settop(state, 0);
        return 0;
    }

    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        luaL_unref(state, LUA_REGISTRYINDEX, mess_ref);
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    /* 创建一段数据引用 */
    struct lua_context *new_context = (struct lua_context *)calloc(1, sizeof(struct lua_context));
    if (__glibc_unlikely(!new_context))
    {
        luaL_unref(state, LUA_REGISTRYINDEX, mess_ref);
        return 0;
    }
    new_context->context_ref_id = mess_ref;

    /* 调用C接口发布消息 */
    if (session_mq_publish_message(sess, topic_id, new_context))
    {
        luaL_unref(state, LUA_REGISTRYINDEX, new_context->context_ref_id);
        free(new_context);
        lua_pushboolean(state, 0);
    }
    else
    {
        lua_pushboolean(state, 1);
    }

    return 1;
}

int lua_mq_ignore_message(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TNUMBER ||
        lua_type(state, -3) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 倒序获取参数 */
    int plugin_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    if (session_mq_ignore_message(sess, topic_id, plugin_id))
        lua_pushboolean(state, 0);
    else
        lua_pushboolean(state, 1);

    return 1;
}

int lua_mq_unignore_message(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TNUMBER ||
        lua_type(state, -3) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 倒序获取参数 */
    int plugin_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    if (session_mq_unignore_message(sess, topic_id, plugin_id))
        lua_pushboolean(state, 0);
    else
        lua_pushboolean(state, 1);

    return 1;
}

int lua_mq_topic_is_active(lua_State *state)
{
    /* 参数个数检查 */
    if (lua_gettop(state) != 4)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 参数类型检查 */
    if (lua_type(state, -1) != LUA_TNUMBER || lua_type(state, -2) != LUA_TLIGHTUSERDATA)
    {
        lua_settop(state, 0);
        return 0;
    }

    /* 倒序获取参数 */
    int topic_id = lua_tointeger(state, -1);
    lua_pop(state, 1);

    struct session *sess = (struct session *)lua_topointer(state, -1);
    if (!sess)
    {
        lua_settop(state, 0);
        return 0;
    }
    lua_settop(state, 0);

    /* 1 means active */
    if (session_mq_topic_is_active(sess, topic_id) == 1)
        lua_pushboolean(state, 1);
    else
        lua_pushboolean(state, 0);

    return 1;
}