client message filter

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
+static void *
-test(int fd) {
+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);
-
+	pthread_t pid_stdin;
-	test(fd);
+	pthread_create(&pid_stdin, NULL, test, &arg);
 
 	pthread_join(pid, NULL); 
 

lualib/skynet.lua

@@ -103,8 +103,8 @@ function skynet.ret(...)
 	coroutine.yield("RETURN", ...)
 end
 
-function skynet.dispatch(f)
+function skynet.default_dispatch(f)
-	c.callback(function(session, address , msg, sz)
+	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(

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;
 

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()

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