This commit is contained in:
Cloud Wu
2018-08-17 21:16:35 +08:00
parent 143baec319
commit 4613fde654

View File

@@ -77,17 +77,6 @@ local function wakeup_all(self, errmsg)
end
end
local function exit_thread(self)
local co = coroutine.running()
if self.__dispatch_thread == co then
self.__dispatch_thread = nil
local connecting = self.__connecting_thread
if connecting then
skynet.wakeup(connecting)
end
end
end
local function dispatch_by_session(self)
local response = self.__response
-- response() return session
@@ -124,7 +113,7 @@ local function dispatch_by_session(self)
wakeup_all(self, errormsg)
end
end
exit_thread(self)
self.__dispatch_thread = nil
end
local function pop_response(self)
@@ -201,7 +190,7 @@ local function dispatch_by_order(self)
wakeup_all(self, errmsg)
end
end
exit_thread(self)
self.__dispatch_thread = nil
end
local function dispatch_function(self)
@@ -233,11 +222,21 @@ local function connect_backup(self)
end
end
local function term_dispatch_thread(self)
if not self.__response and self.__dispatch_thread then
-- dispatch by order, send close signal to dispatch thread
push_response(self, true, false) -- (true, false) is close signal
end
end
local function connect_once(self)
if self.__closed then
return false
end
assert(not self.__sock and not self.__authcoroutine)
-- term current dispatch thread (send a sigal)
term_dispatch_thread(self, self.__dispatch_thread)
local fd,err = socket.open(self.__host, self.__port)
if not fd then
fd = connect_backup(self)
@@ -249,6 +248,11 @@ local function connect_once(self)
socketdriver.nodelay(fd)
end
while self.__dispatch_thread do
-- wait for dispatch thread exit
skynet.yield()
end
self.__sock = setmetatable( {fd} , channel_socket_meta )
self.__dispatch_thread = skynet.fork(dispatch_function(self), self)
@@ -353,18 +357,7 @@ local function block_connect(self, once)
end
function channel:connect(once)
if self.__closed then
if self.__dispatch_thread then
-- closing, wait
assert(self.__connecting_thread == nil, "already connecting")
local co = coroutine.running()
self.__connecting_thread = co
skynet.wait(co)
self.__connecting_thread = nil
end
self.__closed = false
end
self.__closed = false
return block_connect(self, once)
end
@@ -437,13 +430,9 @@ end
function channel:close()
if not self.__closed then
local thread = self.__dispatch_thread
term_dispatch_thread(self)
self.__closed = true
close_channel_socket(self)
if not self.__response and self.__dispatch_thread == thread and thread then
-- dispatch by order, send close signal to dispatch thread
push_response(self, true, false) -- (true, false) is close signal
end
end
end