foobar 1 { # 定义一个新协议 (for RPC used) tag 1 request Person # 把数据类型 Person与 foobar.request 相关联 response { # 定义 foobar.response 的数据类型 ok 0 : boolean } }
一个框架可以是 被 sproto 框架语言自描述的：
.type { .field { name 0 : string buildin 1 : integer type 2 : integer # type is fixed-point number precision when buildin is SPROTO_TINTEGER; When buildin is SPROTO_TSTRING, it means binary string when type is 1. tag 3 : integer array 4 : boolean key 5 : integer # If key exists, array must be true, and it's a map. } name 0 : string fields 1 : *field }
.protocol { name 0 : string tag 1 : integer request 2 : integer # index response 3 : integer # index confirm 4 : boolean # response nil where confirm == true }
.group { type 0 : *type protocol 1 : *protocol }
Wire protocol
每个整数以小端（little endian）格式序列化。
sproto 消息必须是一个用户定义类型结构，每个结构编码成三个部分。header, field, data（头部，字段，数据）。标签（tag）和 小的整数 或 布尔值 会被编码到 field 部分，其他的都在 data 部分。
structsproto_arg { void *ud; constchar *tagname; int tagid; int type; structsproto_type *subtype; void *value; int length; int index; // array base 1 int mainindex; // for map int extra; // SPROTO_TINTEGER: fixed-point presision ; SPROTO_TSTRING 0:utf8 string 1:binary };
functionhost:attach(sp) returnfunction(name, args, session, ud) // 在 sproto 对象内查找 name 协议 local proto = queryproto(sp, name) // 消息头部 { type, session, ud} header_tmp.type = proto.tag header_tmp.session = session header_tmp.ud = ud // 头部进行 0 packing local header = core.encode(self.__package, header_tmp)
if session then self.__session[session] = proto.response ortrue end
// 封装请求内容 if proto.request then local content = core.encode(proto.request, args) return core.pack(header .. content) else return core.pack(header) end end end
这个函数会返回一个函数：
function(name, args, session, ud) ... end
其会根据 name（协议类型/请求类型）来把 代表内容的 args, session 打包。
host:dispatch
我们先来看一下 dispatch方法：
functionhost:dispatch(...) local bin = core.unpack(...) header_tmp.type = nil header_tmp.session = nil header_tmp.ud = nil local header, size = core.decode(self.__package, bin, header_tmp) local content = bin:sub(size + 1) if header.typethen -- request local proto = queryproto(self.__proto, header.type) local result if proto.request then result = core.decode(proto.request, content) end if header_tmp.session then return"REQUEST", proto.name, result, gen_response(self, proto.response, header_tmp.session), header.ud else return"REQUEST", proto.name, result, nil, header.ud end else -- response local session = assert(header_tmp.session, "session not found") local response = assert(self.__session[session], "Unknown session") self.__session[session] = nil if response == truethen return"RESPONSE", session, nil, header.ud else local result = core.decode(response, content) return"RESPONSE", session, result, header.ud end end end
消息发送
在调用 local request = host:attach(sproto.new(proto.c2s))后，建立了一个消息封装函数request。
函数 ：
localfunctionsend_request(name, args) session = session + 1 local str = request(name, args, session) send_package(fd, str) print("Request:", session) end
会将 会话 ID，协议名，参数传递给 消息封装函数。之后，函数：
localfunctionsend_package(fd, pack) localpackage = string.pack(">s2", pack) socket.send(fd, package) end
-- services/gate.lua functionhandler.message(fd, msg, sz) -- recv a package, forward it local c = connection[fd] local agent = c.agent if agent then skynet.redirect(agent, c.client, "client", 1, msg, sz) else skynet.send(watchdog, "lua", "socket", "data", fd, netpack.tostring(msg, sz)) end end
我们的 agent 服务在启动时即注册了 client 类型的消息：
-- examples/agent.lua skynet.register_protocol { name = "client", id = skynet.PTYPE_CLIENT, unpack = function(msg, sz) return host:dispatch(msg, sz) end, dispatch = function(_, _, type, ...) iftype == "REQUEST"then local ok, result = pcall(request, ...) if ok then if result then send_package(result) end else skynet.error(result) end else assert(type == "RESPONSE") error"This example doesn't support request client" end end }
其会使用 host:dispatch来解压消息，然后注册了自己的消息回调函数。
我们注意到，在服务端中，建立消息消息分发器的方式同客户端似乎都不一样：
-- examples/agent.lua functionCMD.start(conf) local fd = conf.client local gate = conf.gate WATCHDOG = conf.watchdog -- slot 1,2 set at main.lua host = sprotoloader.load(1):host "package" send_request = host:attach(sprotoloader.load(2)) skynet.fork(function() whiletruedo send_package(send_request "heartbeat") skynet.sleep(500) end end)
client_fd = fd skynet.call(gate, "lua", "forward", fd) end
skynet.start(function() sprotoloader.save(proto.c2s, 1) sprotoloader.save(proto.s2c, 2) -- don't call skynet.exit() , because sproto.core may unload and the global slot become invalid end)