protocol.lua

--[[
  LICENSE: BSD
  Author: CandyMi[https://github.com/candymi]
]]

local bson = require "mongo.bson"
local bson_encode = bson.encode
local bson_decode = bson.decode
local bson_encode_order = bson.bson_encode_order

local crypt = require "crypt"
local md5 = crypt.md5
local sha1 = crypt.sha1
local xor_str = crypt.xor_str
local hmac_sha1 = crypt.hmac_sha1
local randomkey = crypt.randomkey_ex
local base64decode = crypt.base64decode
local base64encode = crypt.base64encode

local sys = require "sys"
local now = sys.now
local new_tab = sys.new_tab

local next = next
local type = type
local pcall = pcall
local assert = assert

-- local find = string.find
local fmt = string.format
local strpack = string.pack
local strunpack = string.unpack

local concat = table.concat
local unpack = table.unpack

local toint = math.tointeger

local MAX_INT32 = (1 << 31) - 1

local STR_TO_OPCODE = {
  OP_REPLY        = 1,	   -- Reply to a client request. responseTo is set.
  OP_UPDATE       =	2001,	 -- receivedUpdate	checkAuthForUpdate	update.
  OP_INSERT       =	2002,	 -- receivedInsert	checkAuthForInsert	createIndex/insert.
  RESERVED        = 2003,	 -- Formerly used for OP_GET_BY_OID.
  OP_QUERY        =	2004,  --	receivedQuery	checkAuthForQuery	find.
  OP_GET_MORE     = 2005,  --	receivedGetMore	checkAuthForGetMore	listCollections/listIndexes/find.
  OP_DELETE       =	2006,	 -- receivedDelete	checkAuthForDelete	remove
  OP_KILL_CURSORS =	2007,	 -- receivedKillCursors	checkAuthForKillCursors	killCursors
  OP_MSG          = 2013,	 -- Send a message using the format introduced in MongoDB 3.6.
}

local OPCODE_TO_STR = {
  [1]    = "OP_REPLY",
  [2001] = "OP_UPDATE",
  [2002] = "OP_INSERT",
  [2003] = "RESERVED",
  [2004] = "OP_QUERY",
  [2005] = "OP_GET_MORE",
  [2006] = "OP_DELETE",
  [2007] = "OP_KILL_CURSORS",
  [2013] = "OP_MSG",
}

local function sock_read(sock, bytes)
	local buffer = sock:recv(bytes)
	if not buffer then
		return
	end
	if #buffer == bytes then
		return buffer
	end
	bytes = bytes - #buffer
	local buffers = {buffer}
  local sock_recv = sock.recv
	while 1 do
		buffer = sock_recv(sock, bytes)
		if not buffer then
			return
		end
    bytes = bytes - #buffer
		buffers[#buffers+1] = buffer
		if bytes == 0 then
			return concat(buffers)
		end
	end
end

local function getn_nonce_payload(username)
  local nonce = base64encode(randomkey(16))
  return base64encode(concat({"n,,n=" .. username, ",r=" .. nonce})), nonce
end

local function salt_password(key, salt, count)
  salt =  base64decode(salt) .. '\0\0\0\1'
  local output = hmac_sha1(key, salt)
  local input = output
  for _ = 1, count - 1 do
    input = hmac_sha1(key, input)
    output = xor_str(output, input)
  end
  return output
end

---comment 读取公共头部
local function read_header(sock)
  -- print(1.1)
  local data = sock_read(sock, 16)
  if not data then
    return false, "[MONGO ERROR]: Server closed this session when client read header byte from socket."
  end
  -- print(1.2)
  local msg_len, req_id, resp_id, opcode = strunpack("<i4i4i4i4", data)
  return { msg_len = msg_len, req_id = req_id, resp_id = resp_id, opcode = opcode, opcode_str = OPCODE_TO_STR[opcode] }
end

---comment 读取MSG头部
local function read_msg_header(sock)
  local header, err = read_header(sock)
  if not header then
    return false, err
  end
  local data = sock_read(sock, 4)
  if not data then
    return false, "[MONGO ERROR]: Server closed this session when client read header byte from socket."
  end
  local flag = strunpack("<i4", data)
  header["flags"] = {
    checksum = flag & 0x01 == 0x01 and true,
    morecome = flag & 0x02 == 0x02 and true,
    allowed = flag & 0x010000 == 0x010000 and true,
  }
  return header
end

-- 读取OP_REPLY
local function read_reply(self)
  -- print(1)
  local sock = self.sock
  local header, err = read_header(sock)
  if not header then
    return false, err
  end
  -- print(2, header.req_id, header.resp_id ~= self.reqid)
  if header.req_id and header.resp_id ~= self.reqid then
    return false, "[MONGO ERROR]: Unexpected response request ID."
  end
  -- print(3)
  local buf = sock_read(sock, 20)
  if not buf then
    return false, "[MONGO ERROR]: Server closed this session when client read reply byte from socket."
  end
  -- print(4)
  header.flags, header.cuorsor, header.started, header.returned = strunpack("<i4i8i4i4", buf)
  local document = sock_read(sock, header.msg_len - 36)
  if not document then
    return false, "[MONGO ERROR]: Server closed this session when client read document byte from socket."
  end
  -- print(5)
  return header, bson_decode(document)
end

local function read_msg_body(self)
  -- print("开始读取")
  local sock = self.sock
  local header, err = read_msg_header(sock)
  if not header or header.opcode ~= STR_TO_OPCODE['OP_MSG'] then
    self.connected = false
    return false, err or "Invalid MSG TYPE."
  end
  if header.req_id and header.resp_id ~= self.reqid then
    return false, "[MONGO ERROR]: Unexpected response request ID."
  end
  -- print("读取完毕")
  -- var_dump(header)
  if header.msg_len - 20 > 1 then
    sock_read(sock, 1)
  end
  local document = sock_read(sock, header.msg_len - 21)
  if not document then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client read reply."
  end
  self.reqid = self.reqid % MAX_INT32 + 1
  local ok, info = pcall(bson_decode, document)
  if not ok then
    return false, info
  end
  return info
end

-- --------- QUERY --------- --
local function send_query(self, db, table, filter, option)
  local query = {{"find", table}, {"filter", filter}, {"limit", type(option) == 'table' and toint(option.limit) or 0}, {"skip", type(option) == 'table' and toint(option.skip) or 0}, {"$db", db}}
  if type(option) == 'table' then
    if toint(option.cursor) and toint(option.cursor) > 4294967296 then
      query = {{"getMore", toint(option.cursor)}, {"collection", table}, {"$db", db}}
      if toint(option.size) then
        query[#query+1] = {"batchSize", type(option) == 'table' and toint(option.size) or 0}
      end
    else
      -- 排序条件
      if type(option.sort) == 'table' then
        query[#query+1] = {"sort", option.sort}
      end
      -- 如果没有指定cursor但是指定了`size`, 数据如果超出`size`就会返回游标ID.
      if toint(option.size) and toint(option.size) > 0 then
        query[#query+1] = {"batchSize", type(option) == 'table' and toint(option.size) or 0}
      end
    end
    -- 限制返回字段
    if type(option.project) == 'table' then
      query[#query+1] = {"projection", option.project}
    end
  end
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function read_query(self)
  return read_msg_body(self)
end
-- --------- QUERY --------- --

-- --------- INSERT --------- --
local function send_insert(self, db, table, array, option)
  local documents = new_tab(#array, 0)
  for index, tab in ipairs(array) do
    documents[index] = assert(bson_encode(tab))
  end
  local query = {{"insert", table}, {"$db", db}}
  if option then
    if option.ordered then
      query[#query+1] = {"ordered", true}
    end
    if self.have_transaction and option.concern then
      query[#query+1] = {"writeConcern", option.concern}
    end
  end
  local section1 = bson_encode_order(unpack(query))
  local section2 = concat(documents)
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 10 + 4 + #section2, "documents"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function read_insert(self)
  return read_msg_body(self)
end
-- --------- INSERT --------- --

-- --------- UPDATE --------- --
local function send_update(self, db, table, filter, update, option)
  local section1 = bson_encode_order({"update", table}, {"$db", db}, {"ordered", true})
  local section2 = bson_encode({q = filter, u = update, upsert = type(option) == 'table' and option.upsert and true or false, multi = type(option) == 'table' and option.multi and true or false} )
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 8 + 4 + #section2, "updates"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function read_update(self)
  return read_msg_body(self)
end
-- --------- UPDATE --------- --

-- --------- DELETE --------- --
local function send_delete(self, db, table, filter, option)
  local section1 = bson_encode_order({"delete", table}, {"$db", db}, {"ordered", true})
  local section2 = bson_encode({q = filter, limit = type(option) == 'table' and toint(option.one) == 1 and 1 or 0 })
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 8 + 4 + #section2, "deletes"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function read_delete(self)
  return read_msg_body(self)
end
-- --------- DELETE --------- --

-- --------- AGGREGATE --------- --
local function send_aggregate(self, db, table, pipeline, option)
  local query = {{"aggregate", table}, {"cursor", bson.empty_table()}, {"pipeline", pipeline}, {"$db", db}}
  if type(option) == 'table' and toint(option.cursor) and toint(option.cursor) > 4294967296 then
    query = {{"getMore", toint(option.cursor)}, {"collection", table}, {"$db", db}}
    if toint(option.size) and toint(option.size) > 0 then
      query[#query+1] = {"batchSize", toint(option.size)}
    end
  end
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function read_aggregate(self)
  return read_msg_body(self)
end
-- --------- AGGREGATE --------- --

-- --------- MapReduce --------- --
local function send_mapreduce(self, db, table, map, reduce, option)
  option = type(option) == 'table' and option or {}
  local query = {
    {"mapReduce", table},
    {"map", bson.dump(map)},
    {"reduce", bson.dump(reduce)},
    {"out", option.out or { inline = 1.0 }},
  }
  if option.sort then
    query[#query+1] = {"sort", option.sort }
  end
  if option.limit then
    query[#query+1] = {"limit", option.limit }
  end
  query[#query+1] = {"query", type(option.out) == 'table' and option.out or bson.empty_table()}
  query[#query+1] = {"$db", db}
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function read_mapreduce(self)
  return read_msg_body(self)
end
-- --------- MapReduce --------- --

-- --------- Indexed --------- --
local function send_createindex(self, db, table, indexes, option)
  local index_name = {}
  local keys = {}
  for name, value in pairs(indexes) do
    index_name[#index_name+1] = name .. "_" .. value
    keys[name] = value
  end
  local index_array = {
    {
      name = option.name and option.name or ("_" .. concat(index_name, "_") .. "_"),
      key = keys,
      unique = option.unique and 1 or nil,
      sparse = option.sparse and 1 or false,
      background = option.background and 1 or false,
    }
  }
  local query = {{"createIndexes", table}, {"indexes", index_array}, {"$db", db}}
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function send_getindexes(self, db, table)
  local query = {{"listIndexes", table}, {"cursor", bson.empty_table()}, {"$db", db}}
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function send_dropindexes(self, db, table, indexname)
  local query = {{"dropIndexes", table}, {"index", indexname}, {"$db", db}}
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function read_index(self)
  return read_msg_body(self)
end
-- --------- Indexed --------- --

-- --------- GridFs --------- --
local function send_gridfs_files_list(self, db, table, filter, id)
  local query = {{"find", table .. ".files"}, {"filter", filter}, {"$db", db}}
  if toint(id) and toint(id) > 0 then
    query = {{"getMore", toint(id)}, {"collection", table .. ".files"}, {"$db", db}}
  end
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function send_gridfs_chunks_list(self, db, table, filter, id)
  local query = {{"find", table .. ".chunks"}, {"filter", filter}, {"$db", db}}
  if toint(id) and toint(id) > 0 then
    query = {{"getMore", toint(id)}, {"collection", table .. ".chunks"}, {"$db", db}}
  end
  local sections = bson_encode_order(unpack(query))
  return self.sock:send(strpack("<i4i4i4i4i4B", #sections + 21, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0, 0)) and self.sock:send(sections)
end

local function send_gridfs_files_delete(self, db, table, filter, limit)
  local section1 = bson_encode_order({"delete", table .. ".files"}, {"$db", db}, {"ordered", true})
  local section2 = bson_encode_order({"q", filter}, {"limit", toint(limit) and toint(limit) > 0 and toint(limit) or 0})
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 8 + 4 + #section2, "deletes"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function send_gridfs_chunks_delete(self, db, table, filter, limit)
  local section1 = bson_encode_order({"delete", table .. ".chunks"}, {"$db", db}, {"ordered", true})
  local section2 = bson_encode_order({"q", filter}, {"limit", toint(limit) and toint(limit) > 0 and toint(limit) or 0})
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 8 + 4 + #section2, "deletes"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function send_gridfs_files_upload(self, db, table, oid, filename, filelength, md5sum, meta)
  local section1 = bson_encode_order({"insert", table .. ".files"}, {"ordered", true}, {"writeConcern", {w = "majority"}}, {"$db", db})
  local section2 = bson_encode_order({"_id", oid}, {"length", filelength}, {"chunkSize", 261120}, {"uploadDate", now() * 1e3 // 1}, {"md5", md5sum}, {"filename", filename}, {"metadata", meta or {}})
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 10 + 4 + #section2, "documents"), section2 }
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function send_gridfs_chunks_upload(self, db, table, fid, file)
  local section1 = bson_encode_order({"insert", table .. ".chunks"}, {"ordered", true}, {"writeConcern", {w = "majority"}}, {"$db", db})
  local list = new_tab(128, 0)
  local len = #file
  local s, e = 1, 261120
  while true do
    list[#list+1] = { file = file:sub(s, e)}
    if e >= len then
      break
    end
    s = e + 1
    e = s + 261120
  end
  local sections2 = new_tab(128, 0)
  for index, item in ipairs(list) do
    sections2[#sections2+1] = bson_encode_order({"_id", bson.objectid()}, {"files_id", bson.objectid(fid)}, {"n", index - 1}, {"data", bson.binary(item.file)})
  end
  local section2 = concat(sections2)
  local sections = concat{strpack("<B", 0), section1, strpack("<Bi4z", 1, 10 + 4 + #section2, "documents"), section2}
  return self.sock:send(strpack("<i4i4i4i4i4", 20 + #sections, self.reqid, 0, STR_TO_OPCODE["OP_MSG"], 0)) and self.sock:send(sections)
end

local function read_gridfs(self)
  return read_msg_body(self)
end
-- --------- Indexed --------- --

-- --------- HANDSHAKE --------- --
local function send_handshake(self)
  local query = assert(bson_encode_order(table.unpack({{ "isMaster", true }})))
  return self.sock:send(strpack("<i4i4i4i4i4zi4i4", #query + 39, self.reqid, 0, STR_TO_OPCODE["OP_QUERY"], 0x04, 'admin.$cmd', 0, -1)) and self.sock:send(query)
end
-- --------- HANDSHAKE --------- --

-- --------- AUTH --------- --
local function send_auth(self)
  local nonce, _ --[[ non ]] = getn_nonce_payload(self.username)
  local query = assert(bson_encode_order({"saslStart", 1}, {"autoAuthorize", 1}, {"mechanism", self.auth_mode}, {"payload", nonce}))
  local _ = self.sock:send(strpack("<i4i4i4i4i4zi4i4", #query + 29 + #(self.db ..'.$cmd'), self.reqid, 0, STR_TO_OPCODE["OP_QUERY"], 0x04, self.db ..'.$cmd', 0, -1)) and self.sock:send(query)
  local header, response, err = read_reply(self)
  if not header or response.ok ~= 1 then
    return false, err or response.errmsg or "[MONGO ERROR]: Unknown Error."
  end
  if not response['conversationId'] then
    return false, "Invalid conversationId."
  end
  local payload = response["payload"]
  if not payload then
    return false, "Invalid payload."
  end
  payload = base64decode(payload)
  -- var_dump(header); var_dump(response);
  local p = { r = nil, s = nil, i = nil }
  for key, value in payload:gmatch("([^,=]+)=([^,]+)") do
    p[key] = value
  end
  -- 检查启动协议是否匹配.
  -- if not find(p['r'] or "", '^' .. non) then
  --   return false, "Invalid nonce when server return data."
  -- end
  local without_proof = "c=biws,r=" .. p['r']
  local salted_password = salt_password(md5(fmt("%s:mongo:%s", self.username, self.password), true), p['s'], toint(p['i']))
  local client_key = hmac_sha1(salted_password, "Client Key")
  local auth_msg = concat({base64decode(nonce):sub(4), payload, without_proof}, ",")

  query = assert(bson_encode_order({"saslContinue", 1}, {"conversationId", response['conversationId']}, {"payload", base64encode(without_proof .. ',' .. "p=" .. base64encode(xor_str(client_key, hmac_sha1(sha1(client_key), auth_msg))))}))
  local _ = self.sock:send(strpack("<i4i4i4i4i4zi4i4", #query + 29 + #(self.db ..'.$cmd'), self.reqid, 0, STR_TO_OPCODE["OP_QUERY"], 0x04, self.db ..'.$cmd', 0, -1)) and self.sock:send(query)
  header, response, err = read_reply(self)
  if not header or response.ok ~= 1 then
    return false, err or response.errmsg or "[MONGO ERROR]: Unknown Error."
  end
  p = { v = nil}
  response["payload"] = base64decode(response["payload"])
  for key, value in string.gmatch(response["payload"], "([^,=]+)=([^,]+)") do
    p[key] = value
  end
  if p['v'] ~= base64encode(hmac_sha1(hmac_sha1(salted_password, "Server Key"), auth_msg)) then
    return false, "Server returned an invalid signature."
  end
  if not response.done then
    query = assert(bson_encode_order({"saslContinue", 1}, {"conversationId", response['conversationId']}, {"payload", ""}))
    local _ = self.sock:send(strpack("<i4i4i4i4i4zi4i4", #query + 29 + #(self.db ..'.$cmd'), self.reqid, 0, STR_TO_OPCODE["OP_QUERY"], 0x04, self.db ..'.$cmd', 0, -1)) and self.sock:send(query)
    header, response, err = read_reply(self)
    if not header or response.ok ~= 1 or not response.done then
      return false, err or response.errmsg or "[MONGO ERROR]: Authorization failed."
    end
  end
  return true
end
-- --------- AUTH --------- --

local protocol = { __Version__ = 0.1 }

---comment 发送授权认证
function protocol.request_auth(self)
  if type(self.username) == 'string' and type(self.password) == 'string' then
    local tab, err = send_auth(self)
    if not tab then
      return false, err
    end
  end
  return true
end

---comment 心跳与握手消息
function protocol.request_handshake(self)
  local ok = send_handshake(self)
  if not ok then
    return false, "[MONGO ERROR]: Server closed this session when client send hello request."
  end
  local header, response, err = read_reply(self)
  if not header then
    return false, "[MONGO ERROR]: Server connected refuse."
  end
  if response.ok ~= 1 or response.maxWireVersion < 6 then
    -- 如果探测到MongoDB版本低于3.6则当做连接失败
    return false, err or response.errmsg or "[MONGO ERROR]: Version below 3.6 are not supported."
  end
  -- var_dump(header); var_dump(response);
  self.have_transaction = response.maxWireVersion >= 7 -- 是否支持事务
  return response
end

---comment 查询语句
function protocol.request_query(self, db, table, filter, option)
  if not send_query(self, db, table, filter, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send query request."
  end
  return read_query(self)
end

---comment 插入语句
function protocol.request_insert(self, db, table, array, option)
  if not send_insert(self, db, table, array, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send insert request."
  end
  return read_insert(self)
end

---comment 更新语句
function protocol.request_update(self, db, table, filter, update, option)
  if not send_update(self, db, table, filter, update, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send update request."
  end
  return read_update(self)
end

---comment 删除语句
function protocol.request_delete(self, db, table, array, option)
  if not send_delete(self, db, table, array, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send delete request."
  end
  return read_delete(self)
end

---comment 统计查询
function protocol.request_count(self, db, table, filter, option)
  if not send_aggregate(self, db, table, { { ["$match"] = type(filter) == 'table' and filter or {} }, {["$group"] = { _id = bson.null(), n = {['$sum']  = 1 } }} }, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send count request."
  end
  return read_aggregate(self)
end

---comment 聚合查询
function protocol.request_aggregate(self, db, table, filter, option)
  if type(filter) ~= 'table' or not next(filter) then
    filter = bson.empty_array()
  elseif #filter < 1 then
    filter = { filter }
  end
  if not send_aggregate(self, db, table, filter, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send aggregate request."
  end
  return read_aggregate(self)
end

---comment 分布式计算
function protocol.request_mapreduce(self, db, table, map, reduce, option)
  if not send_mapreduce(self, db, table, map, reduce, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send mapreduce request."
  end
  return read_mapreduce(self)
end

---comment 创建索引
function protocol.request_createindex(self, db, table, indexs, option)
  if not send_createindex(self, db, table, indexs, option) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send createIndex request."
  end
  return read_index(self)
end

---comment 获取索引
function protocol.request_getindexes(self, db, table)
  if not send_getindexes(self, db, table) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send getIndexes request."
  end
  return read_index(self)
end

---comment 获取索引
function protocol.request_dropindexes(self, db, table, indexname)
  if not send_dropindexes(self, db, table, indexname) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send dropIndex request."
  end
  return read_index(self)
end

---comment 获取GridFS Files
function protocol.request_gridfs_files_list(self, db, table, filter, id)
  if not send_gridfs_files_list(self, db, table, filter, id) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

---comment 获取GridFS Chunks
function protocol.request_gridfs_chunks_list(self, db, table, filter, id)
  if not send_gridfs_chunks_list(self, db, table, filter, id) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

---comment 删除GridFS Files
function protocol.request_gridfs_files_delete(self, db, table, filter, limit)
  if not send_gridfs_files_delete(self, db, table, filter, limit) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

---comment 删除GridFS Chunks
function protocol.request_gridfs_chunks_delete(self, db, table, filter, limit)
  if not send_gridfs_chunks_delete(self, db, table, filter, limit) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

function protocol.request_gridfs_files_upload(self, db, table, oid, filename, file, md5sum, meta)
  if not send_gridfs_files_upload(self, db, table, oid, filename, file, md5sum, meta) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

function protocol.request_gridfs_chunks_upload(self, db, table, fid, file)
  if not send_gridfs_chunks_upload(self, db, table, fid, file) then
    self.connected = false
    return false, "[MONGO ERROR]: Server closed this session when client send GridFS request."
  end
  return read_gridfs(self)
end

return protocol