remove cluster.rpc and cluster.msg

add cluster.lua for more flexible rpc

examples/rpc.lua

@@ -1,6 +1,6 @@
 local core = require "core"
 local crypto = require "core.crypto"
-local rpc = require "cluster.rpc"
+local cluster = require "core.cluster"
 local zproto = require "zproto"
 
 local proto = zproto:parse [[
@@ -12,39 +12,62 @@ pong 0x2 {
 }
 ]]
 
+assert(proto)
+local function unmarshal(cmd, buf, size)
+	local dat, size = proto:unpack(buf, size, true)
+	local body = proto:decode(cmd, dat, size)
+	return body
+end
 
-local server = rpc.listen {
-	addr = "127.0.0.1:9999",
-	proto = proto,
+local function marshal(cmd, body)
+	if type(cmd) == "string" then
+		cmd = proto:tag(cmd)
+	end
+	local dat, size = proto:encode(cmd, body, true)
+	local buf, size = proto:pack(dat, size, true)
+	return cmd, buf, size
+end
+
+local function call(msg, fd)
+	print("callee", msg.txt, fd)
+	return "pong", msg
+end
+
+local router = setmetatable({}, {__index = function(t, k) return call end})
+
+local server = cluster.new {
+	marshal = marshal,
+	unmarshal = unmarshal,
+	router = router,
 	accept = function(fd, addr)
 		print("accept", fd, addr)
 	end,
 
 	close = function(fd, errno)
 		print("close", fd, errno)
 	end,
-
-	call = function(msg, cmd, fd)
-		print("callee", msg.txt, cmd, fd)
-		return "pong", msg
-	end
 }
 
+server.listen("127.0.0.1:9999")
+
+local client = cluster.new {
+	marshal = marshal,
+	unmarshal = unmarshal,
+	router = router,
+	close = function(fd, errno)
+		print("close", fd, errno)
+	end,
+}
 
 core.start(function()
 	for i = 1, 3 do
 		core.fork(function()
-			local conn = rpc.connect {
-				addr = "127.0.0.1:9999",
-				proto = proto,
-				timeout = 5000,
-				close = function(fd, errno)
-				end,
-			}
-			while true do
+			local fd, err = client.connect("127.0.0.1:9999")
+			print("connect", fd, err)
+			for j = 1, 10000 do
 				local txt = crypto.randomkey(5)
-				local ack, cmd = conn:call("ping", {txt = txt})
-				print("caller", conn, txt, ack.txt)
+				local ack, cmd = client.ping(fd, {txt = txt})
+				print("caller", fd, txt, ack.txt)
 				assert(ack.txt == txt)
 				assert(cmd == proto:tag("pong"))
 				core.sleep(1000)