warning if non-flushed data is lost at close()

Author: glasstiger

src/buffer/index.ts

@@ -329,6 +329,14 @@ class SenderBuffer {
     this.startNewRow();
   }
 
+  /**
+   * Returns the current position of the buffer.
+   * New data will be written into the buffer starting from this position.
+   */
+  currentPosition() {
+    return this.position;
+  }
+
   private checkCapacity(data: string[], base = 0) {
     let length = base;
     for (const str of data) {

src/sender.ts

@@ -168,10 +168,14 @@ class Sender {
   }
 
   /**
-   * Closes the TCP connection to the database. <br>
+   * Closes the connection to the database. <br>
    * Data sitting in the Sender's buffer will be lost unless flush() is called before close().
    */
   async close() {
+    const pos = this.buffer.currentPosition();
+    if (pos > 0) {
+      this.log("warn", `Buffer contains data which has not been flushed, and it will be lost [position=${pos}]`);
+    }
     return this.transport.close();
   }
 

src/transport/tcp.ts

@@ -176,7 +176,6 @@ class TcpTransport implements SenderTransport {
 
   /**
    * Closes the TCP connection to the database. <br>
-   * Data sitting in the Sender's buffer will be lost unless flush() is called before close().
    */
   async close(): Promise<void> {
     if (this.socket) {

test/sender.transport.test.ts

@@ -570,4 +570,46 @@ describe("Sender TCP suite", function () {
     await sender.close();
     await proxy.stop();
   });
+
+  it("warns if data is lost on close()", async function () {
+    // we expect a warning about non-flushed data at close()
+    const expectedMessages = [
+      "Successfully connected to localhost:9088",
+      "Buffer contains data which has not been flushed, and it will be lost [position=54]",
+      /^Connection to .*1:9088 is closed$/,
+    ];
+    const log = (
+      level: "error" | "warn" | "info" | "debug",
+      message: string,
+    ) => {
+      if (level !== "debug") {
+        expect(message).toMatch(expectedMessages.shift());
+      }
+    };
+    const proxy = await createProxy();
+    const sender = new Sender({
+      protocol: "tcp",
+      port: PROXY_PORT,
+      host: PROXY_HOST,
+      log: log,
+    });
+    await sender.connect();
+    await sendData(sender);
+
+    // add another line to the buffer without calling flush()
+    await sender
+        .table("test")
+        .symbol("location", "gb")
+        .floatColumn("temperature", 16.4)
+        .at(1658484775000000000n, "ns");
+
+    // assert that only the first line was sent
+    await assertSentData(
+      proxy,
+      false,
+      "test,location=us temperature=17.1 1658484765000000000\n",
+    );
+    await sender.close();
+    await proxy.stop();
+  });
 });