diff --git a/lualib-src/lua-mongo.c b/lualib-src/lua-mongo.c index 1f6c6b69..1db07e5d 100644 --- a/lualib-src/lua-mongo.c +++ b/lualib-src/lua-mongo.c @@ -10,21 +10,17 @@ #include #include -#define OP_REPLY 1 -#define OP_MSG 1000 -#define OP_UPDATE 2001 -#define OP_INSERT 2002 -#define OP_QUERY 2004 -#define OP_GET_MORE 2005 -#define OP_DELETE 2006 -#define OP_KILL_CURSORS 2007 +#define OP_COMPRESSED 2012 +#define OP_MSG 2013 + +typedef enum { + MSG_CHECKSUM_PRESENT = 1 << 0, + MSG_MORE_TO_COME = 1 << 1, + MSG_EXHAUST_ALLOWED = 1 << 16, +} msg_flags_t; -#define REPLY_CURSORNOTFOUND 1 -#define REPLY_QUERYFAILURE 2 -#define REPLY_AWAITCAPABLE 8 // ignore because mongo 1.6+ always set it #define DEFAULT_CAP 128 - struct connection { int sock; int id; @@ -106,6 +102,13 @@ write_int32(struct buffer *b, int32_t v) { b->ptr[b->size++] = (uv >> 24)&0xff; } +static inline void +write_int8(struct buffer *b, int8_t v) { + uint8_t uv = (uint8_t)v; + buffer_reserve(b, 1); + b->ptr[b->size++] = uv; +} + static inline void write_bytes(struct buffer *b, const void * buf, int sz) { buffer_reserve(b,sz); @@ -113,6 +116,7 @@ write_bytes(struct buffer *b, const void * buf, int sz) { b->size += sz; } +/* static void write_string(struct buffer *b, const char *key, size_t sz) { buffer_reserve(b,sz+1); @@ -120,6 +124,7 @@ write_string(struct buffer *b, const char *key, size_t sz) { b->ptr[b->size+sz] = '\0'; b->size+=sz+1; } +*/ static inline int reserve_length(struct buffer *b) { @@ -138,389 +143,69 @@ write_length(struct buffer *b, int32_t v, int off) { b->ptr[off++] = (uv >> 24)&0xff; } -// 1 integer id -// 2 integer flags -// 3 string collection name -// 4 integer skip -// 5 integer return number -// 6 document query -// 7 document selector (optional) -// return string package -static int -op_query(lua_State *L) { - int id = luaL_checkinteger(L,1); - document query = lua_touserdata(L,6); - if (query == NULL) { - return luaL_error(L, "require query document"); - } - document selector = lua_touserdata(L,7); - int flags = luaL_checkinteger(L, 2); - size_t nsz = 0; - const char *name = luaL_checklstring(L,3,&nsz); - int skip = luaL_checkinteger(L, 4); - int number = luaL_checkinteger(L, 5); +struct header_t { + //int32_t message_length; // total message size, include this + int32_t request_id; // identifier for this message + int32_t response_to; // requestID from the original request(used in responses from the database) + int32_t opcode; // message type - luaL_Buffer b; - luaL_buffinit(L,&b); - - struct buffer buf; - buffer_create(&buf); - int len = reserve_length(&buf); - write_int32(&buf, id); - write_int32(&buf, 0); - write_int32(&buf, OP_QUERY); - write_int32(&buf, flags); - write_string(&buf, name, nsz); - write_int32(&buf, skip); - write_int32(&buf, number); - - int32_t query_len = get_length(query); - int total = buf.size + query_len; - int32_t selector_len = 0; - if (selector) { - selector_len = get_length(selector); - total += selector_len; - } - - write_length(&buf, total, len); - luaL_addlstring(&b, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - luaL_addlstring(&b, (const char *)query, query_len); - - if (selector) { - luaL_addlstring(&b, (const char *)selector, selector_len); - } - - luaL_pushresult(&b); - - return 1; -} + int32_t flags; +}; // 1 string data // 2 result document table // return boolean succ (false -> request id, error document) // number request_id // document first -// string cursor_id -// integer startfrom static int -op_reply(lua_State *L) { +unpack_reply(lua_State *L) { size_t data_len = 0; const char * data = luaL_checklstring(L,1,&data_len); - struct reply_type { -// int32_t length; // total message size, including this - int32_t request_id; // identifier for this message - int32_t response_id; // requestID from the original request - // (used in reponses from db) - int32_t opcode; // request type - int32_t flags; - int32_t cursor_id[2]; - int32_t starting; - int32_t number; - }; - const struct reply_type* reply = (const struct reply_type*)data; + const struct header_t* h = (const struct header_t*)data; - if (data_len < sizeof(*reply)) { + if (data_len < sizeof(*h)) { lua_pushboolean(L, 0); return 1; } - int id = little_endian(reply->response_id); - int flags = little_endian(reply->flags); - if (flags & REPLY_QUERYFAILURE) { - lua_pushboolean(L,0); - lua_pushinteger(L, id); - lua_pushlightuserdata(L, (void *)(reply+1)); - return 3; + int opcode = little_endian(h->opcode); + if (opcode != OP_MSG) { + return luaL_error(L, "Unsupported opcode:%d", opcode); } - int starting_from = little_endian(reply->starting); - int number = little_endian(reply->number); - int sz = (int)data_len - sizeof(*reply); - const uint8_t * doc = (const uint8_t *)(reply+1); + int id = little_endian(h->response_to); + int flags = little_endian(h->flags); - if (lua_istable(L,2)) { - int i = 1; - while (sz > 4) { - lua_pushlightuserdata(L, (void *)doc); - lua_rawseti(L, 2, i); - - int32_t doc_len = get_length((document)doc); - if (doc_len <= 0) { - return luaL_error(L, "Invalid result bson document"); - } - - doc += doc_len; - sz -= doc_len; - - ++i; + if (flags != 0) { + if ((flags & MSG_CHECKSUM_PRESENT) != 0) { + return luaL_error(L, "Unsupported OP_MSG flag checksumPresent"); } - if (i != number + 1) { - lua_pushboolean(L,0); - lua_pushinteger(L, id); - return 2; - } - int c = lua_rawlen(L, 2); - for (;i<=c;i++) { - lua_pushnil(L); - lua_rawseti(L, 2, i); - } - } else { - if (sz >= 4) { - sz -= get_length((document)doc); + + if ((flags ^ MSG_MORE_TO_COME) != 0) { + return luaL_error(L, "Unsupported OP_MSG flag:%d", flags); } } - if (sz != 0) { - return luaL_error(L, "Invalid result bson document"); + + int sz = (int)data_len - sizeof(*h); + + const uint8_t * section = (const uint8_t *)(h+1); + + uint8_t payload_type = *section; + const uint8_t * doc = section+1; + + if (payload_type != 0) { + return luaL_error(L, "Unsupported OP_MSG payload type: %d", payload_type); } - lua_pushboolean(L,1); + + int32_t doc_sz = get_length((document)(doc)); + if ((sz - 1) != doc_sz) { + return luaL_error(L, "Unsupported OP_MSG reply: >1 section"); + } + + lua_pushboolean(L, 1); lua_pushinteger(L, id); - if (number == 0) - lua_pushnil(L); - else - lua_pushlightuserdata(L, (void *)(reply+1)); - if (reply->cursor_id[0] == 0 && reply->cursor_id[1]==0) { - // closed cursor - lua_pushnil(L); - } else { - lua_pushlstring(L, (const char *)(reply->cursor_id), 8); - } - lua_pushinteger(L, starting_from); - - return 5; -} - -/* - 1 string cursor_id - return string package - */ -static int -op_kill(lua_State *L) { - size_t cursor_len = 0; - const char * cursor_id = luaL_tolstring(L, 1, &cursor_len); - if (cursor_len != 8) { - return luaL_error(L, "Invalid cursor id"); - } - - struct buffer buf; - buffer_create(&buf); - - int len = reserve_length(&buf); - write_int32(&buf, 0); - write_int32(&buf, 0); - write_int32(&buf, OP_KILL_CURSORS); - - write_int32(&buf, 0); - write_int32(&buf, 1); - write_bytes(&buf, cursor_id, 8); - - write_length(&buf, buf.size, len); - - lua_pushlstring(L, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - return 1; -} - -/* - 1 string collection - 2 integer single remove - 3 document selector - - return string package - */ -static int -op_delete(lua_State *L) { - document selector = lua_touserdata(L,3); - if (selector == NULL) { - luaL_error(L, "Invalid param"); - } - size_t sz = 0; - const char * name = luaL_checklstring(L,1,&sz); - - luaL_Buffer b; - luaL_buffinit(L,&b); - - struct buffer buf; - buffer_create(&buf); - int len = reserve_length(&buf); - write_int32(&buf, 0); - write_int32(&buf, 0); - write_int32(&buf, OP_DELETE); - write_int32(&buf, 0); - write_string(&buf, name, sz); - write_int32(&buf, lua_tointeger(L,2)); - - int32_t selector_len = get_length(selector); - int total = buf.size + selector_len; - write_length(&buf, total, len); - - luaL_addlstring(&b, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - luaL_addlstring(&b, (const char *)selector, selector_len); - luaL_pushresult(&b); - - return 1; -} - -/* - 1 integer id - 2 string collection - 3 integer number - 4 cursor_id (8 bytes string/ 64bit) - - return string package - */ -static int -op_get_more(lua_State *L) { - int id = luaL_checkinteger(L, 1); - size_t sz = 0; - const char * name = luaL_checklstring(L,2,&sz); - int number = luaL_checkinteger(L, 3); - size_t cursor_len = 0; - const char * cursor_id = luaL_tolstring(L, 4, &cursor_len); - if (cursor_len != 8) { - return luaL_error(L, "Invalid cursor id"); - } - - struct buffer buf; - buffer_create(&buf); - int len = reserve_length(&buf); - write_int32(&buf, id); - write_int32(&buf, 0); - write_int32(&buf, OP_GET_MORE); - write_int32(&buf, 0); - write_string(&buf, name, sz); - write_int32(&buf, number); - write_bytes(&buf, cursor_id, 8); - write_length(&buf, buf.size, len); - - lua_pushlstring(L, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - return 1; -} - -// 1 string collection -// 2 integer flags -// 3 document selector -// 4 document update -// return string package -static int -op_update(lua_State *L) { - document selector = lua_touserdata(L,3); - document update = lua_touserdata(L,4); - if (selector == NULL || update == NULL) { - luaL_error(L, "Invalid param"); - } - size_t sz = 0; - const char * name = luaL_checklstring(L,1,&sz); - - luaL_Buffer b; - luaL_buffinit(L, &b); - - struct buffer buf; - buffer_create(&buf); - // make package header, don't raise L error - int len = reserve_length(&buf); - write_int32(&buf, 0); - write_int32(&buf, 0); - write_int32(&buf, OP_UPDATE); - write_int32(&buf, 0); - write_string(&buf, name, sz); - write_int32(&buf, lua_tointeger(L,2)); - - int32_t selector_len = get_length(selector); - int32_t update_len = get_length(update); - - int total = buf.size + selector_len + update_len; - write_length(&buf, total, len); - - luaL_addlstring(&b, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - luaL_addlstring(&b, (const char *)selector, selector_len); - luaL_addlstring(&b, (const char *)update, update_len); - - luaL_pushresult(&b); - - return 1; -} - -static int -document_length(lua_State *L) { - if (lua_isuserdata(L, 3)) { - document doc = lua_touserdata(L,3); - return get_length(doc); - } - if (lua_istable(L,3)) { - int total = 0; - int s = lua_rawlen(L,3); - int i; - for (i=1;i<=s;i++) { - lua_rawgeti(L, 3, i); - document doc = lua_touserdata(L,-1); - if (doc == NULL) { - lua_pop(L,1); - return luaL_error(L, "Invalid document at %d", i); - } else { - total += get_length(doc); - lua_pop(L,1); - } - } - return total; - } - return luaL_error(L, "Insert need documents"); -} - -// 1 integer flags -// 2 string collection -// 3 documents -// return string package -static int -op_insert(lua_State *L) { - size_t sz = 0; - const char * name = luaL_checklstring(L,2,&sz); - int dsz = document_length(L); - - luaL_Buffer b; - luaL_buffinit(L, &b); - - struct buffer buf; - buffer_create(&buf); - // make package header, don't raise L error - int len = reserve_length(&buf); - write_int32(&buf, 0); - write_int32(&buf, 0); - write_int32(&buf, OP_INSERT); - write_int32(&buf, lua_tointeger(L,1)); - write_string(&buf, name, sz); - - int total = buf.size + dsz; - write_length(&buf, total, len); - - luaL_addlstring(&b, (const char *)buf.ptr, buf.size); - buffer_destroy(&buf); - - if (lua_isuserdata(L,3)) { - document doc = lua_touserdata(L,3); - luaL_addlstring(&b, (const char *)doc, get_length(doc)); - } else { - int s = lua_rawlen(L, 3); - int i; - for (i=1;i<=s;i++) { - lua_rawgeti(L,3,i); - document doc = lua_touserdata(L,-1); - lua_pop(L,1); // must call lua_pop before luaL_addlstring, because addlstring may change stack top - luaL_addlstring(&b, (const char *)doc, get_length(doc)); - } - } - - luaL_pushresult(&b); - - return 1; + lua_pushlightuserdata(L, (void *)(doc)); + return 3; } // string 4 bytes length @@ -535,21 +220,55 @@ reply_length(lua_State *L) { return 1; } +// @param 1 request_id int +// @param 2 flags int +// @param 3 command bson document +// @return +static int +op_msg(lua_State *L) { + int id = luaL_checkinteger(L, 1); + int flags = luaL_checkinteger(L, 2); + document cmd = lua_touserdata(L, 3); + + if (cmd == NULL) { + return luaL_error(L, "opmsg require cmd document"); + } + + luaL_Buffer b; + luaL_buffinit(L, &b); + + struct buffer buf; + buffer_create(&buf); + int len = reserve_length(&buf); + write_int32(&buf, id); + write_int32(&buf, 0); + write_int32(&buf, OP_MSG); + write_int32(&buf, flags); + write_int8(&buf, 0); + + int32_t cmd_len = get_length(cmd); + int total = buf.size + cmd_len; + + write_length(&buf, total, len); + luaL_addlstring(&b, (const char *)buf.ptr, buf.size); + buffer_destroy(&buf); + + luaL_addlstring(&b, (const char *)cmd, cmd_len); + luaL_pushresult(&b); + return 1; +} + + LUAMOD_API int luaopen_skynet_mongo_driver(lua_State *L) { luaL_checkversion(L); luaL_Reg l[] ={ - { "query", op_query }, - { "reply", op_reply }, - { "kill", op_kill }, - { "delete", op_delete }, - { "more", op_get_more }, - { "update", op_update }, - { "insert", op_insert }, + { "reply", unpack_reply }, // 接收响应 { "length", reply_length }, + { "op_msg", op_msg}, { NULL, NULL }, }; luaL_newlib(L,l); return 1; -} +} \ No newline at end of file diff --git a/lualib/skynet/db/mongo.lua b/lualib/skynet/db/mongo.lua index 1fa34b15..00ae86d5 100644 --- a/lualib/skynet/db/mongo.lua +++ b/lualib/skynet/db/mongo.lua @@ -1,5 +1,7 @@ local bson = require "bson" -local socket = require "skynet.socket" + +require "skynet.socket" + local socketchannel = require "skynet.socketchannel" local skynet = require "skynet" local driver = require "skynet.mongo.driver" @@ -68,11 +70,9 @@ local collection_meta = { local function dispatch_reply(so) local len_reply = so:read(4) local reply = so:read(driver.length(len_reply)) - local result = { result = {} } - local succ, reply_id, document, cursor_id, startfrom = driver.reply(reply, result.result) + local result = {} + local succ, reply_id, document = driver.reply(reply) result.document = document - result.cursor_id = cursor_id - result.startfrom = startfrom result.data = reply return reply_id, succ, result end @@ -298,17 +298,37 @@ function mongo_db:runCommand(cmd,cmd_v,...) local sock = conn.__sock local bson_cmd if not cmd_v then - bson_cmd = bson_encode_order(cmd,1) + -- ensure cmd remains in first place + bson_cmd = bson_encode_order(cmd,1, "$db", self.name) else - bson_cmd = bson_encode_order(cmd,cmd_v,...) + bson_cmd = bson_encode_order(cmd,cmd_v, "$db", self.name, ...) end - local pack = driver.query(request_id, 0, self.__cmd, 0, 1, bson_cmd) + + local pack = driver.op_msg(request_id, 0, bson_cmd) -- we must hold req (req.data), because req.document is a lightuserdata, it's a pointer to the string (req.data) local req = sock:request(pack, request_id) local doc = req.document return bson_decode(doc) end +--- send command without response +function mongo_db:send_command(cmd, cmd_v, ...) + local conn = self.connection + local request_id = conn:genId() + local sock = conn.__sock + local bson_cmd + if not cmd_v then + -- ensure cmd remains in first place + bson_cmd = bson_encode_order(cmd, 1, "$db", self.name, "writeConcern", {w=0}) + else + bson_cmd = bson_encode_order(cmd, cmd_v, "$db", self.name, "writeConcern", {w=0}, ...) + end + + local pack = driver.op_msg(request_id, 2, bson_cmd) + sock:request(pack) + return {ok=1} -- fake successful response +end + function mongo_db:getCollection(collection) local col = { connection = self.connection, @@ -326,10 +346,7 @@ function mongo_collection:insert(doc) if doc._id == nil then doc._id = bson.objectid() end - local sock = self.connection.__sock - local pack = driver.insert(0, self.full_name, bson_encode(doc)) - -- flags support 1: ContinueOnError - sock:request(pack) + self.database:send_command("insert", self.name, "documents", {bson_encode(doc)}) end local function werror(r) @@ -353,6 +370,11 @@ function mongo_collection:safe_insert(doc) return werror(r) end +function mongo_collection:raw_safe_insert(doc) + local r = self.database:runCommand("insert", self.name, "documents", {doc}) + return werror(r) +end + function mongo_collection:batch_insert(docs) for i=1,#docs do if docs[i]._id == nil then @@ -360,11 +382,12 @@ function mongo_collection:batch_insert(docs) end docs[i] = bson_encode(docs[i]) end - local sock = self.connection.__sock - local pack = driver.insert(0, self.full_name, docs) - sock:request(pack) + + self.database:send_command("insert", self.name, "documents", docs) end +mongo_collection.insert_many = mongo_collection.batch_insert + function mongo_collection:safe_batch_insert(docs) for i = 1, #docs do if docs[i]._id == nil then @@ -377,16 +400,20 @@ function mongo_collection:safe_batch_insert(docs) return werror(r) end -function mongo_collection:update(selector,update,upsert,multi) - local flags = (upsert and 1 or 0) + (multi and 2 or 0) - local sock = self.connection.__sock - local pack = driver.update(self.full_name, flags, bson_encode(selector), bson_encode(update)) - sock:request(pack) +mongo_collection.safe_insert_many = mongo_collection.safe_batch_insert + +function mongo_collection:update(query,update,upsert,multi) + self.database:send_command("update", self.name, "updates", {bson_encode({ + q = query, + u = update, + upsert = upsert, + multi = multi + })}) end -function mongo_collection:safe_update(selector, update, upsert, multi) +function mongo_collection:safe_update(query, update, upsert, multi) local r = self.database:runCommand("update", self.name, "updates", {bson_encode({ - q = selector, + q = query, u = update, upsert = upsert, multi = multi, @@ -394,6 +421,20 @@ function mongo_collection:safe_update(selector, update, upsert, multi) return werror(r) end +function mongo_collection:batch_update(updates) + local updates_tb = {} + for i = 1, #updates do + updates_tb[i] = bson_encode({ + q = updates[i].query, + u = updates[i].update, + upsert = updates[i].upsert, + multi = updates[i].multi, + }) + end + + self.database:send_command("update", self.name, "updates", updates_tb) +end + function mongo_collection:safe_batch_update(updates) local updates_tb = {} for i = 1, #updates do @@ -409,25 +450,31 @@ function mongo_collection:safe_batch_update(updates) return werror(r) end -function mongo_collection:delete(selector, single) - local sock = self.connection.__sock - local pack = driver.delete(self.full_name, single, bson_encode(selector)) - sock:request(pack) +function mongo_collection:raw_safe_update(update) + local r = self.database:runCommand("update", self.name, "updates", {update}) + return werror(r) end -function mongo_collection:safe_delete(selector, single) +function mongo_collection:delete(query, single) + self.database:runCommand("delete", self.name, "deletes", {bson_encode({ + q = query, + limit = single and 1 or 0, + })}) +end + +function mongo_collection:safe_delete(query, single) local r = self.database:runCommand("delete", self.name, "deletes", {bson_encode({ - q = selector, + q = query, limit = single and 1 or 0, })}) return werror(r) end -function mongo_collection:safe_batch_delete(selectors, single) +function mongo_collection:safe_batch_delete(deletes, single) local delete_tb = {} - for i = 1, #selectors do + for i = 1, #deletes do delete_tb[i] = bson_encode({ - q = selectors[i], + q = deletes[i], limit = single and 1 or 0, }) end @@ -435,30 +482,29 @@ function mongo_collection:safe_batch_delete(selectors, single) return werror(r) end -function mongo_collection:findOne(query, selector) - local conn = self.connection - local request_id = conn:genId() - local sock = conn.__sock - local pack = driver.query(request_id, 0, self.full_name, 0, 1, query and bson_encode(query) or empty_bson, selector and bson_encode(selector)) - -- we must hold req (req.data), because req.document is a lightuserdata, it's a pointer to the string (req.data) - local req = sock:request(pack, request_id) - local doc = req.document - return bson_decode(doc) +function mongo_collection:raw_safe_delete(delete) + local r = self.database:runCommand("delete", self.name, "deletes", {delete}) + return werror(r) end -function mongo_collection:find(query, selector) +function mongo_collection:findOne(query, projection) + local cursor = self:find(query, projection) + return cursor:hasNext() and cursor:next() +end + +function mongo_collection:find(query, projection) return setmetatable( { __collection = self, __query = query and bson_encode(query) or empty_bson, - __selector = selector and bson_encode(selector), + __projection = projection and bson_encode(projection) or empty_bson, __ptr = nil, __data = nil, __cursor = nil, __document = {}, __flags = 0, __skip = 0, - __sortquery = nil, __limit = 0, + __sort = empty_bson, } , cursor_meta) end @@ -479,7 +525,7 @@ function mongo_cursor:sort(key, key_v, ...) local key_list = unfold({}, key, key_v , ...) key = bson_encode_order(table.unpack(key_list)) end - self.__sortquery = bson_encode {['$query'] = self.__query, ['$orderby'] = key} + self.__sort = key return self end @@ -610,21 +656,37 @@ function mongo_collection:findAndModify(doc) return self.database:runCommand(table.unpack(cmd)) end +-- https://docs.mongodb.com/manual/reference/command/aggregate/ +-- collection:aggregate({ { ["$project"] = {tags = 1} } }, {cursor={}}) +-- @param pipeline: array +-- @param options: map +-- @return +function mongo_collection:aggregate(pipeline, options) + assert(pipeline) + local cmd = {"aggregate", self.name, "pipeline", pipeline} + for k, v in pairs(options) do + table.insert(cmd, k) + table.insert(cmd, v) + end + return self.database:runCommand(table.unpack(cmd)) +end + function mongo_cursor:hasNext() if self.__ptr == nil then if self.__document == nil then return false end - local conn = self.__collection.connection - local request_id = conn:genId() - local sock = conn.__sock - local pack + local response + + local database = self.__collection.database if self.__data == nil then - local query = self.__sortquery or self.__query - pack = driver.query(request_id, self.__flags, self.__collection.full_name, self.__skip, self.__limit, query, self.__selector) + local name = self.__collection.name + response = database:runCommand("find", name, "filter", self.__query, "sort", self.__sort, + "skip", self.__skip, "limit", self.__limit, "projection", self.__projection) else - if self.__cursor then - pack = driver.more(request_id, self.__collection.full_name, self.__limit, self.__cursor) + if self.__cursor and self.__cursor > 0 then + local name = self.__collection.name + response = database:runCommand("getMore", bson.int64(self.__cursor), "collection", name, "batchSize", self.__limit) else -- no more self.__document = nil @@ -633,45 +695,31 @@ function mongo_cursor:hasNext() end end - local ok, result = pcall(sock.request,sock,pack, request_id) - - local doc = result.document - local cursor = result.cursor_id - - if ok then - if doc then - local doc = result.result - self.__document = doc - self.__data = result.data - self.__ptr = 1 - self.__cursor = cursor - local limit = self.__limit - if cursor and limit > 0 then - limit = limit - #doc - if limit <= 0 then - -- reach limit - self:close() - end - self.__limit = limit - end - return true - else - self.__document = nil - self.__data = nil - self.__cursor = nil - return false - end - else + if response.ok ~= 1 then self.__document = nil self.__data = nil self.__cursor = nil - if doc then - local err = bson_decode(doc) - error(err["$err"]) - else - error("Reply from mongod error") - end + error(response["$err"] or "Reply from mongod error") end + + local cursor = response.cursor + self.__document = cursor.firstBatch or cursor.nextBatch + self.__data = response + self.__ptr = 1 + self.__cursor = cursor.id + + local limit = self.__limit + if cursor.id > 0 and limit > 0 then + limit = limit - #self.__document + if limit <= 0 then + -- reach limit + self:close() + end + + self.__limit = limit + end + + return true end return true @@ -681,7 +729,7 @@ function mongo_cursor:next() if self.__ptr == nil then error "Call hasNext first" end - local r = bson_decode(self.__document[self.__ptr]) + local r = self.__document[self.__ptr] self.__ptr = self.__ptr + 1 if self.__ptr > #self.__document then self.__ptr = nil @@ -692,11 +740,10 @@ end function mongo_cursor:close() if self.__cursor then - local sock = self.__collection.connection.__sock - local pack = driver.kill(self.__cursor) - sock:request(pack) + local coll = self.__collection + coll.database:send_command("killCursors", coll.name, "cursors", {self.__cursor}) self.__cursor = nil end end -return mongo +return mongo \ No newline at end of file diff --git a/test/testmongodb.lua b/test/testmongodb.lua index 9f1fd810..bff60fab 100644 --- a/test/testmongodb.lua +++ b/test/testmongodb.lua @@ -14,7 +14,7 @@ local function _create_client() { host = host, port = port, username = username, password = password, - authdb = db_name, + authdb = "admin", } ) end @@ -209,7 +209,7 @@ local function test_safe_batch_delete() db.testcoll:safe_batch_delete(docs) local ret = db.testcoll:find() - assert(length == ret:count(), "test safe batch delete failed") + assert((length - del_num) == ret:count(), "test safe batch delete failed") end skynet.start(function()