1

{NodeName = "[xxx.xxx.xxx.xxx]:xxxx"}

1
2
3
4
5
6
7

function cluster.open(port)
    if type(port) == "string" then
        skynet.call(clusterd, "lua", "listen", port)
    else
        skynet.call(clusterd, "lua", "listen", "0.0.0.0", port)
    end
end

1
2
3
4
5
6
7
8
9

function command.listen(source, addr, port)
    local gate = skynet.newservice("gate")
    if port == nil then
        local address = assert(node_address[addr], addr .. " is down")
        addr, port = string.match(address, "([^:]+):(.*)$")
    end
    skynet.call(gate, "lua", "open", { address = addr, port = port })
    skynet.ret(skynet.pack(nil))
end

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

function command.socket(source, subcmd, fd, msg)
    if subcmd == "open" then
        skynet.error(string.format("socket accept from %s", msg))
        -- new cluster agent
        cluster_agent[fd] = false
        local agent = skynet.newservice("clusteragent", skynet.self(), source, fd)
        local closed = cluster_agent[fd]
        cluster_agent[fd] = agent
        if closed then
            skynet.send(agent, "lua", "exit")
            cluster_agent[fd] = nil
        end
    else
        -- ... 省略非相关代码
    end
end

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

-- clusteragent 的启动函数
skynet.start(function()
    skynet.register_protocol {
        name = "client",
        id = skynet.PTYPE_CLIENT,
        unpack = cluster.unpackrequest,
        dispatch = dispatch_request,
    }
    -- fd can write, but don't read fd, the data package will forward from gate though client protocol.
    skynet.call(gate, "lua", "forward", fd)

    skynet.dispatch("lua", function(_,source, cmd, ...)
        if cmd == "exit" then
            socket.close(fd)
            skynet.exit()
        elseif cmd == "namechange" then
            register_name = new_register_name()
        else
            skynet.error(string.format("Invalid command %s from %s", cmd, skynet.address(source)))
        end
    end)
end)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

skynet.start(function()
    channel = sc.channel { -- 这一步只是创建，不会发起连接
            host = init_host,
            port = tonumber(init_port),
            response = read_response,
            nodelay = true,
        }
    skynet.dispatch("lua", function(session , source, cmd, ...)
        local f = assert(command[cmd])
        f(...)
    end)
end)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

function command.register(source, name, addr)
    assert(register_name[name] == nil)
    addr = addr or source
    local old_name = register_name[addr]
    if old_name then
        register_name[old_name] = nil
        clearnamecache()
    end
    register_name[addr] = name
    register_name[name] = addr
    skynet.ret(nil)
    skynet.error(string.format("Register [%s] :%08x", name, addr))
end

1
2
3
4
5
6
7
8

function command.push(addr, msg, sz)
    local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
    if padding then	-- is multi push
        session = new_session
    end

    channel:request(request, nil, padding)
end

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

function handler.message(fd, msg, sz)
    -- recv a package, forward it
    local c = connection[fd]
    local agent = c.agent
    if agent then
        -- It's safe to redirect msg directly , gateserver framework will not free msg.
        skynet.redirect(agent, c.client, "client", fd, msg, sz) -- 重定向到客户端
    else
        skynet.send(watchdog, "lua", "socket", "data", fd, skynet.tostring(msg, sz))
        -- skynet.tostring will copy msg to a string, so we must free msg here.
        skynet.trash(msg, sz)
    end
end

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

-- 经过大量精简后的 dispatch_request
local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push)
    local ok, response
    if cluster.isname(addr) then
        addr = register_name[addr]
    end
    if addr then
        if is_push then
            skynet.rawsend(addr, "lua", msg, sz) -- 将消息发送转指定地址服务
            return -- no response
        else
            if tracetag then
                ok, msg, sz = pcall(skynet.tracecall, tracetag, addr, "lua", msg, sz)
                tracetag = nil
            else
                ok, msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
            end
        end
    else
        ok = false
        msg = "Invalid name"
    end
    if ok then
        response = cluster.packresponse(session, true, msg, sz)
        if type(response) == "table" then
            for _, v in ipairs(response) do
                socket.lwrite(fd, v)
            end
        else
            socket.write(fd, response)
        end
    else
        response = cluster.packresponse(session, false, msg)
        socket.write(fd, response)
    end
end