mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 03:53:09 +00:00
format up by vscode plugin
This commit is contained in:
@@ -7,7 +7,6 @@
|
||||
local socketchannel = require "skynet.socketchannel"
|
||||
local crypt = require "skynet.crypt"
|
||||
|
||||
|
||||
local sub = string.sub
|
||||
local strgsub = string.gsub
|
||||
local strformat = string.format
|
||||
@@ -16,23 +15,23 @@ local strchar = string.char
|
||||
local strrep = string.rep
|
||||
local strunpack = string.unpack
|
||||
local strpack = string.pack
|
||||
local sha1= crypt.sha1
|
||||
local sha1 = crypt.sha1
|
||||
local setmetatable = setmetatable
|
||||
local error = error
|
||||
local tonumber = tonumber
|
||||
local tointeger = math.tointeger
|
||||
|
||||
local _M = { _VERSION = '0.14' }
|
||||
local _M = {_VERSION = "0.14"}
|
||||
|
||||
-- constants
|
||||
local COM_QUERY = '\x03'
|
||||
local COM_PING = '\x0e'
|
||||
local COM_STMT_PREPARE = '\x16'
|
||||
local COM_STMT_EXECUTE = '\x17'
|
||||
local COM_QUERY = "\x03"
|
||||
local COM_PING = "\x0e"
|
||||
local COM_STMT_PREPARE = "\x16"
|
||||
local COM_STMT_EXECUTE = "\x17"
|
||||
local CURSOR_TYPE_NO_CURSOR = 0x00
|
||||
local SERVER_MORE_RESULTS_EXISTS = 8
|
||||
|
||||
local mt = { __index = _M }
|
||||
local mt = {__index = _M}
|
||||
|
||||
-- mysql field value type converters
|
||||
local converters = {}
|
||||
@@ -41,37 +40,37 @@ for i = 0x01, 0x05 do
|
||||
-- tiny, short, long, float, double
|
||||
converters[i] = tonumber
|
||||
end
|
||||
converters[0x08] = tonumber -- long long
|
||||
converters[0x09] = tonumber -- int24
|
||||
converters[0x0d] = tonumber -- year
|
||||
converters[0xf6] = tonumber -- newdecimal
|
||||
converters[0x08] = tonumber -- long long
|
||||
converters[0x09] = tonumber -- int24
|
||||
converters[0x0d] = tonumber -- year
|
||||
converters[0xf6] = tonumber -- newdecimal
|
||||
|
||||
local function _get_byte1(data, i)
|
||||
return strbyte(data,i),i+1
|
||||
return strbyte(data, i), i + 1
|
||||
end
|
||||
|
||||
local function _get_byte2(data, i)
|
||||
return strunpack("<I2",data,i)
|
||||
return strunpack("<I2", data, i)
|
||||
end
|
||||
|
||||
local function _get_byte3(data, i)
|
||||
return strunpack("<I3",data,i)
|
||||
return strunpack("<I3", data, i)
|
||||
end
|
||||
|
||||
local function _get_byte4(data, i)
|
||||
return strunpack("<I4",data,i)
|
||||
return strunpack("<I4", data, i)
|
||||
end
|
||||
|
||||
local function _get_byte8(data, i)
|
||||
return strunpack("<I8",data,i)
|
||||
return strunpack("<I8", data, i)
|
||||
end
|
||||
|
||||
local function _get_float(data, i)
|
||||
return strunpack("<f",data,i)
|
||||
return strunpack("<f", data, i)
|
||||
end
|
||||
|
||||
local function _get_double(data, i)
|
||||
return strunpack("<d",data,i)
|
||||
return strunpack("<d", data, i)
|
||||
end
|
||||
|
||||
local function _set_byte2(n)
|
||||
@@ -103,7 +102,13 @@ local function _from_cstring(data, i)
|
||||
end
|
||||
|
||||
local function _dumphex(bytes)
|
||||
return strgsub(bytes, ".", function(x) return strformat("%02x ", strbyte(x)) end)
|
||||
return strgsub(
|
||||
bytes,
|
||||
".",
|
||||
function(x)
|
||||
return strformat("%02x ", strbyte(x))
|
||||
end
|
||||
)
|
||||
end
|
||||
|
||||
local function _compute_token(password, scramble)
|
||||
@@ -118,22 +123,25 @@ local function _compute_token(password, scramble)
|
||||
local stage3 = sha1(scramble .. stage2)
|
||||
|
||||
local i = 0
|
||||
return strgsub(stage3,".",
|
||||
return strgsub(
|
||||
stage3,
|
||||
".",
|
||||
function(x)
|
||||
i = i + 1
|
||||
-- ~ is xor in lua 5.3
|
||||
return strchar(strbyte(x) ~ strbyte(stage1, i))
|
||||
end)
|
||||
end
|
||||
)
|
||||
end
|
||||
|
||||
local function _compose_packet(self, req)
|
||||
self.packet_no = self.packet_no + 1
|
||||
local size = #req
|
||||
return strpack("<I3Bc"..size,size,self.packet_no,req)
|
||||
self.packet_no = self.packet_no + 1
|
||||
local size = #req
|
||||
return strpack("<I3Bc" .. size, size, self.packet_no, req)
|
||||
end
|
||||
|
||||
local function _recv_packet(self,sock)
|
||||
local data = sock:read( 4)
|
||||
local function _recv_packet(self, sock)
|
||||
local data = sock:read(4)
|
||||
if not data then
|
||||
return nil, nil, "failed to receive packet header: "
|
||||
end
|
||||
@@ -199,19 +207,19 @@ local function _from_length_coded_bin(data, pos)
|
||||
end
|
||||
|
||||
local function _set_length_coded_bin(n)
|
||||
if n<251 then
|
||||
if n < 251 then
|
||||
return strchar(n)
|
||||
end
|
||||
|
||||
if n<(1<<16) then
|
||||
return strpack('<BI2',0xfc,n)
|
||||
if n < (1 << 16) then
|
||||
return strpack("<BI2", 0xfc, n)
|
||||
end
|
||||
|
||||
if n<(1<<24) then
|
||||
return strpack('<BI3',0xfd,n)
|
||||
if n < (1 << 24) then
|
||||
return strpack("<BI3", 0xfd, n)
|
||||
end
|
||||
|
||||
return strpack('<BI8',0xfe,n)
|
||||
return strpack("<BI8", 0xfe, n)
|
||||
end
|
||||
|
||||
local function _from_length_coded_str(data, pos)
|
||||
@@ -250,7 +258,7 @@ local function _parse_err_packet(packet)
|
||||
local errno, pos = _get_byte2(packet, 2)
|
||||
local marker = sub(packet, pos, pos)
|
||||
local sqlstate
|
||||
if marker == '#' then
|
||||
if marker == "#" then
|
||||
-- with sqlstate
|
||||
pos = pos + 1
|
||||
sqlstate = sub(packet, pos, pos + 5 - 1)
|
||||
@@ -291,7 +299,6 @@ local function _parse_field_packet(data)
|
||||
col.default = default
|
||||
end
|
||||
--]]
|
||||
|
||||
return col
|
||||
end
|
||||
|
||||
@@ -333,7 +340,7 @@ local function _recv_field_packet(self, sock)
|
||||
return nil, msg, errno, sqlstate
|
||||
end
|
||||
|
||||
if typ ~= 'DATA' then
|
||||
if typ ~= "DATA" then
|
||||
return nil, "bad field packet type: " .. typ
|
||||
end
|
||||
|
||||
@@ -343,18 +350,18 @@ local function _recv_field_packet(self, sock)
|
||||
end
|
||||
|
||||
local function _recv_decode_packet_resp(self)
|
||||
return function(sock)
|
||||
local packet, typ, err = _recv_packet(self,sock)
|
||||
return function(sock)
|
||||
local packet, typ, err = _recv_packet(self, sock)
|
||||
if not packet then
|
||||
return false, "failed to receive the result packet"..err
|
||||
return false, "failed to receive the result packet" .. err
|
||||
end
|
||||
|
||||
if typ == 'ERR' then
|
||||
if typ == "ERR" then
|
||||
local errno, msg, sqlstate = _parse_err_packet(packet)
|
||||
return false, strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate)
|
||||
return false, strformat("errno:%d, msg:%s,sqlstate:%s", errno, msg, sqlstate)
|
||||
end
|
||||
|
||||
if typ == 'EOF' then
|
||||
if typ == "EOF" then
|
||||
return false, "old pre-4.1 authentication protocol not supported"
|
||||
end
|
||||
|
||||
@@ -362,10 +369,10 @@ local function _recv_decode_packet_resp(self)
|
||||
end
|
||||
end
|
||||
|
||||
local function _mysql_login(self,user,password,database,on_connect)
|
||||
local function _mysql_login(self, user, password, database, on_connect)
|
||||
return function(sockchannel)
|
||||
local dispatch_resp = _recv_decode_packet_resp(self)
|
||||
local packet = sockchannel:response( dispatch_resp )
|
||||
local packet = sockchannel:response(dispatch_resp)
|
||||
|
||||
self.protocol_ver = strbyte(packet)
|
||||
|
||||
@@ -393,7 +400,7 @@ local function _mysql_login(self,user,password,database,on_connect)
|
||||
local more_capabilities
|
||||
more_capabilities, pos = _get_byte2(packet, pos)
|
||||
|
||||
self._server_capabilities = self._server_capabilities|more_capabilities<<16
|
||||
self._server_capabilities = self._server_capabilities | more_capabilities << 16
|
||||
|
||||
local len = 21 - 8 - 1
|
||||
pos = pos + 1 + 10
|
||||
@@ -403,17 +410,20 @@ local function _mysql_login(self,user,password,database,on_connect)
|
||||
error "2nd part of scramble not found"
|
||||
end
|
||||
|
||||
local scramble = scramble1..scramble_part2
|
||||
local scramble = scramble1 .. scramble_part2
|
||||
local token = _compute_token(password, scramble)
|
||||
local client_flags = 260047;
|
||||
local req = strpack("<I4I4c24zs1z",
|
||||
local client_flags = 260047
|
||||
local req =
|
||||
strpack(
|
||||
"<I4I4c24zs1z",
|
||||
client_flags,
|
||||
self._max_packet_size,
|
||||
strrep("\0", 24), -- TODO: add support for charset encoding
|
||||
strrep("\0", 24), -- TODO: add support for charset encoding
|
||||
user,
|
||||
token,
|
||||
database)
|
||||
local authpacket=_compose_packet(self,req)
|
||||
database
|
||||
)
|
||||
local authpacket = _compose_packet(self, req)
|
||||
sockchannel:request(authpacket, dispatch_resp)
|
||||
if on_connect then
|
||||
on_connect(self)
|
||||
@@ -429,7 +439,7 @@ end
|
||||
|
||||
local function _compose_query(self, query)
|
||||
self.packet_no = -1
|
||||
local cmd_packet = COM_QUERY..query
|
||||
local cmd_packet = COM_QUERY .. query
|
||||
return _compose_packet(self, cmd_packet)
|
||||
end
|
||||
|
||||
@@ -441,50 +451,50 @@ end
|
||||
|
||||
--参数字段类型转换
|
||||
local store_types = {
|
||||
number=function(v)
|
||||
number = function(v)
|
||||
if not tointeger(v) then
|
||||
return _set_byte2(0x05),_set_double(v)
|
||||
return _set_byte2(0x05), _set_double(v)
|
||||
else
|
||||
return _set_byte2(0x08),_set_byte8(v)
|
||||
return _set_byte2(0x08), _set_byte8(v)
|
||||
end
|
||||
end,
|
||||
string=function(v)
|
||||
return _set_byte2(0x0f),_set_length_coded_bin(#v)..v
|
||||
string = function(v)
|
||||
return _set_byte2(0x0f), _set_length_coded_bin(#v) .. v
|
||||
end,
|
||||
--bool转换为0,1
|
||||
boolean=function(v)
|
||||
boolean = function(v)
|
||||
if v then
|
||||
return _set_byte2(0x01),strchar(1)
|
||||
return _set_byte2(0x01), strchar(1)
|
||||
else
|
||||
return _set_byte2(0x01),strchar(0)
|
||||
return _set_byte2(0x01), strchar(0)
|
||||
end
|
||||
end,
|
||||
end
|
||||
}
|
||||
|
||||
local function _compose_stmt_execute(self,stmt,cursor_type,args)
|
||||
local function _compose_stmt_execute(self, stmt, cursor_type, args)
|
||||
local arg_num = #args
|
||||
if arg_num ~= stmt.param_count then
|
||||
error("require stmt.param_count "..stmt.param_count.." get arg_num "..arg_num)
|
||||
error("require stmt.param_count " .. stmt.param_count .. " get arg_num " .. arg_num)
|
||||
end
|
||||
|
||||
self.packet_no = -1
|
||||
|
||||
local cmd_packet = strpack("<c1I4BI4",COM_STMT_EXECUTE,stmt.prepare_id,cursor_type,0x01)
|
||||
local cmd_packet = strpack("<c1I4BI4", COM_STMT_EXECUTE, stmt.prepare_id, cursor_type, 0x01)
|
||||
if arg_num > 0 then
|
||||
local null_count=(arg_num+7)//8
|
||||
local f,ts,vs
|
||||
local types_buf=""
|
||||
local values_buf=""
|
||||
for _,v in pairs(args) do
|
||||
f= store_types[type(v)]
|
||||
local null_count = (arg_num + 7) // 8
|
||||
local f, ts, vs
|
||||
local types_buf = ""
|
||||
local values_buf = ""
|
||||
for _, v in pairs(args) do
|
||||
f = store_types[type(v)]
|
||||
if not f then
|
||||
error("invalid parameter type",type(v))
|
||||
error("invalid parameter type", type(v))
|
||||
end
|
||||
ts,vs = f(v)
|
||||
types_buf=types_buf..ts
|
||||
values_buf=values_buf..vs
|
||||
ts, vs = f(v)
|
||||
types_buf = types_buf .. ts
|
||||
values_buf = values_buf .. vs
|
||||
end
|
||||
cmd_packet = cmd_packet..strrep("\0",null_count)..strchar(0x01)..types_buf..values_buf
|
||||
cmd_packet = cmd_packet .. strrep("\0", null_count) .. strchar(0x01) .. types_buf .. values_buf
|
||||
end
|
||||
|
||||
return _compose_packet(self, cmd_packet)
|
||||
@@ -494,26 +504,26 @@ local function read_result(self, sock)
|
||||
local packet, typ, err = _recv_packet(self, sock)
|
||||
if not packet then
|
||||
return nil, err
|
||||
--error( err )
|
||||
--error( err )
|
||||
end
|
||||
|
||||
if typ == "ERR" then
|
||||
local errno, msg, sqlstate = _parse_err_packet(packet)
|
||||
return nil, msg, errno, sqlstate
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
end
|
||||
|
||||
if typ == 'OK' then
|
||||
if typ == "OK" then
|
||||
local res = _parse_ok_packet(packet)
|
||||
if res and res.server_status&SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
if res and res.server_status & SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
return res, "again"
|
||||
end
|
||||
return res
|
||||
end
|
||||
|
||||
if typ ~= 'DATA' then
|
||||
if typ ~= "DATA" then
|
||||
return nil, "packet type " .. typ .. " not supported"
|
||||
--error( "packet type " .. typ .. " not supported" )
|
||||
--error( "packet type " .. typ .. " not supported" )
|
||||
end
|
||||
|
||||
-- typ == 'DATA'
|
||||
@@ -524,7 +534,7 @@ local function read_result(self, sock)
|
||||
local col, err, errno, sqlstate = _recv_field_packet(self, sock)
|
||||
if not col then
|
||||
return nil, err, errno, sqlstate
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
end
|
||||
cols[i] = col
|
||||
end
|
||||
@@ -535,9 +545,9 @@ local function read_result(self, sock)
|
||||
return nil, err
|
||||
end
|
||||
|
||||
if typ ~= 'EOF' then
|
||||
if typ ~= "EOF" then
|
||||
--error ( "unexpected packet type " .. typ .. " while eof packet is ".. "expected" )
|
||||
return nil, "unexpected packet type " .. typ .. " while eof packet is ".. "expected"
|
||||
return nil, "unexpected packet type " .. typ .. " while eof packet is " .. "expected"
|
||||
end
|
||||
|
||||
-- typ == 'EOF'
|
||||
@@ -552,16 +562,16 @@ local function read_result(self, sock)
|
||||
return nil, err
|
||||
end
|
||||
|
||||
if typ == 'EOF' then
|
||||
if typ == "EOF" then
|
||||
local warning_count, status_flags = _parse_eof_packet(packet)
|
||||
if status_flags&SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
if status_flags & SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
return rows, "again"
|
||||
end
|
||||
break
|
||||
end
|
||||
|
||||
-- if typ ~= 'DATA' then
|
||||
-- return nil, 'bad row packet type: ' .. typ
|
||||
-- return nil, 'bad row packet type: ' .. typ
|
||||
-- end
|
||||
|
||||
-- typ == 'DATA'
|
||||
@@ -574,24 +584,24 @@ local function read_result(self, sock)
|
||||
end
|
||||
|
||||
local function _query_resp(self)
|
||||
return function(sock)
|
||||
local res, err, errno, sqlstate = read_result(self,sock)
|
||||
return function(sock)
|
||||
local res, err, errno, sqlstate = read_result(self, sock)
|
||||
if not res then
|
||||
local badresult ={}
|
||||
local badresult = {}
|
||||
badresult.badresult = true
|
||||
badresult.err = err
|
||||
badresult.errno = errno
|
||||
badresult.sqlstate = sqlstate
|
||||
return true , badresult
|
||||
return true, badresult
|
||||
end
|
||||
if err ~= "again" then
|
||||
return true, res
|
||||
end
|
||||
local multiresultset = {res}
|
||||
multiresultset.multiresultset = true
|
||||
local i =2
|
||||
while err =="again" do
|
||||
res, err, errno, sqlstate = read_result(self,sock)
|
||||
local i = 2
|
||||
while err == "again" do
|
||||
res, err, errno, sqlstate = read_result(self, sock)
|
||||
if not res then
|
||||
multiresultset.badresult = true
|
||||
multiresultset.err = err
|
||||
@@ -599,15 +609,15 @@ local function _query_resp(self)
|
||||
multiresultset.sqlstate = sqlstate
|
||||
return true, multiresultset
|
||||
end
|
||||
multiresultset[i]=res
|
||||
i=i+1
|
||||
multiresultset[i] = res
|
||||
i = i + 1
|
||||
end
|
||||
return true, multiresultset
|
||||
end
|
||||
end
|
||||
|
||||
function _M.connect(opts)
|
||||
local self = setmetatable( {}, mt)
|
||||
local self = setmetatable({}, mt)
|
||||
|
||||
local max_packet_size = opts.max_packet_size
|
||||
if not max_packet_size then
|
||||
@@ -620,11 +630,12 @@ function _M.connect(opts)
|
||||
local user = opts.user or ""
|
||||
local password = opts.password or ""
|
||||
|
||||
local channel = socketchannel.channel {
|
||||
local channel =
|
||||
socketchannel.channel {
|
||||
host = opts.host,
|
||||
port = opts.port or 3306,
|
||||
auth = _mysql_login(self,user,password,database,opts.on_connect),
|
||||
overload = opts.overload,
|
||||
auth = _mysql_login(self, user, password, database, opts.on_connect),
|
||||
overload = opts.overload
|
||||
}
|
||||
self.sockchannel = channel
|
||||
-- try connect first only once
|
||||
@@ -644,7 +655,7 @@ function _M.query(self, query)
|
||||
if not self.query_resp then
|
||||
self.query_resp = _query_resp(self)
|
||||
end
|
||||
return sockchannel:request( querypacket, self.query_resp )
|
||||
return sockchannel:request(querypacket, self.query_resp)
|
||||
end
|
||||
|
||||
local function read_prepare_result(self, sock)
|
||||
@@ -670,61 +681,61 @@ local function read_prepare_result(self, sock)
|
||||
if typ ~= "OK" then
|
||||
resp.badresult = true
|
||||
resp.errno = 300201
|
||||
resp.err = "first typ must be OK,now"..typ
|
||||
return false,resp
|
||||
resp.err = "first typ must be OK,now" .. typ
|
||||
return false, resp
|
||||
end
|
||||
resp.prepare_id,resp.field_count,resp.param_count,resp.warning_count = strunpack("<I4I2I2xI2",packet,2)
|
||||
resp.prepare_id, resp.field_count, resp.param_count, resp.warning_count = strunpack("<I4I2I2xI2", packet, 2)
|
||||
|
||||
resp.params = {}
|
||||
resp.fields = {}
|
||||
|
||||
if resp.param_count >0 then
|
||||
local param = _recv_field_packet(self,sock)
|
||||
if resp.param_count > 0 then
|
||||
local param = _recv_field_packet(self, sock)
|
||||
while param do
|
||||
table.insert(resp.params,param)
|
||||
param = _recv_field_packet(self,sock)
|
||||
table.insert(resp.params, param)
|
||||
param = _recv_field_packet(self, sock)
|
||||
end
|
||||
end
|
||||
if resp.field_count>0 then
|
||||
local field = _recv_field_packet(self,sock)
|
||||
if resp.field_count > 0 then
|
||||
local field = _recv_field_packet(self, sock)
|
||||
while field do
|
||||
table.insert(resp.fields,field)
|
||||
field = _recv_field_packet(self,sock)
|
||||
table.insert(resp.fields, field)
|
||||
field = _recv_field_packet(self, sock)
|
||||
end
|
||||
end
|
||||
|
||||
return true,resp
|
||||
return true, resp
|
||||
end
|
||||
|
||||
local function _prepare_resp(self,sql)
|
||||
return function(sock)
|
||||
return read_prepare_result(self,sock,sql)
|
||||
local function _prepare_resp(self, sql)
|
||||
return function(sock)
|
||||
return read_prepare_result(self, sock, sql)
|
||||
end
|
||||
end
|
||||
|
||||
-- 注册预处理语句
|
||||
function _M.prepare(self,sql)
|
||||
function _M.prepare(self, sql)
|
||||
local querypacket = _compose_stmt_prepare(self, sql)
|
||||
local sockchannel = self.sockchannel
|
||||
if not self.prepare_resp then
|
||||
self.prepare_resp = _prepare_resp(self)
|
||||
end
|
||||
return sockchannel:request( querypacket, self.prepare_resp )
|
||||
return sockchannel:request(querypacket, self.prepare_resp)
|
||||
end
|
||||
|
||||
local function _get_datetime(data,pos)
|
||||
local len,year,month,day,hour,minute,second
|
||||
local function _get_datetime(data, pos)
|
||||
local len, year, month, day, hour, minute, second
|
||||
local value
|
||||
len,pos = _from_length_coded_bin(data,pos)
|
||||
if len==7 then
|
||||
year,month,day,hour,minute,second,pos=string.unpack("<I2BBBBB",data,pos)
|
||||
value = strformat("%04d-%02d-%02d %02d:%02d:%02d",year,month,day,hour,minute,second)
|
||||
len, pos = _from_length_coded_bin(data, pos)
|
||||
if len == 7 then
|
||||
year, month, day, hour, minute, second, pos = string.unpack("<I2BBBBB", data, pos)
|
||||
value = strformat("%04d-%02d-%02d %02d:%02d:%02d", year, month, day, hour, minute, second)
|
||||
else
|
||||
value = "2017-09-09 20:08:09"
|
||||
--unsupported format
|
||||
pos=pos+len
|
||||
pos = pos + len
|
||||
end
|
||||
return value,pos
|
||||
return value, pos
|
||||
end
|
||||
|
||||
local _binary_parser = {
|
||||
@@ -743,31 +754,31 @@ local _binary_parser = {
|
||||
[0xfb] = _from_length_coded_str,
|
||||
[0xfc] = _from_length_coded_str,
|
||||
[0xfd] = _from_length_coded_str,
|
||||
[0xfe] = _from_length_coded_str,
|
||||
[0xfe] = _from_length_coded_str
|
||||
}
|
||||
|
||||
local function _parse_row_data_binary(data, cols, compact)
|
||||
local ncols = #cols
|
||||
-- 空位图,前两个bit系统保留 (列数量 + 7 + 2) / 8
|
||||
local null_count=(ncols+9)//8
|
||||
local pos = 2+null_count
|
||||
local null_count = (ncols + 9) // 8
|
||||
local pos = 2 + null_count
|
||||
local value
|
||||
|
||||
--空字段表
|
||||
local null_fields= {}
|
||||
local field_index=1
|
||||
local null_fields = {}
|
||||
local field_index = 1
|
||||
local byte
|
||||
for i=2,pos-1 do
|
||||
byte = strbyte(data,i)
|
||||
for j=0,7 do
|
||||
if field_index>2 then
|
||||
if byte&(1<<j)==0 then
|
||||
null_fields[field_index-2]=false
|
||||
for i = 2, pos - 1 do
|
||||
byte = strbyte(data, i)
|
||||
for j = 0, 7 do
|
||||
if field_index > 2 then
|
||||
if byte & (1 << j) == 0 then
|
||||
null_fields[field_index - 2] = false
|
||||
else
|
||||
null_fields[field_index-2]=true
|
||||
null_fields[field_index - 2] = true
|
||||
end
|
||||
end
|
||||
field_index=field_index+1
|
||||
field_index = field_index + 1
|
||||
end
|
||||
end
|
||||
|
||||
@@ -780,9 +791,9 @@ local function _parse_row_data_binary(data, cols, compact)
|
||||
if not null_fields[i] then
|
||||
parser = _binary_parser[typ]
|
||||
if not parser then
|
||||
error("_parse_row_data_binary()error,unsupported field type "..typ)
|
||||
error("_parse_row_data_binary()error,unsupported field type " .. typ)
|
||||
end
|
||||
value,pos = parser(data,pos)
|
||||
value, pos = parser(data, pos)
|
||||
if compact then
|
||||
row[i] = value
|
||||
else
|
||||
@@ -794,30 +805,30 @@ local function _parse_row_data_binary(data, cols, compact)
|
||||
return row
|
||||
end
|
||||
|
||||
local function read_execute_result(self,sock)
|
||||
local function read_execute_result(self, sock)
|
||||
local packet, typ, err = _recv_packet(self, sock)
|
||||
if not packet then
|
||||
return nil, err
|
||||
--error( err )
|
||||
--error( err )
|
||||
end
|
||||
|
||||
if typ == "ERR" then
|
||||
local errno, msg, sqlstate = _parse_err_packet(packet)
|
||||
return nil, msg, errno, sqlstate
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
end
|
||||
|
||||
if typ == 'OK' then
|
||||
if typ == "OK" then
|
||||
local res = _parse_ok_packet(packet)
|
||||
if res and res.server_status&SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
if res and res.server_status & SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
return res, "again"
|
||||
end
|
||||
return res
|
||||
end
|
||||
|
||||
if typ ~= 'DATA' then
|
||||
if typ ~= "DATA" then
|
||||
return nil, "packet type " .. typ .. " not supported"
|
||||
--error( "packet type " .. typ .. " not supported" )
|
||||
--error( "packet type " .. typ .. " not supported" )
|
||||
end
|
||||
|
||||
-- typ == 'DATA'
|
||||
@@ -835,13 +846,13 @@ local function read_execute_result(self,sock)
|
||||
col = _parse_field_packet(packet)
|
||||
if not col then
|
||||
break
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
--error( strformat("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
|
||||
end
|
||||
table.insert(cols,col)
|
||||
table.insert(cols, col)
|
||||
end
|
||||
|
||||
--没有记录集返回
|
||||
if #cols<1 then
|
||||
if #cols < 1 then
|
||||
return {}
|
||||
end
|
||||
|
||||
@@ -852,40 +863,40 @@ local function read_execute_result(self,sock)
|
||||
packet, typ, err = _recv_packet(self, sock)
|
||||
if typ == "EOF" then
|
||||
local warning_count, status_flags = _parse_eof_packet(packet)
|
||||
if status_flags&SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
if status_flags & SERVER_MORE_RESULTS_EXISTS ~= 0 then
|
||||
return rows, "again"
|
||||
end
|
||||
break
|
||||
end
|
||||
row = _parse_row_data_binary(packet,cols,compact)
|
||||
row = _parse_row_data_binary(packet, cols, compact)
|
||||
if not col then
|
||||
break
|
||||
end
|
||||
table.insert(rows,row)
|
||||
table.insert(rows, row)
|
||||
end
|
||||
|
||||
return rows
|
||||
end
|
||||
|
||||
local function _execute_resp(self)
|
||||
return function(sock)
|
||||
local res, err, errno, sqlstate = read_execute_result(self,sock)
|
||||
return function(sock)
|
||||
local res, err, errno, sqlstate = read_execute_result(self, sock)
|
||||
if not res then
|
||||
local badresult ={}
|
||||
local badresult = {}
|
||||
badresult.badresult = true
|
||||
badresult.err = err
|
||||
badresult.errno = errno
|
||||
badresult.sqlstate = sqlstate
|
||||
return true , badresult
|
||||
return true, badresult
|
||||
end
|
||||
if err ~= "again" then
|
||||
return true, res
|
||||
end
|
||||
local mulitresultset = {res}
|
||||
mulitresultset.mulitresultset = true
|
||||
local i =2
|
||||
while err =="again" do
|
||||
res, err, errno, sqlstate = read_execute_result(self,sock)
|
||||
local i = 2
|
||||
while err == "again" do
|
||||
res, err, errno, sqlstate = read_execute_result(self, sock)
|
||||
if not res then
|
||||
mulitresultset.badresult = true
|
||||
mulitresultset.err = err
|
||||
@@ -893,11 +904,11 @@ local function _execute_resp(self)
|
||||
mulitresultset.sqlstate = sqlstate
|
||||
return true, mulitresultset
|
||||
end
|
||||
mulitresultset[i]=res
|
||||
i=i+1
|
||||
mulitresultset[i] = res
|
||||
i = i + 1
|
||||
end
|
||||
return true, mulitresultset
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
--[[
|
||||
@@ -908,50 +919,50 @@ end
|
||||
sqlstate
|
||||
err
|
||||
]]
|
||||
function _M.execute(self, stmt,...)
|
||||
function _M.execute(self, stmt, ...)
|
||||
-- 检查参数,不能为nil
|
||||
local p_n = select('#',...)
|
||||
local p_n = select("#", ...)
|
||||
local p_v
|
||||
for i=1,p_n do
|
||||
p_v = select(i,...)
|
||||
if p_v==nil then
|
||||
for i = 1, p_n do
|
||||
p_v = select(i, ...)
|
||||
if p_v == nil then
|
||||
return {
|
||||
badresult=true,
|
||||
errno=30902,
|
||||
err="parameter "..i.." is nil",
|
||||
badresult = true,
|
||||
errno = 30902,
|
||||
err = "parameter " .. i .. " is nil"
|
||||
}
|
||||
end
|
||||
end
|
||||
|
||||
local querypacket,er = _compose_stmt_execute(self,stmt,CURSOR_TYPE_NO_CURSOR,{...})
|
||||
local querypacket, er = _compose_stmt_execute(self, stmt, CURSOR_TYPE_NO_CURSOR, {...})
|
||||
if not querypacket then
|
||||
return {
|
||||
badresult=true,
|
||||
errno=30902,
|
||||
err=er,
|
||||
badresult = true,
|
||||
errno = 30902,
|
||||
err = er
|
||||
}
|
||||
end
|
||||
local sockchannel = self.sockchannel
|
||||
if not self.execute_resp then
|
||||
self.execute_resp = _execute_resp(self)
|
||||
end
|
||||
return sockchannel:request( querypacket, self.execute_resp )
|
||||
return sockchannel:request(querypacket, self.execute_resp)
|
||||
end
|
||||
|
||||
function _M.ping(self)
|
||||
local querypacket,er = _compose_ping(self)
|
||||
local querypacket, er = _compose_ping(self)
|
||||
if not querypacket then
|
||||
return {
|
||||
badresult=true,
|
||||
errno=30902,
|
||||
err=er,
|
||||
badresult = true,
|
||||
errno = 30902,
|
||||
err = er
|
||||
}
|
||||
end
|
||||
local sockchannel = self.sockchannel
|
||||
if not self.query_resp then
|
||||
self.query_resp = _query_resp(self)
|
||||
end
|
||||
return sockchannel:request( querypacket, self.query_resp )
|
||||
return sockchannel:request(querypacket, self.query_resp)
|
||||
end
|
||||
|
||||
function _M.server_ver(self)
|
||||
@@ -959,19 +970,19 @@ function _M.server_ver(self)
|
||||
end
|
||||
|
||||
local escape_map = {
|
||||
['\0'] = "\\0",
|
||||
['\b'] = "\\b",
|
||||
['\n'] = "\\n",
|
||||
['\r'] = "\\r",
|
||||
['\t'] = "\\t",
|
||||
['\26'] = "\\Z",
|
||||
['\\'] = "\\\\",
|
||||
["\0"] = "\\0",
|
||||
["\b"] = "\\b",
|
||||
["\n"] = "\\n",
|
||||
["\r"] = "\\r",
|
||||
["\t"] = "\\t",
|
||||
["\26"] = "\\Z",
|
||||
["\\"] = "\\\\",
|
||||
["'"] = "\\'",
|
||||
['"'] = '\\"',
|
||||
['"'] = '\\"'
|
||||
}
|
||||
|
||||
function _M.quote_sql_str( str)
|
||||
return strformat("'%s'", strgsub(str, "[\0\b\n\r\t\26\\\'\"]", escape_map))
|
||||
function _M.quote_sql_str(str)
|
||||
return strformat("'%s'", strgsub(str, '[\0\b\n\r\t\26\\\'"]', escape_map))
|
||||
end
|
||||
|
||||
function _M.set_compact_arrays(self, value)
|
||||
|
||||
Reference in New Issue
Block a user