lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-25 12:43:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

use socket_channle:changehost, this may also fix #1130

Browse Source
This commit is contained in:
Cloud Wu
2019-12-04 15:33:01 +08:00
parent b295b875c8
commit 99a7d745fb
2 changed files with 11 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
service/clusterd.lua
View File

@@ -35,7 +35,7 @@ local function open_channel(t, key)
local host, port = string.match(address, "([^:]+):(.*)$") local host, port = string.match(address, "([^:]+):(.*)$")
c = node_sender[key] c = node_sender[key]
if c == nil then if c == nil then
c = skynet.newservice("clustersender", key, nodename) c = skynet.newservice("clustersender", key, nodename, host, port)
if node_sender[key] then if node_sender[key] then
-- double check -- double check
skynet.kill(c) skynet.kill(c)

44
service/clustersender.lua
View File

@@ -5,11 +5,9 @@ local cluster = require "skynet.cluster.core"
local channel local channel
local session = 1 local session = 1
local node, nodename = ... local node, nodename, init_host, init_port = ...
local command = {} local command = {}
local waiting = {}
local function send_request(addr, msg, sz) local function send_request(addr, msg, sz)
-- msg is a local pointer, cluster.packrequest will free it -- msg is a local pointer, cluster.packrequest will free it
@@ -31,16 +29,7 @@ local function send_request(addr, msg, sz)
return channel:request(request, current_session, padding) return channel:request(request, current_session, padding)
end end
local function wait()
local co = coroutine.running()
table.insert(waiting, co)
skynet.wait(co)
end
function command.req(...) function command.req(...)
if channel == nil then
wait()
end
local ok, msg = pcall(send_request, ...) local ok, msg = pcall(send_request, ...)
if ok then if ok then
if type(msg) == "table" then if type(msg) == "table" then
@@ -55,9 +44,6 @@ function command.req(...)
end end
function command.push(addr, msg, sz) function command.push(addr, msg, sz)
if channel == nil then
wait()
end
local request, new_session, padding = cluster.packpush(addr, session, msg, sz) local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
if padding then -- is multi push if padding then -- is multi push
session = new_session session = new_session
@@ -73,30 +59,18 @@ local function read_response(sock)
end end
function command.changenode(host, port) function command.changenode(host, port)
local c = sc.channel { channel:changehost(host, tonumber(port))
host = host, channel:connect(true)
port = tonumber(port),
response = read_response,
nodelay = true,
}
local succ, err = pcall(c.connect, c, true)
if channel then
channel:close()
end
if succ then
channel = c
for k, co in ipairs(waiting) do
waiting[k] = nil
skynet.wakeup(co)
end
skynet.ret(skynet.pack(nil)) skynet.ret(skynet.pack(nil))
else
channel = nil -- reset channel
skynet.response()(false)
end
end end
skynet.start(function() 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, ...) skynet.dispatch("lua", function(session , source, cmd, ...)
local f = assert(command[cmd]) local f = assert(command[cmd])
f(...) f(...)