service client and a simple DB example

Makefile

@@ -1,4 +1,4 @@
-all : skynet snlua.so logger.so skynet.so gate.so client skynet-master
+all : skynet snlua.so logger.so skynet.so gate.so client.so client skynet-master
 
 skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c skynet_harbor.c
 	gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq
@@ -15,8 +15,11 @@ gate.so : gate/mread.c gate/ringbuffer.c gate/map.c gate/main.c
 skynet.so : lua-skynet.c
 	gcc -Wall -g -fPIC --shared $^ -o $@
 
+client.so : service_client.c
+	gcc -Wall -g -fPIC --shared $^ -o $@
+
 client : client.c
-	gcc -Wall -g $^ -o $@
+	gcc -Wall -g $^ -o $@ -lpthread
 
 skynet-master : master/master.c
 	gcc -g -Wall -o $@ $^ -lzmq

agent.lua

@@ -1,7 +1,16 @@
 local skynet = require "skynet"
+local client = ...
 
-print("agent",...)
-
-skynet.dispatch(function(msg , addr)
-	print("[agent]",addr,msg)
+skynet.dispatch(function(msg,session)
+	if session == 0 then
+		print("client command",msg)
+		local result = skynet.call("SIMPLEDB",msg)
+		skynet.send(client,0,result)
+	else
+		print("server command",msg)
+		if msg == "CLOSE" then
+			skynet.kill(client)
+			skynet.exit()
+		end
+	end
 end)

client.c

@@ -1,3 +1,4 @@
+#include <pthread.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <arpa/inet.h>
@@ -7,6 +8,46 @@
 #include <stdio.h>
 #include <stdint.h>
 #include <string.h>
+#include <errno.h>
+
+struct args {
+	int fd;
+};
+
+static void
+readall(int fd, void * buffer, size_t sz) {
+	for (;;) {
+		int err = recv(fd , buffer, sz, MSG_WAITALL);
+		if (err < 0) {
+			if (errno == EAGAIN || errno == EINTR)
+				continue;
+			break;
+		}
+		return;
+	}
+	perror("Socket error");
+	exit(1);
+}
+
+static void *
+_read(void *ud) {
+	struct args *p = ud;
+	int fd = p->fd;
+	fflush(stdout);
+	for (;;) {
+		uint8_t header[2];
+		fflush(stdout);
+		readall(fd, header, 2);
+		size_t len = header[0] | header[1] << 8;
+		if (len>0) {
+			char tmp[len+1];
+			readall(fd, tmp, len);
+			tmp[len]='\0';
+			printf("%s\n",tmp);
+		}
+	}
+	return NULL;
+}
 
 static void
 test(int fd) {
@@ -50,8 +91,14 @@ main(int argc, char * argv[]) {
 		return 1;
 	}
 
+	struct args arg = { fd };
+	pthread_t pid ;
+	pthread_create(&pid, NULL, _read, &arg);
+
 	test(fd);
 
+	pthread_join(pid, NULL); 
+
 	close(fd);
 
 	return 0;

gate/main.c

@@ -118,10 +118,16 @@ _report(struct skynet_context * ctx, const char * data, ...) {
 static void
 _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) {
 	struct connection * agent = _id_to_agent(g,uid);
-	char * tmp = malloc(len + 32);
-	int n = snprintf(tmp,len+32,"%d data ",uid);
-	memcpy(tmp+n,data,len);
-	skynet_send(ctx, agent->agent ? agent->agent : WATCHDOG, 0, tmp, len + n);
+	if (agent->agent) {
+		char * tmp = malloc(len);
+		memcpy(tmp,data,len);
+		skynet_send(ctx, agent->agent, 0, tmp, len);
+	} else {
+		char * tmp = malloc(len + 32);
+		int n = snprintf(tmp,len+32,"%d data ",uid);
+		memcpy(tmp+n,data,len);
+		skynet_send(ctx, WATCHDOG, 0, tmp, len + n);
+	}
 }
 
 static int

globallog.lua

@@ -1,6 +1,6 @@
 local skynet = require "skynet"
 
-skynet.dispatch(function(message, from, session)
+skynet.dispatch(function(message, session , from)
 	print("[GLOBALLOG]",session, from,message)
 end)
 

main.lua

@@ -8,5 +8,7 @@ local watchdog = skynet.launch("snlua","watchdog.lua")
 print("watchdog",watchdog)
 local gate = skynet.launch("gate","8888 4 0")
 print("gate",gate)
+local db = skynet.launch("snlua","simpledb.lua")
+print("simpledb",db)
 
 skynet.exit()

main_log.lua

@@ -5,7 +5,4 @@ print("Log server start")
 local log = skynet.launch("snlua","globallog.lua")
 print("log",log)
 
-local db = skynet.launch("snlua","simpledb.lua")
-print("simpledb",db)
-
 skynet.exit()

service_client.c

@@ -0,0 +1,32 @@
+#include "skynet.h"
+
+#include <sys/uio.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <assert.h>
+#include <stdio.h>
+#include <string.h>
+
+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;
+
+	struct iovec buffer[2];
+	uint8_t head[2] = { sz & 0xff , sz >> 8 & 0xff };
+	buffer[0].iov_base = head;
+	buffer[0].iov_len = 2;
+	buffer[1].iov_base = (void *)msg;
+	buffer[1].iov_len = sz;
+
+	writev(fd, buffer, 2);
+}
+
+int
+client_init(void * dummy, struct skynet_context *ctx, const char * args) {
+	int fd = strtol(args, NULL, 10);
+	skynet_callback(ctx, (void*)(intptr_t)fd, _cb);
+
+	return 0;
+}

simpledb.lua

@@ -16,7 +16,12 @@ end
 skynet.dispatch(function(message, from, session)
 	print("simpledb",message, from, session)
 	local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)")
-	command[cmd](key,value)
+	local f = command[cmd]
+	if f then
+		f(key,value)
+	else
+		skynet.ret("Invalid command : "..message)
+	end
 end)
 
 skynet.register "SIMPLEDB"

skynet.lua

@@ -65,6 +65,10 @@ function skynet.exit()
 	c.command("EXIT")
 end
 
+function skynet.kill(name)
+	c.command("KILL",name)
+end
+
 skynet.send = c.send
 
 function skynet.call(addr, message)
@@ -83,7 +87,7 @@ function skynet.dispatch(f)
 			co = coroutine.create(f)
 			session_coroutine_id[co] = session
 			session_coroutine_address[co] = address
-			suspend(co, coroutine.resume(co, message, address, session))
+			suspend(co, coroutine.resume(co, message, session, address))
 		else
 			session_id_coroutine[session] = nil
 			suspend(co, coroutine.resume(co, message))

skynet_server.c

@@ -128,17 +128,20 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	}
 }
 
-static void
+static int
 _drop_queue(struct message_queue *q) {
 	// todo: send message back to message source
 	struct skynet_message msg;
+	int s = 0;
 	while(!skynet_mq_pop(q, &msg)) {
+		++s;
 		if (skynet_harbor_message_isremote(msg.source)) {
 			skynet_harbor_message_close(&msg);
 		}
 		free(msg.data);
 	}
 	skynet_mq_release(q);
+	return s;
 }
 
 int
@@ -151,8 +154,10 @@ skynet_context_message_dispatch(void) {
 
 	struct skynet_context * ctx = skynet_handle_grab(handle);
 	if (ctx == NULL) {
-		skynet_error(NULL, "Drop message queue %x ", handle);
-		_drop_queue(q);
+		int s = _drop_queue(q);
+		if (s>0) {
+			skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s);
+		}
 		return 0;
 	}
 
@@ -226,6 +231,21 @@ skynet_command(struct skynet_context * context, const char * cmd , int session,
 		return NULL;
 	}
 
+	if (strcmp(cmd,"KILL") == 0) {
+		uint32_t handle = 0;
+		if (parm[0] == ':') {
+			handle = strtoul(parm+1, NULL, 16);
+		} else if (parm[0] == '.') {
+			handle = skynet_handle_findname(parm+1);
+		} else {
+			// todo : kill global service
+		}
+		if (handle) {
+			skynet_handle_retire(handle);
+		}
+		return NULL;
+	}
+
 	if (strcmp(cmd,"LAUNCH") == 0) {
 		size_t sz = strlen(parm);
 		char tmp[sz+1];

watchdog.lua

@@ -1,23 +1,34 @@
 local skynet = require "skynet"
 
 local command = {}
+local agent_all = {}
 
 function command:open(parm)
 	local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
 	fd = tonumber(fd)
 	skynet.send("LOG", 0, string.format("%d %d %s",self,fd,addr))
-	local agent = skynet.launch("snlua","agent.lua",self)
+	local client = skynet.launch("client",fd)
+	print("client",client)
+	local agent = skynet.launch("snlua","agent.lua",client)
 	if agent then
+		agent_all[self] = agent
 		skynet.send("gate",0, "forward ".. self .. " " .. agent)
 	end
 end
 
 function command:close()
 	skynet.send("LOG",0,string.format("close %d",self))
+	skynet.send(agent_all[self],1,"CLOSE")
+	agent_all[self] = nil
 end
 
 function command:data(data)
-	skynet.send("LOG",0,string.format("data %d size=%d",self,#data))
+	local agent = agent_all[self]
+	if agent then
+		skynet.send(agent,0,data)
+	else
+		skynet.send("LOG",0,string.format("data %d size=%d",self,#data))
+	end
 end
 
 skynet.dispatch(function(message)