From 1c3b563ef8133816752ee45499ab0b645500c357 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Wed, 20 Mar 2019 19:44:08 +0800
Subject: [PATCH] add wait queue

---
 service/clustersender.lua | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/service/clustersender.lua b/service/clustersender.lua
index d9fa29ab..250d70b2 100644
--- a/service/clustersender.lua
+++ b/service/clustersender.lua
@@ -8,6 +8,7 @@ local channel
 local session = 1
 
 local command = {}
+local waiting = {}
 
 local function send_request(addr, msg, sz)
 	-- msg is a local pointer, cluster.packrequest will free it
@@ -30,7 +31,16 @@ local function send_request(addr, msg, sz)
 	return channel:request(request, current_session, padding)
 end
 
+local function wait()
+	local co = coroutine.running()
+	table.insert(waiting, co)
+	skynet.wait(co)
+end
+
 function command.req(...)
+	if channel == nil then
+		wait()
+	end
 	local ok, msg = pcall(send_request, ...)
 	if ok then
 		if type(msg) == "table" then
@@ -45,6 +55,9 @@ function command.req(...)
 end
 
 function command.push(addr, msg, sz)
+	if channel == nil then
+		wait()
+	end
 	local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
 	if padding then	-- is multi push
 		session = new_session
@@ -68,10 +81,18 @@ function command.changenode(host, port)
 			nodelay = true,
 		}
 	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))
 	else
+		channel = nil	-- reset channel
 		skynet.response()(false)
 	end
 end