diff --git a/client-src/client.c b/client-src/client.c index 2df5a10d..1c5f1c87 100644 --- a/client-src/client.c +++ b/client-src/client.c @@ -14,7 +14,7 @@ struct args { int fd; }; -static void +static int readall(int fd, void * buffer, size_t sz) { for (;;) { int err = recv(fd , buffer, sz, MSG_WAITALL); @@ -23,7 +23,7 @@ readall(int fd, void * buffer, size_t sz) { continue; break; } - return; + return err; } perror("Socket error"); exit(1); @@ -37,7 +37,8 @@ _read(void *ud) { for (;;) { uint8_t header[2]; fflush(stdout); - readall(fd, header, 2); + if (readall(fd, header, 2) == 0) + break; size_t len = header[0] << 8 | header[1]; if (len>0) { char tmp[len+1]; @@ -49,8 +50,11 @@ _read(void *ud) { return NULL; } -static void -test(int fd) { +static void * +test(void *ud) { + struct args *p = ud; + int fd = p->fd; + char tmp[1024]; while (!feof(stdin)) { fgets(tmp,sizeof(tmp),stdin); @@ -68,6 +72,7 @@ test(int fd) { perror("send data"); } } + return NULL; } int @@ -94,8 +99,8 @@ main(int argc, char * argv[]) { struct args arg = { fd }; pthread_t pid ; pthread_create(&pid, NULL, _read, &arg); - - test(fd); + pthread_t pid_stdin; + pthread_create(&pid_stdin, NULL, test, &arg); pthread_join(pid, NULL); diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 7d1d65a0..3c4d16b6 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -103,8 +103,8 @@ function skynet.ret(...) coroutine.yield("RETURN", ...) end -function skynet.dispatch(f) - c.callback(function(session, address , msg, sz) +function skynet.default_dispatch(f) + return function(session, address , msg, sz) if session <= 0 then session = - session co = coroutine.create(f) @@ -117,9 +117,15 @@ function skynet.dispatch(f) session_id_coroutine[session] = nil suspend(co, coroutine.resume(co, msg, sz)) end - end) + end end +function skynet.dispatch(f) + c.callback(skynet.default_dispatch(f)) +end + +skynet.filter = assert(c.callback) + function skynet.start(f) local session = c.command("TIMEOUT","0") local co = coroutine.create( diff --git a/service-src/service_client.c b/service-src/service_client.c index 25af7807..6412183b 100644 --- a/service-src/service_client.c +++ b/service-src/service_client.c @@ -10,7 +10,6 @@ static void _cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { - assert(session == 0); assert(sz <= 65535); int fd = (int)(intptr_t)ud; diff --git a/service/agent.lua b/service/agent.lua index d60676f3..10145b5d 100644 --- a/service/agent.lua +++ b/service/agent.lua @@ -1,11 +1,24 @@ local skynet = require "skynet" local client = ... -skynet.dispatch(function(msg, sz , session, address) +local dispatch_handler = skynet.default_dispatch(function (msg,sz) local message = skynet.tostring(msg,sz) - print("command source",address) local result = skynet.call("SIMPLEDB",message) - skynet.send(address, result) + skynet.ret(result) +end) + +local session_id = 0 +skynet.filter(function (session, address , msg, sz) + if address == client then + assert(session == 0) + print("client message",skynet.tostring(msg,sz)) + -- It's client, there is no session + session_id = session_id + 1 + session = - session_id + else + print("skynet message",msg,sz) + end + dispatch_handler(session,address, msg,sz) end) skynet.start(function() diff --git a/service/watchdog.lua b/service/watchdog.lua index 7c89a366..99f67afb 100644 --- a/service/watchdog.lua +++ b/service/watchdog.lua @@ -11,6 +11,7 @@ function command:open(parm) fd = tonumber(fd) print("agent open",self,string.format("%d %d %s",self,fd,addr)) local client = skynet.launch("client",fd) + print("client",client) local agent = skynet.launch("snlua","agent",client) print("watchdog launch agent client:",agent,client) if agent then