diff --git a/Makefile b/Makefile index 38843622..4bc87d0d 100644 --- a/Makefile +++ b/Makefile @@ -12,7 +12,7 @@ snlua.so : service_lua.c gate.so : gate/mread.c gate/ringbuffer.c gate/main.c gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Igate -skynet.so : lua-skynet.c +skynet.so : lua-skynet.c lua-serialize/serialize.c gcc -Wall -g -fPIC --shared $^ -o $@ client.so : service_client.c diff --git a/agent.lua b/agent.lua index 9f09e6fd..49588038 100644 --- a/agent.lua +++ b/agent.lua @@ -1,7 +1,8 @@ local skynet = require "skynet" local client = ... -skynet.dispatch(function(msg,session) +skynet.dispatch(function(msg, sz , session, address) + local message = skynet.tostring(msg,sz) if session == 0 then print("client command",msg) local result = skynet.call("SIMPLEDB",msg) diff --git a/globallog.lua b/globallog.lua index b202f4ce..a01e5e51 100644 --- a/globallog.lua +++ b/globallog.lua @@ -1,6 +1,7 @@ local skynet = require "skynet" -skynet.dispatch(function(message, session , from) +skynet.dispatch(function(msg, sz , session , from) + local message = skynet.tostring(msg,sz) print("[GLOBALLOG]", from,message) end) diff --git a/launcher.lua b/launcher.lua index 2faf4720..d5d703f6 100644 --- a/launcher.lua +++ b/launcher.lua @@ -3,7 +3,8 @@ local string = string local instance = {} -skynet.dispatch(function(message, session, address) +skynet.dispatch(function(msg, sz , session, address) + local message = skynet.tostring(msg,sz) if session == 0 then -- init notice local reply = instance[address] diff --git a/lua-serialize/Makefile b/lua-serialize/Makefile new file mode 100644 index 00000000..e6ebf95c --- /dev/null +++ b/lua-serialize/Makefile @@ -0,0 +1,8 @@ +all : luaseri.so + +luaseri.so : serialize.c + gcc -Wall -fPIC -O2 -D LUAMODULE -o $@ --shared $^ + +clean : + rm luaseri.so + diff --git a/lua-serialize/README b/lua-serialize/README new file mode 100644 index 00000000..01122518 --- /dev/null +++ b/lua-serialize/README @@ -0,0 +1 @@ +see https://github.com/cloudwu/lua-serialize \ No newline at end of file diff --git a/lua-serialize/luaseri.h b/lua-serialize/luaseri.h new file mode 100644 index 00000000..23eb4cf9 --- /dev/null +++ b/lua-serialize/luaseri.h @@ -0,0 +1,9 @@ +#ifndef LUA_SERIALIZE_H +#define LUA_SERIALIZE_H + +#include + +int _luaseri_pack(lua_State *L); +int _luaseri_unpack(lua_State *L); + +#endif diff --git a/lua-serialize/serialize.c b/lua-serialize/serialize.c new file mode 100644 index 00000000..f573a983 --- /dev/null +++ b/lua-serialize/serialize.c @@ -0,0 +1,721 @@ +#include +#include +#include +#include +#include +#include + +#define TYPE_NIL 0 +#define TYPE_BOOLEAN 1 +// hibits 0 false 1 true +#define TYPE_NUMBER 2 +// hibits 0 : 0 , 1: byte, 2:word, 4: dword, 8 : double +#define TYPE_USERDATA 3 +#define TYPE_SHORT_STRING 4 +// hibits 0~31 : len +#define TYPE_LONG_STRING 5 +#define TYPE_TABLE 6 + +#define MAX_COOKIE 32 +#define COMBINE_TYPE(t,v) ((t) | (v) << 3) + +#define BLOCK_SIZE 128 +#define MAX_DEPTH 32 + +struct block { + struct block * next; + char buffer[BLOCK_SIZE]; +}; + +struct write_block { + struct block * head; + int len; + struct block * current; + int ptr; +}; + +struct read_block { + char * buffer; + struct block * current; + int len; + int ptr; +}; + +inline static struct block * +blk_alloc(void) { + struct block *b = malloc(sizeof(struct block)); + b->next = NULL; + return b; +} + +inline static void +wb_push(struct write_block *b, const void *buf, int sz) { + const char * buffer = buf; + if (b->ptr == BLOCK_SIZE) { +_again: + b->current = b->current->next = blk_alloc(); + b->ptr = 0; + } + if (b->ptr <= BLOCK_SIZE - sz) { + memcpy(b->current->buffer + b->ptr, buffer, sz); + b->ptr+=sz; + b->len+=sz; + } else { + int copy = BLOCK_SIZE - b->ptr; + memcpy(b->current->buffer + b->ptr, buffer, copy); + buffer += copy; + b->len += copy; + sz -= copy; + goto _again; + } +} + +static void +wb_init(struct write_block *wb , struct block *b) { + if (b==NULL) { + wb->head = blk_alloc(); + wb->len = 0; + wb->current = wb->head; + wb->ptr = 0; + wb_push(wb, &wb->len, sizeof(wb->len)); + } else { + wb->head = b; + int * plen = (int *)b->buffer; + int sz = *plen; + wb->len = sz; + while (b->next) { + sz -= BLOCK_SIZE; + b = b->next; + } + wb->current = b; + wb->ptr = sz; + } +} + +static struct block * +wb_close(struct write_block *b) { + b->current = b->head; + b->ptr = 0; + wb_push(b, &b->len, sizeof(b->len)); + b->current = NULL; + return b->head; +} + +static void +wb_free(struct write_block *wb) { + struct block *blk = wb->head; + while (blk) { + struct block * next = blk->next; + free(blk); + blk = next; + } + wb->head = NULL; + wb->current = NULL; + wb->ptr = 0; + wb->len = 0; +} + +static void +rball_init(struct read_block * rb, char * buffer, int size) { + rb->buffer = buffer; + rb->current = NULL; + rb->len = size; + rb->ptr = 0; +} + +#ifdef LUAMODULE + +static int +rb_init(struct read_block *rb, struct block *b) { + rb->buffer = NULL; + rb->current = b; + memcpy(&(rb->len),b->buffer,sizeof(rb->len)); + rb->ptr = sizeof(rb->len); + rb->len -= rb->ptr; + return rb->len; +} + +#endif + +static void * +rb_read(struct read_block *rb, void *buffer, int sz) { + if (rb->len < sz) { + return NULL; + } + + if (rb->buffer) { + int ptr = rb->ptr; + rb->ptr += sz; + rb->len -= sz; + return rb->buffer + ptr; + } + + if (rb->ptr == BLOCK_SIZE) { + struct block * next = rb->current->next; + free(rb->current); + rb->current = next; + rb->ptr = 0; + } + + int copy = BLOCK_SIZE - rb->ptr; + + if (sz <= copy) { + void * ret = rb->current->buffer + rb->ptr; + rb->ptr += sz; + rb->len -= sz; + return ret; + } + + char * tmp = buffer; + + memcpy(tmp, rb->current->buffer + rb->ptr, copy); + sz -= copy; + tmp += copy; + rb->len -= copy; + + for (;;) { + struct block * next = rb->current->next; + free(rb->current); + rb->current = next; + + if (sz < BLOCK_SIZE) { + memcpy(tmp, rb->current->buffer, sz); + rb->ptr = sz; + rb->len -= sz; + return buffer; + } + memcpy(tmp, rb->current->buffer, BLOCK_SIZE); + sz -= BLOCK_SIZE; + tmp += BLOCK_SIZE; + rb->len -= BLOCK_SIZE; + } +} + +static void +rb_close(struct read_block *rb) { + while (rb->current) { + struct block * next = rb->current->next; + free(rb->current); + rb->current = next; + } + rb->len = 0; + rb->ptr = 0; +} + +static inline void +wb_nil(struct write_block *wb) { + int n = TYPE_NIL; + wb_push(wb, &n, 1); +} + +static inline void +wb_boolean(struct write_block *wb, int boolean) { + int n = COMBINE_TYPE(TYPE_BOOLEAN , boolean ? 1 : 0); + wb_push(wb, &n, 1); +} + +static inline void +wb_integer(struct write_block *wb, int v) { + if (v == 0) { + int n = COMBINE_TYPE(TYPE_NUMBER , 0); + wb_push(wb, &n, 1); + } else if (v<0) { + int n = COMBINE_TYPE(TYPE_NUMBER , 4); + wb_push(wb, &n, 1); + wb_push(wb, &v, 4); + } else if (v<0x100) { + int n = COMBINE_TYPE(TYPE_NUMBER , 1); + wb_push(wb, &n, 1); + uint8_t byte = (uint8_t)v; + wb_push(wb, &byte, 1); + } else if (v<0x10000) { + int n = COMBINE_TYPE(TYPE_NUMBER , 2); + wb_push(wb, &n, 1); + uint16_t word = (uint16_t)v; + wb_push(wb, &word, 2); + } else { + int n = COMBINE_TYPE(TYPE_NUMBER , 4); + wb_push(wb, &n, 1); + wb_push(wb, &v, 4); + } +} + +static inline void +wb_number(struct write_block *wb, double v) { + int n = COMBINE_TYPE(TYPE_NUMBER , 8); + wb_push(wb, &n, 1); + wb_push(wb, &v, 8); +} + +static inline void +wb_pointer(struct write_block *wb, void *v) { + int n = TYPE_USERDATA; + wb_push(wb, &n, 1); + wb_push(wb, &v, sizeof(v)); +} + +static inline void +wb_string(struct write_block *wb, const char *str, int len) { + if (len < MAX_COOKIE) { + int n = COMBINE_TYPE(TYPE_SHORT_STRING, len); + wb_push(wb, &n, 1); + if (len > 0) { + wb_push(wb, str, len); + } + } else { + assert(len < 0x10000); + int n = TYPE_LONG_STRING; + wb_push(wb, &n, 1); + uint16_t x = (uint16_t) len; + wb_push(wb, &x, 2); + wb_push(wb, str, len); + } +} + +static void _pack_one(lua_State *L, struct write_block *b, int index, int depth); + +static int +wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) { + int array_size = lua_rawlen(L,index); + if (array_size > MAX_COOKIE-1) { + int n = COMBINE_TYPE(TYPE_TABLE, MAX_COOKIE-1); + wb_push(wb, &n, 1); + wb_integer(wb, array_size); + } else { + int n = COMBINE_TYPE(TYPE_TABLE, array_size); + wb_push(wb, &n, 1); + } + + int i; + for (i=1;i<=array_size;i++) { + lua_rawgeti(L,index,i); + _pack_one(L, wb, -1, depth); + lua_pop(L,1); + } + + return array_size; +} + +static void +wb_table_hash(lua_State *L, struct write_block * wb, int index, int depth, int array_size) { + lua_pushnil(L); + while (lua_next(L, index) != 0) { + if (lua_type(L,-2) == LUA_TNUMBER) { + lua_Number k = lua_tonumber(L,-2); + lua_Integer x = lua_tointeger(L,-2); + if (k == (lua_Number)x && x>0 && x<=array_size) { + lua_pop(L,1); + continue; + } + } + _pack_one(L,wb,-2,depth); + _pack_one(L,wb,-1,depth); + lua_pop(L, 1); + } + wb_nil(wb); +} + +static void +wb_table(lua_State *L, struct write_block *wb, int index, int depth) { + if (index < 0) { + index = lua_gettop(L) + index + 1; + } + int array_size = wb_table_array(L, wb, index, depth); + wb_table_hash(L, wb, index, depth, array_size); +} + +static void +_pack_one(lua_State *L, struct write_block *b, int index, int depth) { + if (depth > MAX_DEPTH) { + wb_free(b); + luaL_error(L, "serialize can't pack too depth table"); + } + int type = lua_type(L,index); + switch(type) { + case LUA_TNIL: + wb_nil(b); + break; + case LUA_TNUMBER: { + lua_Integer x = lua_tointeger(L,index); + lua_Number n = lua_tonumber(L,index); + if ((lua_Number)x==n) { + wb_integer(b, x); + } else { + wb_number(b,n); + } + break; + } + case LUA_TBOOLEAN: + wb_boolean(b, lua_toboolean(L,index)); + break; + case LUA_TSTRING: { + size_t sz = 0; + const char *str = lua_tolstring(L,index,&sz); + if (sz >= 0x10000) { + wb_free(b); + luaL_error(L,"string is too long to serialize"); + } + wb_string(b, str, sz); + break; + } + case LUA_TLIGHTUSERDATA: + wb_pointer(b, lua_touserdata(L,index)); + break; + case LUA_TTABLE: + wb_table(L, b, index, depth+1); + break; + default: + wb_free(b); + luaL_error(L, "Unsupport type %s to serialize", lua_typename(L, type)); + } +} + +static void +_pack_from(lua_State *L, struct write_block *b, int from) { + int n = lua_gettop(L) - from; + int i; + for (i=1;i<=n;i++) { + _pack_one(L, b , from + i, 0); + } +} + +#ifdef LUAMODULE + +static int +_pack(lua_State *L) { + struct write_block b; + wb_init(&b, NULL); + _pack_from(L,&b,0); + struct block * ret = wb_close(&b); + lua_pushlightuserdata(L,ret); + return 1; +} + +static int +_append(lua_State *L) { + struct write_block b; + wb_init(&b, lua_touserdata(L,1)); + _pack_from(L,&b,1); + struct block * ret = wb_close(&b); + lua_pushlightuserdata(L,ret); + return 1; +} + +#endif + +static inline void +__invalid_stream(lua_State *L, struct read_block *rb, int line) { + int len = rb->len; + if (rb->buffer == NULL) { + rb_close(rb); + } + luaL_error(L, "Invalid serialize stream %d (line:%d)", len, line); +} + +#define _invalid_stream(L,rb) __invalid_stream(L,rb,__LINE__) + +static double +_get_number(lua_State *L, struct read_block *rb, int cookie) { + switch (cookie) { + case 0: + return 0; + case 1: { + uint8_t n = 0; + uint8_t * pn = rb_read(rb,&n,1); + if (pn == NULL) + _invalid_stream(L,rb); + return *pn; + } + case 2: { + uint16_t n = 0; + uint16_t * pn = rb_read(rb,&n,2); + if (pn == NULL) + _invalid_stream(L,rb); + return *pn; + } + case 4: { + int n = 0; + int * pn = rb_read(rb,&n,4); + if (pn == NULL) + _invalid_stream(L,rb); + return *pn; + } + case 8: { + double n = 0; + double * pn = rb_read(rb,&n,8); + if (pn == NULL) + _invalid_stream(L,rb); + return *pn; + } + default: + _invalid_stream(L,rb); + return 0; + } +} + +static void * +_get_pointer(lua_State *L, struct read_block *rb) { + void * userdata = 0; + void ** v = (void **)rb_read(rb,&userdata,sizeof(userdata)); + if (v == NULL) { + _invalid_stream(L,rb); + } + return *v; +} + +static void +_get_buffer(lua_State *L, struct read_block *rb, int len) { + char tmp[len]; + char * p = rb_read(rb,tmp,len); + lua_pushlstring(L,p,len); +} + +static void _unpack_one(lua_State *L, struct read_block *rb); + +static void +_unpack_table(lua_State *L, struct read_block *rb, int array_size) { + if (array_size == MAX_COOKIE-1) { + uint8_t type = 0; + uint8_t *t = rb_read(rb, &type, 1); + if (t==NULL || (*t & 7) != TYPE_NUMBER) { + _invalid_stream(L,rb); + } + array_size = (int)_get_number(L,rb,*t >> 3); + } + lua_createtable(L,array_size,0); + int i; + for (i=1;i<=array_size;i++) { + _unpack_one(L,rb); + lua_rawseti(L,-2,i); + } + for (;;) { + _unpack_one(L,rb); + if (lua_isnil(L,-1)) { + lua_pop(L,1); + return; + } + _unpack_one(L,rb); + lua_rawset(L,-3); + } +} + +static void +_push_value(lua_State *L, struct read_block *rb, int type, int cookie) { + switch(type) { + case TYPE_NIL: + lua_pushnil(L); + break; + case TYPE_BOOLEAN: + lua_pushboolean(L,cookie); + break; + case TYPE_NUMBER: + lua_pushnumber(L,_get_number(L,rb,cookie)); + break; + case TYPE_USERDATA: + lua_pushlightuserdata(L,_get_pointer(L,rb)); + break; + case TYPE_SHORT_STRING: + _get_buffer(L,rb,cookie); + break; + case TYPE_LONG_STRING: { + uint16_t len = 0; + uint16_t *plen = rb_read(rb, &len, 2); + if (plen == NULL) { + _invalid_stream(L,rb); + } + _get_buffer(L,rb,*plen); + break; + } + case TYPE_TABLE: { + _unpack_table(L,rb,cookie); + break; + } + } +} + +static void +_unpack_one(lua_State *L, struct read_block *rb) { + uint8_t type = 0; + uint8_t *t = rb_read(rb, &type, 1); + if (t==NULL) { + _invalid_stream(L, rb); + } + _push_value(L, rb, *t & 0x7, *t>>3); +} + +#ifdef LUAMODULE + +static int +_unpack(lua_State *L) { + struct block * blk = lua_touserdata(L,1); + if (blk == NULL) { + return luaL_error(L, "Need a block to unpack"); + } + lua_settop(L,0); + struct read_block rb; + rb_init(&rb, blk); + + int i; + for (i=0;;i++) { + if (i%16==15) { + lua_checkstack(L,i); + } + uint8_t type = 0; + uint8_t *t = rb_read(&rb, &type, 1); + if (t==NULL) + break; + _push_value(L, &rb, *t & 0x7, *t>>3); + } + + rb_close(&rb); + + return lua_gettop(L); +} + +static int +_dump_mem(const char * buffer, int len, int size) { + int i; + for (i=0;ibuffer ,sizeof(len)); + len -= sizeof(len); + printf("Len = %d\n",len); + len = _dump_mem(b->buffer + sizeof(len), BLOCK_SIZE - sizeof(len) , len); + while (len > 0) { + b=b->next; + len = _dump_mem(b->buffer, BLOCK_SIZE , len); + } + printf("\n"); + return 0; +} + +#endif + +static void +_seri(lua_State *L, struct block *b) { + uint32_t len = 0; + memcpy(&len, b->buffer ,sizeof(len)); + + len -= 4; + uint8_t * buffer = malloc(len); + uint8_t * ptr = buffer; + int sz = len; + if (len < BLOCK_SIZE - 4) { + memcpy(ptr, b->buffer+4, len); + } else { + memcpy(ptr, b->buffer+4, BLOCK_SIZE-4); + ptr += BLOCK_SIZE-4; + len -= BLOCK_SIZE-4; + b = b->next; + + while(len>0) { + if (len >= BLOCK_SIZE) { + memcpy(ptr, b->buffer, BLOCK_SIZE); + ptr += BLOCK_SIZE; + len -= BLOCK_SIZE; + } else { + memcpy(ptr, b->buffer, len); + break; + } + b = b->next; + } + } + + lua_pushlightuserdata(L, buffer); + lua_pushinteger(L, sz); +} + +#ifdef LUAMODULE + +static int +_serialize(lua_State *L) { + struct block *b = lua_touserdata(L,1); + if (b==NULL) { + return luaL_error(L, "dump null pointer"); + } + + _seri(L,b); + + return 2; +} + +#endif + +int +_luaseri_unpack(lua_State *L) { + if (lua_isnoneornil(L,1)) { + return 0; + } + void * buffer = lua_touserdata(L,1); + if (buffer == NULL) { + return luaL_error(L, "deserialize null pointer"); + } + int len = luaL_checkinteger(L,2); + + lua_settop(L,0); + struct read_block rb; + rball_init(&rb, buffer, len); + + int i; + for (i=0;;i++) { + if (i%16==15) { + lua_checkstack(L,i); + } + uint8_t type = 0; + uint8_t *t = rb_read(&rb, &type, 1); + if (t==NULL) + break; + _push_value(L, &rb, *t & 0x7, *t>>3); + } + + // Need not free buffer + + return lua_gettop(L); +} + +int +_luaseri_pack(lua_State *L) { + struct write_block wb; + wb_init(&wb, NULL); + _pack_from(L,&wb,0); + struct block * b = wb_close(&wb); + _seri(L,b); + + while (b) { + struct block * next = b->next; + free(b); + b = next; + } + + return 2; +} + +#ifdef LUAMODULE + +int +luaopen_luaseri(lua_State *L) { + luaL_Reg l[] = { + { "pack", _pack }, + { "unpack", _unpack }, + { "append", _append }, + { "serialize", _serialize }, + { "deserialize", _luaseri_unpack }, + { "dump", _dump }, + { NULL, NULL }, + }; + luaL_newlib(L,l); + return 1; +} + +#endif \ No newline at end of file diff --git a/lua-serialize/test.lua b/lua-serialize/test.lua new file mode 100644 index 00000000..6d787cdd --- /dev/null +++ b/lua-serialize/test.lua @@ -0,0 +1,28 @@ +s = require "luaseri" + +a = s.pack { hello={3,4}, false, 1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9 } + +s.dump(a) + +a = s.append(a, 42,4.2,-1,1000,80000,"hello",true,false,nil,"1234567890123456789012345678901234567890") + +s.dump(a) +print(a) + +function pr(t,...) + for k,v in pairs(t) do + print(k,v) + end + print(...) +end + +print ("------") + +local seri, length = s.serialize(a) +print(seri, length) + +pr(s.unpack(a)) + +print("-------") + +pr(s.deserialize(seri, length)) diff --git a/lua-serialize/test2.lua b/lua-serialize/test2.lua new file mode 100644 index 00000000..e2177179 --- /dev/null +++ b/lua-serialize/test2.lua @@ -0,0 +1,15 @@ +local s = require "luaseri" + +addressbook = { + name = "Alice", + id = 12345, + phone = { + { number = "1301234567" }, + { number = "87654321", type = "WORK" }, + } +} + +for i=1,100000 do + local u = s.pack (addressbook) + local t = s.unpack(u) +end diff --git a/lua-skynet.c b/lua-skynet.c index 2a34f045..e1cc7f37 100644 --- a/lua-skynet.c +++ b/lua-skynet.c @@ -1,4 +1,5 @@ #include "skynet.h" +#include "lua-serialize/luaseri.h" #include #include @@ -35,8 +36,9 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, } else { lua_pushinteger(L, session); lua_pushstring(L, addr); - lua_pushlstring(L, msg, sz); - r = lua_pcall(L, 3, 0 , trace); + lua_pushlightuserdata(L,(void *)msg); + lua_pushinteger(L,sz); + r = lua_pcall(L, 4, 0 , trace); } if (r == LUA_OK) return; @@ -138,6 +140,17 @@ _error(lua_State *L) { return 0; } +static int +_tostring(lua_State *L) { + if (lua_isnoneornil(L,1)) { + return 0; + } + char * msg = lua_touserdata(L,1); + int sz = luaL_checkinteger(L,2); + lua_pushlstring(L,msg,sz); + return 1; +} + int luaopen_skynet_c(lua_State *L) { luaL_checkversion(L); @@ -146,6 +159,9 @@ luaopen_skynet_c(lua_State *L) { { "command" , _command }, { "callback" , _callback }, { "error", _error }, + { "tostring", _tostring }, + { "pack", _luaseri_pack }, + { "unpack", _luaseri_unpack }, { NULL, NULL }, }; lua_pushcfunction(L, traceback); diff --git a/redis-cli.lua b/redis-cli.lua index f8b3755c..1346e98d 100644 --- a/redis-cli.lua +++ b/redis-cli.lua @@ -2,6 +2,8 @@ local skynet = require "skynet" local string = string local table = table local tonumber = tonumber +local ipairs = ipairs +local unpack = unpack local redis_server = ... local fd local write_fd @@ -15,36 +17,67 @@ local function init_fd(fdstr) read_fd = "READ " .. fd .. " " end -skynet.dispatch(function(message, ...) - skynet.send(".connection", write_fd .. message) +local function compose_message(msg) + local lines = { "*" .. #msg } + for _,v in ipairs(msg) do + table.insert(lines,"$"..#v) + table.insert(lines,v) + end + table.insert(lines,"") + + local cmd = table.concat(lines,"\r\n") + return cmd +end + +local redcmd = {} + +redcmd[42] = function(data) -- '*' + local n = tonumber(data) + if n < 1 then + skynet.ret(skynet.pack(true, nil)) + return + end + local bulk = {} + for i = 1,n do + local line = skynet.call(".connection", readline_fd) + local bytes = tonumber(string.sub(line,2) + 2) + local data = skynet.call(".connection", read_fd .. bytes) + table.insert(result, string.sub(data,1,-3)) + end + skynet.ret(skynet.pack(unpack(bulk))) +end + +redcmd[36] = function(data) -- '$' + local bytes = tonumber(data) + if bytes < 0 then + skynet.ret(skynet.pack(true, nil)) + return + end + local firstline = skynet.call(".connection", read_fd .. (bytes + 2)) + skynet.ret(skynet.pack(string.sub(firstline,1,-3))) +end + +redcmd[43] = function(data) -- '+' + skynet.ret(skynet.pack(true, data)) +end + +redcmd[45] = function(data) -- '-' + skynet.ret(skynet.pack(false, data)) +end + +redcmd[58] = function(data) -- ':' + skynet.ret(skynet.pack(true, tonumber(data))) +end + +skynet.dispatch(function(msg, sz, session, address) + local message = { skynet.unpack(msg,sz) } + skynet.send(".connection", write_fd .. compose_message(message)) local result = skynet.call(".connection", readline_fd) local firstchar = string.byte(result) - if firstchar == 42 then -- '*' - local n = tonumber(string.sub(result,2)) - if n < 1 then - skynet.ret(result .. "\r\n") - return - end - local bulk = { result } - for i = 1,n do - local line = skynet.call(".connection", readline_fd) - table.insert(result, line) - local bytes = tonumber(string.sub(line,2) + 2) - table.insert(result, bytes) - end - table.insert(result,"") - skynet.ret(table.concat(result,"\r\n")) - elseif firstchar == 36 then -- '$' - local bytes = tonumber(string.sub(result,2)) - if bytes < 0 then - skynet.ret(result .. "\r\n") - return - end - local firstline = skynet.call(".connection", read_fd .. (bytes + 2)) - skynet.ret(result .. "\r\n" .. firstline) - else - skynet.ret(result .. "\r\n") - end + local data = string.sub(result,2) + local f = redcmd[firstchar] + assert(f) + f(data) end) skynet.start(function() diff --git a/simpledb.lua b/simpledb.lua index c82fe639..28229285 100644 --- a/simpledb.lua +++ b/simpledb.lua @@ -13,7 +13,8 @@ function command.SET(key, value) skynet.ret(last) end -skynet.dispatch(function(message, session, from) +skynet.dispatch(function(msg, sz , session, from) + local message = skynet.tostring(msg,sz) print("simpledb",message, from, session) local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") local f = command[cmd] diff --git a/skynet.lua b/skynet.lua index d8f79f1d..619a2eca 100644 --- a/skynet.lua +++ b/skynet.lua @@ -9,14 +9,14 @@ local session_id_coroutine = {} local session_coroutine_id = {} local session_coroutine_address = {} -local function suspend(co, result, command, param) +local function suspend(co, result, command, param, size) assert(result, command) if command == "CALL" or command == "SLEEP" then session_id_coroutine[param] = co elseif command == "RETURN" then local co_session = session_coroutine_id[co] local co_address = session_coroutine_address[co] - c.send(co_address, co_session, param) + c.send(co_address, co_session, param, size) else assert(command == nil, command) session_coroutine_id[co] = nil @@ -70,29 +70,39 @@ function skynet.kill(name) end skynet.send = c.send +skynet.pack = c.pack +skynet.tostring = c.tostring +skynet.unpack = c.unpack -function skynet.call(addr, message) - local session = c.send(addr, -1, message) - return coroutine.yield("CALL", session) +function skynet.call(addr, deseri , ...) + local t = type(deseri) + if t == "function" then + local session = c.send(addr, -1, ...) + return deseri(coroutine.yield("CALL", session)) + else + assert(t=="string") + local session = c.send(addr, -1, deseri) + return c.tostring(coroutine.yield("CALL", session)) + end end -function skynet.ret(message) - coroutine.yield("RETURN", message) +function skynet.ret(...) + coroutine.yield("RETURN", ...) end function skynet.dispatch(f) - c.callback(function(session, address , message) + c.callback(function(session, address , msg, sz) if session <= 0 then session = - session co = coroutine.create(f) session_coroutine_id[co] = session session_coroutine_address[co] = address - suspend(co, coroutine.resume(co, message, session, address)) + suspend(co, coroutine.resume(co, msg, sz, session, address)) else local co = session_id_coroutine[session] assert(co, session) session_id_coroutine[session] = nil - suspend(co, coroutine.resume(co, message)) + suspend(co, coroutine.resume(co, msg, sz)) end end) end diff --git a/testredis.lua b/testredis.lua index ab098afd..e5f3da89 100644 --- a/testredis.lua +++ b/testredis.lua @@ -5,11 +5,10 @@ skynet.dispatch() local command ="*2\r\n$3\r\nGET\r\n$1\r\nA\r\n" skynet.start(function() - local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:6379") + local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:7379") print("redis-cli:", cli) assert(cli) - local result = skynet.call(cli, command) - print(result) + print(skynet.call(cli, skynet.unpack, skynet.pack("GET","A"))) skynet.exit() end) diff --git a/watchdog.lua b/watchdog.lua index 12f836dd..00f32c89 100644 --- a/watchdog.lua +++ b/watchdog.lua @@ -31,7 +31,8 @@ function command:data(data) end end -skynet.dispatch(function(message) +skynet.dispatch(function(msg, sz) + local message = skynet.tostring(msg,sz) local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") id = tonumber(id) local f = command[cmd]