fix: multiple find (#186)

Author: manyuanrong

README.md

@@ -5,7 +5,7 @@
 [![tag](https://img.shields.io/github/tag/manyuanrong/deno_mongo.svg)](https://github.com/manyuanrong/deno_mongo/releases)
 [![Build Status](https://github.com/manyuanrong/deno_mongo/workflows/ci/badge.svg?branch=master)](https://github.com/manyuanrong/deno_mongo/actions)
 [![license](https://img.shields.io/github/license/manyuanrong/deno_mongo.svg)](https://github.com/manyuanrong/deno_mongo)
-[![tag](https://img.shields.io/badge/deno-v1.5.2-green.svg)](https://github.com/denoland/deno)
+[![tag](https://img.shields.io/badge/deno-v1.8.1-green.svg)](https://github.com/denoland/deno)
 
 ## Links
 
@@ -16,7 +16,7 @@
 ## Examples
 
 ```ts
-import { MongoClient, Bson } from "https://deno.land/x/mongo@v0.21.0/mod.ts";
+import { Bson, MongoClient } from "https://deno.land/x/mongo@v0.22.0/mod.ts";
 
 const client = new MongoClient();
 await client.connect("mongodb://localhost:27017");
@@ -56,7 +56,9 @@ const user1 = await users.findOne({ _id: insertId });
 const all_users = await users.find({ username: { $ne: null } }).toArray();
 
 // find by ObjectId
-const user1_id = await users.findOne({ _id: new Bson.ObjectId("SOME OBJECTID STRING") });
+const user1_id = await users.findOne({
+  _id: new Bson.ObjectId("SOME OBJECTID STRING"),
+});
 
 // count
 const count = await users.count({ username: { $ne: null } });
@@ -70,13 +72,13 @@ const docs = await users.aggregate([
 // updateOne
 const { matchedCount, modifiedCount, upsertedId } = await users.updateOne(
   { username: { $ne: null } },
-  { $set: { username: "USERNAME" } }
+  { $set: { username: "USERNAME" } },
 );
 
 // updateMany
 const { matchedCount, modifiedCount, upsertedId } = await users.updateMany(
   { username: { $ne: null } },
-  { $set: { username: "USERNAME" } }
+  { $set: { username: "USERNAME" } },
 );
 
 // deleteOne

src/protocol/protocol.ts

@@ -6,14 +6,21 @@ import { parseHeader } from "./header.ts";
 import { deserializeMessage, Message, serializeMessage } from "./message.ts";
 
 type Socket = Deno.Reader & Deno.Writer;
+interface CommandTask {
+  requestId: number;
+  db: string;
+  body: Document;
+}
 
 let nextRequestId = 0;
 
 export class WireProtocol {
   #socket: Socket;
-  #pending = false;
-  #pendingOps: Map<number, Deferred<Message>> = new Map();
+  #isPendingResponse = false;
+  #isPendingRequest = false;
+  #pendingResponses: Map<number, Deferred<Message>> = new Map();
   #reader: BufReader;
+  #commandQueue: CommandTask[] = [];
 
   #connectionId: number = 0;
 
@@ -38,26 +45,18 @@ export class WireProtocol {
 
   async command<T = Document>(db: string, body: Document): Promise<T[]> {
     const requestId = nextRequestId++;
-    const chunks = serializeMessage({
+    const commandTask = {
       requestId,
-      responseTo: 0,
-      sections: [
-        {
-          document: {
-            ...body,
-            $db: db,
-          },
-        },
-      ],
-    });
+      db,
+      body,
+    };
 
-    for (const chunk of chunks) {
-      await Deno.writeAll(this.#socket, chunk);
-    }
+    this.#commandQueue.push(commandTask);
+    this.send();
 
-    this.#pendingOps.set(requestId, deferred());
+    this.#pendingResponses.set(requestId, deferred());
     this.receive();
-    const message = await this.#pendingOps.get(requestId);
+    const message = await this.#pendingResponses.get(requestId);
 
     let documents: T[] = [];
 
@@ -72,10 +71,35 @@ export class WireProtocol {
     return documents;
   }
 
+  private async send() {
+    if (this.#isPendingRequest) return;
+    this.#isPendingRequest = true;
+    while (this.#commandQueue.length > 0) {
+      const task = this.#commandQueue.shift()!;
+      const chunks = serializeMessage({
+        requestId: task.requestId,
+        responseTo: 0,
+        sections: [
+          {
+            document: {
+              ...task.body,
+              $db: task.db,
+            },
+          },
+        ],
+      });
+
+      for (const chunk of chunks) {
+        await Deno.writeAll(this.#socket, chunk);
+      }
+    }
+    this.#isPendingRequest = false;
+  }
+
   private async receive() {
-    if (this.#pending) return;
-    this.#pending = true;
-    while (this.#pendingOps.size > 0) {
+    if (this.#isPendingResponse) return;
+    this.#isPendingResponse = true;
+    while (this.#pendingResponses.size > 0) {
       const headerBuffer = await this.#reader.readFull(new Uint8Array(16));
       assert(headerBuffer);
       const header = parseHeader(headerBuffer!);
@@ -84,10 +108,10 @@ export class WireProtocol {
       );
       assert(bodyBuffer);
       const reply = deserializeMessage(header, bodyBuffer!);
-      const pendingMessage = this.#pendingOps.get(header.responseTo);
-      this.#pendingOps.delete(header.responseTo);
+      const pendingMessage = this.#pendingResponses.get(header.responseTo);
+      this.#pendingResponses.delete(header.responseTo);
       pendingMessage?.resolve(reply);
     }
-    this.#pending = false;
+    this.#isPendingResponse = false;
   }
 }

tests/cases/03_curd.ts

@@ -190,6 +190,23 @@ export default function curdTests() {
     assertEquals(notFound, []);
   });
 
+  testWithClient("test multiple queries at the same time", async (client) => {
+    const db = client.database("test");
+    const users = db.collection("mongo_test_users");
+
+    const result = await Promise.all([
+      users.findOne({}, { projection: { username: 1 } }),
+      users.findOne({}, { projection: { username: 1 } }),
+      users.findOne({}, { projection: { username: 1 } }),
+    ]);
+
+    assertEquals(result, [
+      { _id: "aaaaaaaaaaaaaaaaaaaaaaaa", username: "user1" },
+      { _id: "aaaaaaaaaaaaaaaaaaaaaaaa", username: "user1" },
+      { _id: "aaaaaaaaaaaaaaaaaaaaaaaa", username: "user1" },
+    ]);
+  });
+
   testWithClient("testCount", async (client) => {
     const db = client.database("test");
     const users = db.collection("mongo_test_users");
@@ -238,16 +255,6 @@ export default function curdTests() {
     assertEquals(user1, ["USER2", "user1"]);
   });
 
-  // // TODO mongdb_rust official library has not implemented this feature
-  // // testWithClient("testCreateIndexes", async (client) => {
-  // //   const db = client.database("test");
-  // //   const collection = db.collection("mongo_indexes");
-  // //   const result = await collection.createIndexes([
-  // //     { keys: { created_at: 1 }, options: { expireAfterSeconds: 10000 } }
-  // //   ]);
-  // //   console.log(result);
-  // // });
-
   testWithClient("testDropConnection", async (client) => {
     const db = client.database("test");
     await db.collection("mongo_test_users_2").drop();