mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 20:23:06 +00:00
bugfix: auth before connected
This commit is contained in:
@@ -89,6 +89,8 @@ function redis.connect(dbname)
|
|||||||
port = db_conf.port or 6379,
|
port = db_conf.port or 6379,
|
||||||
auth = redis_login(db_conf.auth, db_conf.db),
|
auth = redis_login(db_conf.auth, db_conf.db),
|
||||||
}
|
}
|
||||||
|
-- try connect first
|
||||||
|
channel:connect()
|
||||||
return setmetatable( { channel }, meta )
|
return setmetatable( { channel }, meta )
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -180,6 +182,8 @@ function redis.watch(dbname)
|
|||||||
}
|
}
|
||||||
obj.__sock = channel
|
obj.__sock = channel
|
||||||
|
|
||||||
|
-- try connect first
|
||||||
|
channel:connect()
|
||||||
return setmetatable( obj, watchmeta )
|
return setmetatable( obj, watchmeta )
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|||||||
@@ -337,6 +337,7 @@ function socket.channel(desc)
|
|||||||
__connecting = {},
|
__connecting = {},
|
||||||
__sock = false,
|
__sock = false,
|
||||||
__closed = false,
|
__closed = false,
|
||||||
|
__authcoroutine = false,
|
||||||
}
|
}
|
||||||
|
|
||||||
return setmetatable(c, channel_meta)
|
return setmetatable(c, channel_meta)
|
||||||
@@ -346,7 +347,8 @@ local function close_channel_socket(self)
|
|||||||
if self.__sock then
|
if self.__sock then
|
||||||
local so = self.__sock
|
local so = self.__sock
|
||||||
self.__sock = false
|
self.__sock = false
|
||||||
socket.close(so[1])
|
-- never raise error
|
||||||
|
pcall(socket.close,so[1])
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -420,12 +422,13 @@ local function dispatch_response(self)
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
local function try_connect(self)
|
local function connect_once(self)
|
||||||
assert(not self.__sock)
|
assert(not self.__sock and not self.__authcoroutine)
|
||||||
local fd = socket.open(self.__host, self.__port)
|
local fd = socket.open(self.__host, self.__port)
|
||||||
if not fd then
|
if not fd then
|
||||||
return
|
return false
|
||||||
end
|
end
|
||||||
|
self.__authcoroutine = coroutine.running()
|
||||||
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
||||||
skynet.fork(dispatch_response, self)
|
skynet.fork(dispatch_response, self)
|
||||||
|
|
||||||
@@ -434,46 +437,20 @@ local function try_connect(self)
|
|||||||
if not ok then
|
if not ok then
|
||||||
close_channel_socket(self)
|
close_channel_socket(self)
|
||||||
if message ~= socket_error then
|
if message ~= socket_error then
|
||||||
error(message)
|
print("socket: auth failed", message)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
self.__authcoroutine = false
|
||||||
|
return ok
|
||||||
end
|
end
|
||||||
|
|
||||||
|
return true
|
||||||
end
|
end
|
||||||
|
|
||||||
function channel:connect()
|
local function try_connect(self)
|
||||||
if self.__sock then
|
|
||||||
return true
|
|
||||||
end
|
|
||||||
if self.__closed then
|
|
||||||
self.__closed = false
|
|
||||||
end
|
|
||||||
if #self.__connecting > 0 then
|
|
||||||
-- connecting in other coroutine
|
|
||||||
local co = coroutine.running()
|
|
||||||
table.insert(self.__connecting, co)
|
|
||||||
skynet.wait()
|
|
||||||
else
|
|
||||||
self.__connecting[1] = true
|
|
||||||
try_connect(self)
|
|
||||||
self.__connecting[1] = nil
|
|
||||||
for i=2, #self.__connecting do
|
|
||||||
local co = self.__connecting[i]
|
|
||||||
self.__connecting[i] = nil
|
|
||||||
skynet.wakeup(co)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
if self.__sock then
|
|
||||||
return true
|
|
||||||
else
|
|
||||||
return false
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
local function reconnect_channel(self)
|
|
||||||
local t = 100
|
local t = 100
|
||||||
while not self.__closed do
|
while not self.__closed do
|
||||||
if self:connect() then
|
if connect_once(self) then
|
||||||
return
|
return
|
||||||
end
|
end
|
||||||
if t > 1000 then
|
if t > 1000 then
|
||||||
@@ -487,11 +464,52 @@ local function reconnect_channel(self)
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
function channel:request(request, response)
|
local function block_connect(self)
|
||||||
if not self.__sock then
|
if self.__sock then
|
||||||
assert(not self.__closed)
|
local authco = self.__authcoroutine
|
||||||
reconnect_channel(self)
|
if not authco then
|
||||||
|
return true
|
||||||
|
end
|
||||||
|
if authco == coroutine.running() then
|
||||||
|
-- authing
|
||||||
|
return true
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
if self.__closed then
|
||||||
|
return false
|
||||||
|
end
|
||||||
|
if #self.__connecting > 0 then
|
||||||
|
-- connecting in other coroutine
|
||||||
|
local co = coroutine.running()
|
||||||
|
table.insert(self.__connecting, co)
|
||||||
|
skynet.wait()
|
||||||
|
-- check connection again
|
||||||
|
return block_connect(self)
|
||||||
|
end
|
||||||
|
|
||||||
|
self.__connecting[1] = true
|
||||||
|
try_connect(self)
|
||||||
|
self.__connecting[1] = nil
|
||||||
|
for i=2, #self.__connecting do
|
||||||
|
local co = self.__connecting[i]
|
||||||
|
self.__connecting[i] = nil
|
||||||
|
skynet.wakeup(co)
|
||||||
|
end
|
||||||
|
|
||||||
|
-- check again
|
||||||
|
return block_connect(self)
|
||||||
|
end
|
||||||
|
|
||||||
|
function channel:connect()
|
||||||
|
if self.__closed then
|
||||||
|
self.__closed = false
|
||||||
|
end
|
||||||
|
|
||||||
|
return block_connect(self)
|
||||||
|
end
|
||||||
|
|
||||||
|
function channel:request(request, response)
|
||||||
|
assert(block_connect(self))
|
||||||
|
|
||||||
if not socket.write(self.__sock[1], request) then
|
if not socket.write(self.__sock[1], request) then
|
||||||
return self:request(request, response)
|
return self:request(request, response)
|
||||||
@@ -531,10 +549,8 @@ function channel:request(request, response)
|
|||||||
end
|
end
|
||||||
|
|
||||||
function channel:response(response)
|
function channel:response(response)
|
||||||
if not self.__sock then
|
assert(block_connect(self))
|
||||||
assert(not self.__closed)
|
|
||||||
reconnect_channel(self)
|
|
||||||
end
|
|
||||||
assert(type(response) == "function")
|
assert(type(response) == "function")
|
||||||
|
|
||||||
local co = coroutine.running()
|
local co = coroutine.running()
|
||||||
|
|||||||
Reference in New Issue
Block a user