#include "common.h"

#include "utils.h"

#include "lbuffer.h"

#include "context.h"

#include "context_lua.h"

#include "node_lua.h"

#include "worker.h"

#include "uv_handle_base.h"

#include "lua_tcp_handle.h"

#include "lua_udp_handle.h"

#include "lua_timer_handle.h"

#include "errors.h"

char context_lua_t::m_root_coro_table_key = 0;

char context_lua_t::m_ref_session_table_key = 0;

char context_lua_t::m_ref_timer_table_key = 0;

context_lua_t::context_lua_t()

: m_lstate(NULL),

m_ref_session_count(0),

m_ref_session_base(INT_MAX)

{

}

context_lua_t::~context_lua_t()

{

if (m_lstate) {

lua_close(m_lstate);

m_lstate = NULL;

}

}

bool context_lua_t::init(int32_t argc, char* argv[], char* env[])

{

m_lstate = luaL_newstateex(this);

if (argc <= 0) {

printf("[error] context:0x%08x init failed: lua file is needed to initialize lua context\n", m_handle); /* to be implemented : put reason to another context and output it */

return false;

}

int32_t envc = 2;

if (argc > MAX_CTX_ARGC || !lua_checkstack(m_lstate, argc + envc)) {

printf("[error] context:0x%08x init failed: too many arguments to initialize lua context %s\n", m_handle, argv[0]); /* to be implemented : put reason to another context and output it */

return false;

}

int32_t total_length = 0;

int32_t lengths[MAX_CTX_ARGC];

for (int32_t i = 0; i < argc; ++i) {

lengths[i] = strlen(argv[i]);

lua_pushlstring(m_lstate, argv[i], lengths[i]);

total_length += lengths[i] + 1;

}

char** envp = env;

char* lua_path = NULL;

char* lua_cpath = NULL;

while (*envp) {

if (strncmp(*envp, "LUA_PATH=", 9) == 0) {

lua_path = *envp + 9;

} else if (strncmp(*envp, "LUA_CPATH=", 10) == 0) {

lua_cpath = *envp + 10;

}

++envp;

}

lua_pushstring(m_lstate, lua_path);

lua_pushstring(m_lstate, lua_cpath);

char* args = (char*)nl_calloc(total_length, 1);

char* argp = args;

for (int32_t i = 0; i < argc; ++i) {

memcpy(argp, argv[i], lengths[i]);

argp += lengths[i];

*argp++ = (i != argc - 1) ? ' ' : '\0';

}

printf("[alert] context:0x%08x init: %s\n", m_handle, args); /* to be implemented : put reason to another context and output it */

nl_free(args);

return singleton_ref(node_lua_t).context_send(this, m_handle, 0, LUA_CTX_INIT, (int64_t)argc);

}

bool context_lua_t::deinit(const char *arg)

{

for (std::set<uint32_t>::iterator it = m_context_wait_handles.begin(); it != m_context_wait_handles.end(); ++it) {

singleton_ref(node_lua_t).context_send(*it, m_handle, LUA_REFNIL, LUA_CTX_WAKEUP, UV_OK);

}

m_context_wait_handles.clear();

if (m_lstate) {

lua_close(m_lstate);

m_lstate = NULL;

}

printf("[alert] context:0x%08x deinit: %s\n", m_handle, arg); /* to be implemented : put reason to another context and output it */

return true;

}

void context_lua_t::on_worker_attached()

{

m_shared = m_worker->m_context_lua_shared;

m_shared->m_yielding_destination = 0;

m_shared->m_yielding_depth = 0;

m_shared->m_yielding_finalize = NULL;

m_shared->m_yielding_finalize_userdata = NULL;

m_shared->m_yielding_status = UV_OK;

m_shared->m_yielding_timeout = 0;

}

void context_lua_t::on_worker_detached()

{

m_shared = NULL;

}

#define LUA_COLIBNAME "coroutine"

LUAMOD_API int (luaopen_my_coroutine)(lua_State *L);

#define LUA_TCPLIBNAME "tcp"

LUAMOD_API int (luaopen_tcp)(lua_State *L);

#define LUA_UDPLIBNAME "udp"

LUAMOD_API int (luaopen_udp)(lua_State *L);

#define LUA_BUFFERLIBNAME "buffer"

LUAMOD_API int (luaopen_buffer)(lua_State *L);

#define LUA_CONTEXTLIBNAME "context"

LUAMOD_API int (luaopen_context)(lua_State *L);

#define LUA_TIMERLIBNAME "timer"

LUAMOD_API int (luaopen_timer)(lua_State *L);

#define LUA_BSONLIBNAME "bson"

LUAMOD_API int (luaopen_bson)(lua_State *L);

void context_lua_t::lua_open_libs(lua_State *L)

{

static const luaL_Reg loadedlibs[] = {

{"_G", luaopen_base},

{LUA_LOADLIBNAME, luaopen_package},

{LUA_COLIBNAME, luaopen_my_coroutine},

{LUA_TABLIBNAME, luaopen_table},

{LUA_IOLIBNAME, luaopen_io},

{LUA_OSLIBNAME, luaopen_os},

{LUA_STRLIBNAME, luaopen_string},

{LUA_BITLIBNAME, luaopen_bit32},

{LUA_MATHLIBNAME, luaopen_math},

{LUA_DBLIBNAME, luaopen_debug},

#if defined(LUA_CACHELIB)

{LUA_CACHELIB, luaopen_cache},

#endif

{LUA_TCPLIBNAME, luaopen_tcp},

{LUA_UDPLIBNAME, luaopen_udp},

{LUA_BUFFERLIBNAME, luaopen_buffer},

{LUA_CONTEXTLIBNAME, luaopen_context},

{LUA_TIMERLIBNAME, luaopen_timer},

{LUA_BSONLIBNAME, luaopen_bson},

{NULL, NULL}

};

static const luaL_Reg preloadedlibs[] = {

{ NULL, NULL }

};

const luaL_Reg *lib;

/* call open functions from 'loadedlibs' and set results to global table */

for (lib = loadedlibs; lib->func; lib++) {

luaL_requiref(L, lib->name, lib->func, 1);

lua_pop(L, 1); /* remove lib */

}

/* add open functions from 'preloadedlibs' into 'package.preload' table */

luaL_getsubtable(L, LUA_REGISTRYINDEX, "_PRELOAD");

for (lib = preloadedlibs; lib->func; lib++) {

lua_pushcfunction(L, lib->func);

lua_setfield(L, -2, lib->name);

}

lua_pop(L, 1); /* remove _PRELOAD table */

}

void context_lua_t::unload()

{

#if defined(LUA_CACHELIB)

luaL_closesharedclosure();

#endif

}

void context_lua_t::lua_load_env(lua_State *L, char* env[])

{

lua_checkstack(L, 3);

luaL_getsubtable(L, LUA_REGISTRYINDEX, "_LOADED");

lua_getfield(L, -1, LUA_LOADLIBNAME); /* _LOADED[name] */

if (lua_istable(L, -1)) {

char** envp = env;

lua_getfield(L, -1, "path");

size_t path_len;

const char* lua_path = lua_tolstring(L, -1, &path_len);

if (lua_path != NULL && path_len > 0) {

char* lua_path_env = (char*)nl_malloc(path_len + 10);

strcpy(lua_path_env, "LUA_PATH=");

strcpy(lua_path_env + 9, lua_path);

*envp++ = lua_path_env;

}

lua_pop(L, 1);

lua_getfield(L, -1, "cpath");

size_t cpath_len;

const char* lua_cpath = lua_tolstring(L, -1, &cpath_len);

if (lua_cpath != NULL && cpath_len > 0) {

char* lua_cpath_env = (char*)nl_malloc(cpath_len + 11);

strcpy(lua_cpath_env, "LUA_CPATH=");

strcpy(lua_cpath_env + 10, lua_cpath);

*envp++ = lua_cpath_env;

}

lua_pop(L, 3);

return;

}

lua_pop(L, 2);

return;

}

void context_lua_t::lua_init_env(lua_State *L, int32_t envc)

{

if (envc <= 0) return;

const char* path = lua_tostring(L, -envc);

const char* cpath = lua_tostring(L, -envc + 1);

if (path == NULL && cpath == NULL) {

lua_pop(L, envc);

return;

}

lua_checkstack(L, 3);

luaL_getsubtable(L, LUA_REGISTRYINDEX, "_LOADED");

lua_getfield(L, -1, LUA_LOADLIBNAME); /* _LOADED[name] */

if (lua_istable(L, -1)) {

if (path) {

lua_pushvalue(L, -envc - 2);

lua_setfield(L, -2, "path");

}

if (cpath) {

lua_pushvalue(L, -envc - 1);

lua_setfield(L, -2, "cpath");

}

}

lua_pop(L, envc + 2);

return;

}

void context_lua_t::lua_free_env(lua_State *L, char* env[])

{

char** envp = env;

while (*envp) {

nl_free((void*)*envp);

*envp++ = NULL;

}

}

void context_lua_t::lua_init(lua_State *L)

{

lua_open_libs(L);

lua_pushlightuserdata(L, &m_root_coro_table_key); /* create root coro table */

lua_createtable(L, 0, 4);

lua_rawset(L, LUA_REGISTRYINDEX);

lua_pushlightuserdata(L, &m_ref_session_table_key);

lua_createtable(L, 4, 0);

lua_rawset(L, LUA_REGISTRYINDEX);

lua_pushlightuserdata(L, &m_ref_timer_table_key);

lua_createtable(L, 4, 0);

lua_rawset(L, LUA_REGISTRYINDEX);

}

lua_State* context_lua_t::lua_new_root_coro(lua_State *L)

{

context_lua_t* lctx = context_lua_t::lua_get_context(L);

if (!lctx->m_cached_root_coros.empty()) {

lua_State *co = lctx->m_cached_root_coros.back();

lctx->m_cached_root_coros.pop_back();

return co;

}

lua_checkstack(L, 3);

lua_pushlightuserdata(L, &m_root_coro_table_key);

lua_rawget(L, LUA_REGISTRYINDEX);

lua_State *co = lua_newthread(L);

lua_pushboolean(L, 1);

lua_rawset(L, -3);

lua_pop(L, 1); /* pop root_coro_table */

return co;

}

void context_lua_t::lua_free_root_coro(lua_State *co)

{

context_lua_t* lctx = context_lua_t::lua_get_context(co);

if (lua_status(co) == LUA_OK && lctx->m_cached_root_coros.size() < CONTEXT_ROOT_CORO_CACHE_MAX) {

lctx->m_cached_root_coros.push_back(co);

return;

}

lua_checkstack(co, 3);

lua_pushlightuserdata(co, &m_root_coro_table_key);

lua_rawget(co, LUA_REGISTRYINDEX);

lua_pushthread(co);

lua_pushnil(co);

lua_rawset(co, -3);

lua_pop(co, 1); /* pop root_coro_table */

}

int32_t context_lua_t::lua_ref_yield_coroutine(lua_State *co)

{

lua_checkstack(co, 2);

lua_pushlightuserdata(co, &m_ref_session_table_key);

lua_rawget(co, LUA_REGISTRYINDEX);

lua_pushthread(co);

int32_t session = lua_ref_session(co, -2);

lua_pop(co, 1); /* pop ref_session_table */

++lua_get_context(co)->m_ref_session_count;

return session;

}

int32_t context_lua_t::lua_ref_timer(lua_State *L, int32_t session, uint64_t timeout, uint64_t repeat, bool is_cancel, void *userdata, timeout_callback_t callback)

{

lua_checkstack(L, 2);

lua_pushlightuserdata(L, &m_ref_timer_table_key);

lua_rawget(L, LUA_REGISTRYINDEX);

lua_timer_handle_t* timer = lua_timer_handle_t::create_timer(L, session, timeout, repeat, is_cancel, userdata, callback);

lua_rawseti(L, -2, session);

lua_pop(L, 1); /* pop ref_timer_table */

return timer->get_handle_ref();

}

void context_lua_t::lua_unref_timer(lua_State *L, int32_t session)

{

lua_checkstack(L, 2);

lua_pushlightuserdata(L, &m_ref_timer_table_key);

lua_rawget(L, LUA_REGISTRYINDEX);

lua_rawgeti(L, -1, session);

lua_timer_handle_t* timer = (lua_timer_handle_t*)lua_touserdata(L, -1);

if (timer) {

timer->close();

lua_pop(L, 1);

lua_pushnil(L);

lua_rawseti(L, -2, session);

lua_pop(L, 1);

} else {

lua_pop(L, 2);

}

}

int32_t context_lua_t::yielding_finalize(lua_State *root_coro, int32_t status)

{

if (!root_coro) {

ASSERT(status != UV_OK); /* error occurs directly, status mustn't be UV_OK! */

m_shared->m_yielding_status = status;

}

if (m_shared->m_yielding_finalize != NULL) {

int32_t fnz_status = m_shared->m_yielding_finalize(root_coro, m_lstate, m_shared->m_yielding_finalize_userdata, m_shared->m_yielding_destination);

if (root_coro) m_shared->m_yielding_status = fnz_status;

m_shared->m_yielding_finalize = NULL;

}

m_shared->m_yielding_depth = 0;

m_shared->m_yielding_destination = 0;

m_shared->m_yielding_finalize_userdata = NULL;

m_shared->m_yielding_timeout = 0;

return m_shared->m_yielding_status;

}

int32_t context_lua_t::common_yield_continue(lua_State* L, int status, lua_KContext ctx)

{

int32_t old_top = (int32_t)ctx;

if (!lua_toboolean(L, old_top + 1) && context_lua_t::lua_get_context(L)->get_yielding_status() != UV_OK) {

context_lua_t::lua_throw(L, old_top + 2);

}

lua_pop(L, 2); //pop out two useless arguments

return lua_gettop(L) - old_top;

}

int32_t context_lua_t::lua_yield_send(lua_State *L, uint32_t destination, yiled_finalize_t fnz, void* fnz_ud, lua_KFunction yieldk, int64_t timeout)

{

int stacks;

context_lua_t* lctx = context_lua_t::lua_get_context(L);

lctx->m_shared->m_yielding_destination = destination;

lctx->m_shared->m_yielding_depth = 1;

lctx->m_shared->m_yielding_finalize = fnz; /* fnz should not be null better */

lctx->m_shared->m_yielding_finalize_userdata = fnz_ud;

lctx->m_shared->m_yielding_status = UV_OK;

lctx->m_shared->m_yielding_timeout = timeout;

if ((stacks = lua_checkstack(L, LUA_MINSTACK)) && lua_yieldable(L)) { /* reserve LUA_MINSTACK stack for resume result */

return lua_yieldk(L, 0, (lua_KContext)lua_gettop(L), yieldk);

} else {

if (stacks) { /* stack enough, but not yield-able */

lctx->yielding_finalize(NULL, NL_EYIELD);

int32_t top = lua_gettop(L);

lua_pushboolean(L, 0);

lua_pushinteger(L, NL_EYIELD);

lua_pushnil(L);

lua_pushnil(L);

return (yieldk == NULL) ? 4 : yieldk(L, LUA_YIELD, (lua_KContext)top);

} else { /* error jump directly */

lctx->yielding_finalize(NULL, NL_ESTACKLESS);

return luaL_error(L, "attempt to yield across a stack-less coroutine");

}

}

}

int32_t context_lua_t::common_callback_adjust(lua_State* L)

{

lua_pop(L, 2);

return 2;

}

int32_t context_lua_t::lua_ref_callback_entry_finish(lua_State *L, int32_t status)

{

if (!status) { /* error message is at top */

uint32_t handle = lua_get_context_handle(L);

const char* error = lua_tostring(L, -1);

/* to be implemented : put reason to another context and output it */

printf("[error] context:0x%08x error: %s\n", handle, error ? error : "unknown error");

}

return 0; //return none result out!!!

}

int32_t context_lua_t::lua_ref_callback_entry_continue(lua_State *L, int status, lua_KContext ctx)

{

return lua_ref_callback_entry_finish(L, (status == LUA_YIELD));

}

int32_t context_lua_t::lua_ref_callback_entry(lua_State *L)

{

int upvalue_count = 0;

while (!lua_isnone(L, lua_upvalueindex(upvalue_count + 1))) {

++upvalue_count;

}

ASSERT(upvalue_count > 1);

lua_CFunction adjust_resume_args = (lua_CFunction)lua_touserdata(L, lua_upvalueindex(upvalue_count));

if (adjust_resume_args) {

adjust_resume_args(L);

}

--upvalue_count;

int resume_count = lua_gettop(L);

lua_checkstack(L, upvalue_count);

lua_pushvalue(L, lua_upvalueindex(upvalue_count));

lua_insert(L, 1); /* put callback closure as first arg */

for (int i = 1; i < upvalue_count; ++i) {

lua_pushvalue(L, lua_upvalueindex(i));

}

/* lua_callback_entry_continue makes the coroutine can be recovered. */

int status = lua_pcallk(L, upvalue_count + resume_count - 1, LUA_MULTRET, 0, 0, lua_ref_callback_entry_continue);

return lua_ref_callback_entry_finish(L, (status == LUA_OK));

}

int32_t context_lua_t::lua_ref_session(lua_State *L, int32_t idx)

{

if (lua_isnil(L, -1)) {

lua_pop(L, 1); /* remove from stack */

return LUA_REFNIL; /* `nil' has a unique fixed reference */

}

context_lua_t* lctx = lua_get_context(L);

idx = lua_absindex(L, idx);

for (int i = 0; i < INT_MAX; ++i) {

if (lctx->m_ref_session_base > 0) {

lua_rawgeti(L, idx, lctx->m_ref_session_base);

if (lua_isnil(L, -1)) {

lua_pop(L, 1);

lua_rawseti(L, idx, lctx->m_ref_session_base);

return lctx->m_ref_session_base--;

}

lua_pop(L, 1);

--lctx->m_ref_session_base;

} else {

lctx->m_ref_session_base = INT_MAX;

}

}

lua_pop(L, 1); /* remove from stack */

return LUA_REFNIL; /* `nil' has a unique fixed reference */

}

void context_lua_t::lua_unref_session(lua_State *L, int32_t idx, int32_t session)

{

if (session != LUA_REFNIL) {

idx = lua_absindex(L, idx);

lua_pushnil(L);

lua_rawseti(L, idx, session);

}

}

/* ref a new callback (remove argc args and the top closure) */

int32_t context_lua_t::lua_ref_callback(lua_State *L, uint32_t argc, int32_t ref, lua_CFunction adjust_resume_args)

{

luaL_checktype(L, -1, LUA_TFUNCTION);

lua_checkstack(L, 2);

lua_pushlightuserdata(L, (void*)adjust_resume_args);

lua_pushcclosure(L, lua_ref_callback_entry, argc + 2); /* make the c closure callback */

lua_pushlightuserdata(L, &m_ref_session_table_key);

lua_rawget(L, LUA_REGISTRYINDEX); /* fetch callback_func_table */

lua_pushvalue(L, -2); /* put the c closure callback at top */

if (ref == LUA_REFNIL) {

ref = lua_ref_session(L, -2);

++lua_get_context(L)->m_ref_session_count;

lua_pop(L, 2); /* pop callback_func_table and the c closure callback */

} else {

lua_rawseti(L, -2, ref);

lua_pop(L, 2); /* pop callback_func_table and the c closure callback */

lua_unref_timer(L, ref);

}

return ref;

}

context_lua_t* context_lua_t::lua_get_context(lua_State *L)

{

context_lua_t* ctx;

lua_getallocf(L, (void**)&ctx);

return ctx;

}

uint32_t context_lua_t::lua_get_context_handle(lua_State *L)

{

return lua_get_context(L)->m_handle;

}

int32_t context_lua_t::lua_strerror(lua_State *L, int32_t idx)

{

lua_checkstack(L, 1);

lua_pushstring(L, common_strerror(lua_tointeger(L, idx)));

return 1;

}

void context_lua_t::lua_throw(lua_State *L, int32_t idx)

{

lua_strerror(L, idx);

lua_error(L);

}

///////////////////////////////////////////////////////////////////////////////////////////

void context_lua_t::lua_push_ref_session(lua_State *L, int32_t ref)

{

if (ref != LUA_REFNIL) {

lua_checkstack(L, 2);

lua_pushlightuserdata(L, &m_ref_session_table_key);

lua_rawget(L, LUA_REGISTRYINDEX);

lua_rawgeti(L, -1, ref);

lua_remove(L, -2);

} else {

lua_pushnil(L);

}

}

void context_lua_t::lua_free_ref_session(lua_State *L, int32_t ref)

{

if (ref != LUA_REFNIL) {

lua_checkstack(L, 1);

lua_pushlightuserdata(L, &m_ref_session_table_key);

lua_rawget(L, LUA_REGISTRYINDEX);

lua_unref_session(L, -1, ref);

lua_pop(L, 1); /* pop ref_session_table and top coroutine */

--lua_get_context(L)->m_ref_session_count;

lua_unref_timer(L, ref);

}

}

void context_lua_t::lua_pushmessage(lua_State *L, message_t& message)

{

message_array_t *array = NULL;

switch (message_data_type(message)) {

case NIL:

lua_pushnil(L);

return;

case TBOOLEAN:

lua_pushboolean(L, message_bool(message));

return;

case USERDATA:

lua_pushlightuserdata(L, message_userdata(message));

return;

case NUMBER:

lua_pushnumber(L, message_number(message));

return;

case INTEGER:

lua_pushinteger(L, message_integer(message));

return;

case BUFFER:

create_buffer(L, message_buffer(message));

return;

case STRING:

lua_pushstring(L, message_string(message));

return;

case BSON:

if (message_bson(message)->extract) {

bson_decode(message_bson(message), L);

} else {

create_bson(message_bson(message), L);

}

return;

case TERROR:

lua_pushinteger(L, message_error(message));

return;

case ARRAY:

array = message_array(message);

if (array != NULL && array->m_count > 0) {

lua_checkstack(L, array->m_count);

for (int32_t i = 0; i < array->m_count; ++i) { //extract them

lua_pushmessage(L, array->m_array[i]);

}

} else {

lua_pushnil(L);

}

return;

default:

lua_pushnil(L);

return;

}

}

int32_t context_lua_t::resume_coroutine(lua_State *co, int32_t narg)

{

int32_t status, yieldup_status;

while ((status = lua_resume(co, m_lstate, narg)) == LUA_YIELD) {

ASSERT(yielding_up()); /* must be yielding up(root coroutine mustn't yield directly) */

if ((yieldup_status = yielding_finalize(co, UV_OK)) == UV_OK) {

return 1;

} else {

lua_pushboolean(co, 0);

lua_pushinteger(co, yieldup_status);

lua_pushnil(co);

lua_pushnil(co);

narg = 4;

}

}

ASSERT(status == LUA_OK);

lua_settop(co, 0);

lua_free_root_coro(co);

return status == LUA_OK;

}

int32_t context_lua_t::wakeup_ref_session(int32_t ref, message_t& message, bool free_callback)

{

if (ref != LUA_REFNIL && m_lstate != NULL) {

lua_push_ref_session(m_lstate, ref);

int32_t type = lua_type(m_lstate, -1);

if (type == LUA_TNIL) {

lua_pop(m_lstate, 1);

return 0;

}

lua_State *co;

bool blocking = (type == LUA_TTHREAD);

if (blocking) {

co = lua_tothread(m_lstate, -1);

lua_pop(m_lstate, 1);

} else {

co = lua_new_root_coro(m_lstate);

lua_xmove(m_lstate, co, 1);

}

if (blocking || free_callback) {

lua_free_ref_session(m_lstate, ref);

}

int32_t oldTop = lua_gettop(co);

lua_checkstack(co, 2);

lua_pushboolean(co, !message_is_pure_error(message));

lua_pushmessage(co, message);

lua_checkstack(co, 2);

lua_pushinteger(co, message.m_source);

lua_pushinteger(co, message.m_session);

int32_t newTop = lua_gettop(co);

resume_coroutine(co, newTop - oldTop);

return 1;

}

return 0;

}

void context_lua_t::on_received(message_t& message)

{

switch (message_raw_type(message)) {

case LUA_CTX_INIT:

lua_ctx_init(message);

return;

case CONTEXT_QUERY:

response_context_query(message);

break;

case CONTEXT_REPLY:

response_context_reply(message);

break;

case LUA_CTX_WAIT:

response_context_wait(message);

break;

case LUA_CTX_WAKEUP:

response_context_wakeup(message);

break;

case RESPONSE_TCP_LISTEN:

response_tcp_listen(message);

break;

case RESPONSE_TCP_ACCEPT:

response_tcp_accept(message);

break;

case RESPONSE_TCP_CONNECT:

response_tcp_connect(message);

break;

case RESPONSE_TCP_READ:

response_tcp_read(message);

break;

case RESPONSE_TCP_WRITE:

response_tcp_write(message);

break;

case RESPONSE_TCP_CLOSING:

response_tcp_closing(message);

break;

case RESPONSE_UDP_OPEN:

response_udp_open(message);

break;

case RESPONSE_UDP_WRITE:

response_udp_write(message);

break;

case RESPONSE_UDP_READ:

response_udp_read(message);

break;

case RESPONSE_UDP_CLOSING:

response_udp_closing(message);

break;

case RESPONSE_HANDLE_CLOSE:

response_handle_close(message);

break;

case RESPONSE_TIMEOUT:

response_timeout(message);

break;

case SYSTEM_CTX_DESTROY:

singleton_ref(node_lua_t).context_destroy(this, message.m_source, message_string(message));

return;

default:

break;

}

if (!is_active()) {

singleton_ref(node_lua_t).context_destroy(this, m_handle, "lua context exit normally");

}

}

void context_lua_t::on_dropped(message_t& message)

{

switch (message_raw_type(message)) {

case RESPONSE_TCP_LISTEN:

case RESPONSE_TCP_ACCEPT:

case RESPONSE_TCP_CONNECT:

case RESPONSE_UDP_OPEN:

if (message_is_userdata(message)) {

uv_handle_base_t* handle = (uv_handle_base_t*)message_userdata(message);

lua_handle_base_t::close_uv_handle(handle);

}

break;

case CONTEXT_QUERY:

if (message.m_source != 0 && message.m_session != LUA_REFNIL) {

singleton_ref(node_lua_t).context_send(message.m_source, m_handle, message.m_session, CONTEXT_REPLY, NL_ENOREPLY);

}

return;

case LUA_CTX_WAIT:

singleton_ref(node_lua_t).context_send(message.m_source, (uint32_t)message_integer(message), LUA_REFNIL, LUA_CTX_WAKEUP, UV_OK);

return;

default:

break;

}

}

void context_lua_t::lua_ctx_init(message_t& message)

{

int32_t argc = (int32_t)message_integer(message);

int32_t envc = lua_gettop(m_lstate) - argc;

const char *file = lua_tostring(m_lstate, 1);

if (luaL_loadfile(m_lstate, file) == LUA_OK) {

lua_State *co;

lua_init(m_lstate);

lua_insert(m_lstate, -envc - 1);

lua_init_env(m_lstate, envc);

co = lua_new_root_coro(m_lstate);

lua_pushlightuserdata(m_lstate, NULL);

lua_pushcclosure(m_lstate, lua_ref_callback_entry, argc + 1); /* make the c closure callback */

lua_xmove(m_lstate, co, 1); /* move top c closure to co */

lua_pop(m_lstate, 1); /* pop the file name on top */

resume_coroutine(co, 0);

if (!is_active()) {

singleton_ref(node_lua_t).context_destroy(this, m_handle, "lua context exit normally");

}

return;

}

const char *error = lua_tostring(m_lstate, -1);

singleton_ref(node_lua_t).context_destroy(this, m_handle, "fail to initialize lua context %s, %s", file, error);

}

void context_lua_t::response_context_query(message_t& response)

{

if (response.m_source == 0) return;

if (!m_context_recv_sessions.wakeup_once(response.m_source, this, response) && !m_context_recv_sessions.wakeup_once(0, this, response)) {

if (response.m_session != LUA_REFNIL) {

singleton_ref(node_lua_t).context_send(response.m_source, m_handle, response.m_session, CONTEXT_REPLY, NL_ENOREPLY);

}

}

}

void context_lua_t::response_context_reply(message_t& response)

{

wakeup_ref_session(response.m_session, response, true);

}

void context_lua_t::response_context_wait(message_t& response)

{

m_context_wait_handles.insert(response.m_source);

}

void context_lua_t::response_context_wakeup(message_t& response)

{

m_context_wait_sessions.wakeup_all(response.m_source, this, response, true);

}

void context_lua_t::response_tcp_listen(message_t& response)

{

lua_tcp_listen_handle_t::wakeup_listen(m_lstate, response);

}

void context_lua_t::response_tcp_accept(message_t& response)

{

lua_tcp_listen_handle_t::wakeup_accept(m_lstate, response);

}

void context_lua_t::response_tcp_connect(message_t& response)

{

lua_tcp_socket_handle_t::wakeup_connect(m_lstate, response);

}

void context_lua_t::response_tcp_write(message_t& response)

{

wakeup_ref_session(response.m_session, response, true);

}

void context_lua_t::response_tcp_read(message_t& response)

{

lua_tcp_socket_handle_t::wakeup_read(m_lstate, response);

}

void context_lua_t::response_tcp_closing(message_t& response)

{

if (message_error(response) == NL_ETCPLCLOSED) { /* listen socket */

lua_tcp_listen_handle_t::wakeup_accept(m_lstate, response);

} else {

lua_tcp_socket_handle_t::wakeup_read(m_lstate, response);

}

}

void context_lua_t::response_udp_open(message_t& response)

{

lua_udp_handle_t::wakeup_open(m_lstate, response);

}

void context_lua_t::response_udp_write(message_t& response)

{

wakeup_ref_session(response.m_session, response, true);

}

void context_lua_t::response_udp_read(message_t& response)

{

lua_udp_handle_t::wakeup_read(m_lstate, response);

}

void context_lua_t::response_udp_closing(message_t& response)

{

lua_udp_handle_t::wakeup_read(m_lstate, response);

}

void context_lua_t::response_handle_close(message_t& response)

{

lua_handle_base_t::release_ref(m_lstate, response.m_session, (handle_set)(int32_t)message_integer(response));

}

void context_lua_t::response_timeout(message_t& response)

{

lua_timer_handle_t::wakeup(m_lstate, response);

}

////////////////////////////////////////////////context lua api/////////////////////////////////////////////////

int32_t context_lua_t::context_strerror(lua_State *L)

{

return lua_strerror(L, -1);

}

int32_t context_lua_t::context_create(lua_State *L)

{

int32_t argc = lua_gettop(L);

if (argc <= 0) {

luaL_error(L, "lua file is needed to initialize lua context");

}

if (argc > MAX_CTX_ARGC) {

luaL_error(L, "too many arguments to initialize lua context");

}

char* argv[MAX_CTX_ARGC];

char* env[16] = { NULL };

lua_load_env(L, env);

for (int32_t i = 0; i < argc; ++i) {

argv[i] = (char*)luaL_checkstring(L, i + 1);

}

uint32_t parent = lua_get_context_handle(L);

uint32_t handle = singleton_ref(node_lua_t).context_create<context_lua_t>(parent, argc, argv, env);

lua_free_env(L, env);

if (handle > 0) {

lua_pushinteger(L, handle);

return 1;

}

return 0;

}

int32_t context_lua_t::context_destroy(lua_State *L)

{

int32_t type = lua_type(L, 1);

uint32_t src_handle = lua_get_context_handle(L);

if (type == LUA_TNIL || type == LUA_TSTRING) { //kill self

singleton_ref(node_lua_t).context_destroy(src_handle, src_handle, lua_tostring(L, 1));

return 0;

}

int32_t handle = luaL_checkunsigned(L, 1);

if (handle > 0) {

singleton_ref(node_lua_t).context_destroy(handle, src_handle, lua_tostring(L, 2));

}

return 0;

}

int32_t context_lua_t::context_check_message(lua_State *L, int32_t idx, uint32_t msg_type, message_t& message)

{

buffer_t* buffer;

bson_t* bson_ptr;

int32_t* bson_data;

switch (lua_type(L, idx)) {

case LUA_TNUMBER:

if (lua_isinteger(L, idx)) {

message.m_type = MAKE_MESSAGE_TYPE(msg_type, INTEGER);

message.m_data.m_integer = (int64_t)lua_tointeger(L, idx);

} else {

message.m_type = MAKE_MESSAGE_TYPE(msg_type, NUMBER);

message.m_data.m_number = (double)lua_tonumber(L, idx);

}

message.m_source = lua_get_context_handle(L);

return 1;

case LUA_TSTRING:

message.m_type = MAKE_MESSAGE_TYPE(msg_type, STRING);

message.m_data.m_string = nl_strdup(lua_tostring(L, idx));

message.m_source = lua_get_context_handle(L);

return 1;

case LUA_TBOOLEAN:

message.m_type = MAKE_MESSAGE_TYPE(msg_type, TBOOLEAN);

message.m_data.m_bool = (bool)lua_toboolean(L, idx);

message.m_source = lua_get_context_handle(L);

return 1;

case LUA_TNIL:

message.m_type = MAKE_MESSAGE_TYPE(msg_type, NIL);

message.m_source = lua_get_context_handle(L);

return 1;

case LUA_TUSERDATA:

buffer = (buffer_t*)luaL_testudata(L, idx, BUFFER_METATABLE);

if (buffer) {

message.m_type = MAKE_MESSAGE_TYPE(msg_type, BUFFER);

message.m_data.m_buffer = buffer_grab(*buffer);

message.m_source = lua_get_context_handle(L);

return 1;

}

bson_data = (int32_t*)luaL_testudata(L, idx, BSON_METATABLE);

if (bson_data) {

message.m_type = MAKE_MESSAGE_TYPE(msg_type, BSON);

message.m_data.m_bson = bson_new(bson_data, false);

message.m_source = lua_get_context_handle(L);

return 1;

}

lua_pushstring(L, "transfer data type not supported");

return 0;

case LUA_TLIGHTUSERDATA:

message.m_type = MAKE_MESSAGE_TYPE(msg_type, USERDATA);

message.m_data.m_userdata = (void*)lua_touserdata(L, idx);

message.m_source = lua_get_context_handle(L);

return 1;

case LUA_TTABLE:

message.m_type = MAKE_MESSAGE_TYPE(msg_type, BSON);

message.m_data.m_bson = bson_new(NULL, true);

message.m_source = lua_get_context_handle(L);

return bson_encode(message.m_data.m_bson, L, idx) ? 1 : 0;

default:

lua_pushstring(L, "transfer data type not supported");

return 0;

}

}

int32_t context_lua_t::context_check_message(lua_State *L, int32_t idx, uint32_t count, uint32_t msg_type, message_t& message)

{

if (count <= 1) {

return context_check_message(L, idx, msg_type, message);

}

message_array_t* array = message_array_create(count);

message.m_type = MAKE_MESSAGE_TYPE(msg_type, ARRAY);

message.m_data.m_array = array;

message.m_source = lua_get_context_handle(L);

for (int32_t i = 0; i < count; ++i) {

if (context_check_message(L, idx + i, msg_type, array->m_array[i]) == 0) {

return 0;

}

}

return 1;

}

int32_t context_lua_t::context_send(lua_State *L, int32_t idx, uint32_t count, uint32_t handle, int32_t session, uint32_t msg_type)

{

message_t message;

int32_t status = context_check_message(L, idx, count, msg_type, message);

if (status == 0) {

message_clean(message);

return NL_ETRANSTYPE;

}

message.m_session = session;

if (singleton_ref(node_lua_t).context_send(handle, message)) {

return UV_OK;

}

message_clean(message);

return NL_ENOCONTEXT;

}

int32_t context_lua_t::context_send(lua_State *L)

{

uint32_t handle = luaL_checkunsigned(L, 1);

int32_t top = lua_gettop(L);