diff --git a/lualib/skynet/db/mongo.lua b/lualib/skynet/db/mongo.lua index ac8739da..0c58485d 100644 --- a/lualib/skynet/db/mongo.lua +++ b/lualib/skynet/db/mongo.lua @@ -88,6 +88,38 @@ local function __parse_addr(addr) return host, tonumber(port) end +local function werror(r) + local ok = (r.ok == 1 and not r.writeErrors and not r.writeConcernError and not r.errmsg) + + local err + if not ok then + if r.writeErrors then + err = r.writeErrors[1].errmsg + elseif r.writeConcernError then + err = r.writeConcernError.errmsg + else + err = r.errmsg + end + end + return ok, err, r +end + +local function wbulkerror(r) + local ok, err = werror(r) + local cursor + if (r.nErrors and r.nErrors ~= 0) and r.cursor then + cursor = {} + err = err and err .. ";" or "" + for i, v in ipairs(r.cursor.firstBatch) do + if v.ok ~= 1 then + cursor[#cursor+1] = i + err = err .. v.errmsg .. ";" + end + end + end + return ok, err, r, cursor +end + local auth_method = {} local function mongo_auth(mongoc) @@ -197,6 +229,61 @@ function mongo_client:runCommand(...) return self.admin:runCommand(...) end +local function filter(tbl, key, value) + if value == nil then return end + tbl[#tbl+1] = key + tbl[#tbl+1] = type(value) == "table" and bson_encode(value) or value +end + +local bulkWrite = {} +bulkWrite.insert = function(nsindex, doc) + return bson_encode_order("insert", nsindex, "document", bson_encode(doc.insert)) +end + +bulkWrite.delete = function(nsindex, doc) + local args = {"delete", nsindex, "filter", bson_encode(doc.filter), "multi", doc.multi or false} + filter(args, "hit", doc.hit) + filter(args, "collation", doc.collation) + return bson_encode_order(table.unpack(args)) +end + +--update仅支持原子操作更新/聚合管道更新 详情阅读官方文档 +--arrayFilters 过滤方式需要遍历encode 暂时不支持 +bulkWrite.update = function(nsindex, doc) + local args = {"update", nsindex, "filter", bson_encode(doc.filter), "updateMods", bson_encode(doc.update), + "multi", doc.multi or false, "upsert", doc.upsert or false} + filter(args, "hit", doc.hit) + filter(args, "constants", doc.constants) + filter(args, "collation", doc.collation) + return bson_encode_order(table.unpack(args)) +end + +--bulkWrite 依赖 mongo8.0x +--忽略let,cursor,writeConcern 参数 +---@param datas table 待写入数据 +--- - op string 写入数据方式 参见:bulkWrite +--- - collection string 数据写入目标, 用于构建nsInfo; 注意:采用 空间.集合方式 +--- - ... 操作具体参数 insert采用:insert填充document, update采用 update 填充updateMods +--- 示例:{op = "insert", collection = "log.online", insert = {count=0}} +--- {op = "update", collection = "log.online", update = {["$set"] = {count = 2}}, upsert = true}} +function mongo_client:bulkWrite(datas, comment, ordered, verify, errorsOnly) + local ops, ns, map = {}, {}, {} + for i, v in ipairs(datas) do + if not map[v.collection] then + map[v.collection] = #ns --ops 索引从0开始 + ns[#ns+1] = bson_encode({ns = v.collection}) + end + local f = assert(bulkWrite[v.op], v.op) + ops[i] = f(map[v.collection], v) + end + local args = {"bulkWrite", 1, "ops", ops, "nsInfo", ns} + filter(args, "ordered", ordered) --是否有序执行 默认有序 + filter(args, "bypassDocumentValidation", verify) --是否验证 默认验证 + filter(args, "comment", comment) --日志跟踪注释 + filter(args, "errorsOnly",errorsOnly) -- 仅返回错误信息 + return wbulkerror(self:runCommand(table.unpack(args))) +end + function auth_method:auth_mongodb_cr(user,password) local password = md5.sumhexa(string.format("%s:mongo:%s",user,password)) local result= self:runCommand "getnonce" @@ -355,22 +442,6 @@ function mongo_collection:insert(doc) self.database:send_command("insert", self.name, "documents", {bson_encode(doc)}) end -local function werror(r) - local ok = (r.ok == 1 and not r.writeErrors and not r.writeConcernError and not r.errmsg) - - local err - if not ok then - if r.writeErrors then - err = r.writeErrors[1].errmsg - elseif r.writeConcernError then - err = r.writeConcernError.errmsg - else - err = r.errmsg - end - end - return ok, err, r -end - function mongo_collection:safe_insert(doc) local r = self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)}) return werror(r)