款台信息同步 [完成]

README.org

@@ -43,3 +43,4 @@
           2. [ ] 会员日统计
     2. [X] 数据同步
        1. [X] 门店信息
+       2. [X] 款台信息

nginx/lua/luabiz/node_cs.lua

@@ -68,6 +68,7 @@ end
 
 
 
+
 local function async(db_st, interval)
   local co = coroutine.wrap(
     function()

nginx/lua/luabiz/tk_casher.lua

+local _CASHER = {}
+
+local MODULE="S"                -- Settle
+
+-- 加载模块
+local mysql   = loadmod("resty.mysql")
+local zx_base = loadmod("zx_base")
+local json    = loadmod('cjson')
+local db_conf = "tk7_dbs.json"
+
+
+-- 协和组
+local tasks = {}
+
+
+local function EOF()
+  zx_base:log(MODULE, "-----------------------> END[TK_CASHER]")
+  ngx.eof()
+end
+
+
+-- 门店信息
+-- @param table tk_biz 节点机数据库对象
+--        table tk_control 总控机数据库对象
+-- @return nil
+local function casher_sql(tk_biz, tk_control)
+  zx_base:log(MODULE, "-----------------------> START[TK_CASHER]")
+
+  -- 新建mysql连接
+  local db, err = mysql:new()
+  if not db then
+    zx_base:log(MODULE, "[TK_CASHER]", "failed to instantiate mysql: ", err)
+    return false
+  end
+  -- 连接节点机MySQL
+  local ok, err, errcode, sqlstate = db:connect{
+    host     = tk_biz['host'],
+    port     = tk_biz['port'],
+    database = tk_biz['database'],
+    user     = tk_biz['user'],
+    password = tk_biz['password'],
+    timeout  = tk_biz['timeout'],
+    charset  = 'utf8'
+  }
+  if not ok then
+    zx_base:log(MODULE, "[TK_CASHER]", "failed to connect:", json.encode(tk_biz), ":",
+                err, ":", errcode, sqlstate)
+    return
+  end
+
+
+  -- 查询节点机
+  local sql = string.format([[SELECT * FROM tk_casher where update_time > %s]],
+    -- os.date("%Y%m%d000000", os.time()-24*60*60)
+    "20180716000000"
+  )
+
+  local cashers
+  cashers, err, errcode, sqlstate = db:query(sql)
+  if not cashers then
+    zx_base:log(MODULE, "[TK_CASHER]", sql)
+    zx_base:log(MODULE, "[TK_CASHER]", "failed to query:", json.encode(tk_biz), ":",
+                err, ":", errcode, ":", sqlstate, ".")
+    return
+  end
+  if #cashers == 0 then
+    ngx.say("no cashers found")
+    ngx.eof()
+    return
+  end
+
+
+
+  -- 重置MySQL连接
+  zx_base:close_db(db)
+  db, err = mysql:new()
+  if not db then
+    zx_base:log(MODULE, "[TK_CASHER]", "failed to instantiate mysql: ", err)
+    return false
+  end
+  -- 连接总控机MySQL
+  ok, err, errcode, sqlstate = db:connect{
+    host     = tk_control['host'],
+    port     = tk_control['port'],
+    database = tk_control['database'],
+    user     = tk_control['user'],
+    password = tk_control['password'],
+    timeout  = tk_control['timeout'],
+    charset  = 'utf8'
+  }
+  if not ok then
+    zx_base:log(MODULE, "[TK_CASHER]", "failed to connect:", json.encode(tk_biz), ":", errcode, sqlstate)
+    return
+  end
+
+
+  -- 插入总控机
+  for i = 1, #cashers do
+    local casher = cashers[i]
+    local sql_result = zx_base:sql_concate("duplicate", "tk_casher", casher)
+
+    ok, err, errcode, sqlstate = db:query(sql_result)
+    if not ok then
+      zx_base:log(MODULE, "[TK_CASHER]", sql_result)
+      zx_base:log(MODULE, "[TK_CASHER]", "failed to query:", json.encode(tk_control), ":",
+                  err, ":", errcode, ":", sqlstate, ".")
+    end
+  end
+
+  ngx.say("SUCCESS")
+  EOF()
+end
+
+
+local function async(tk_biz, tk_control)
+  local co = coroutine.wrap(
+    function()
+      casher_sql(tk_biz, tk_control)
+    end
+  )
+  table.insert(tasks, co)
+end
+
+local function dispatch()
+  local i = 1
+  while true do
+    if tasks[i] == nil then
+      if tasks[1] == nil then
+        break
+      end
+      i = 1
+    end
+
+    local res = tasks[i]()
+    if not res then
+      table.remove(tasks, i)
+    else
+      i = i + 1
+    end
+  end
+end
+
+function _CASHER.run()
+  local dbs_st = zx_base:db_read(db_conf)
+  for i = 1, #dbs_st.tk_biz do
+    async(dbs_st.tk_biz[i], dbs_st.tk_control)
+  end
+  dispatch()
+end
+
+return _CASHER