chore(nodejs): grow the buffer if data does not fit in it

glasstiger · glasstiger · commit 6b6b3b2ef0ae · 2022-08-15T09:20:05.000+01:00
diff --git a/notes.md b/notes.md
@@ -2,11 +2,9 @@
 > ./scripts/generateCerts.sh . questdbPwd123
 
 ### TODO:
-- The client should grow the buffer and call flush() immediately instead of throwing on insufficient buffer size,
-like other clients do.
-Alternatively, the sender could be growing the buffer on row writes and shrinking it on a subsequent flush() call.
-The bufferSize parameter in the constructor can also be optional if this is done.
-- Config object for Sender
+- ~~The client should grow the buffer instead of throwing on insufficient buffer size, like other clients do~~
+- If the buffer had to be extended, shrink it back to original size on a subsequent flush() call?
+- Config object for Sender, make bufferSize optional by setting a default
 - ~~Github actions to run 'npm test' on each commit and run 'npm publish' when version is bumped in package.json~~
 - ~~Provide a builder style API:
   builder.table('tab').symbol('symName', any).intColumn('intCol', 34).timestampColumn('tsCol', 23232323).atNow();~~
diff --git a/src/sender.js b/src/sender.js
@@ -26,6 +26,7 @@ class Sender {
     /** @private */ bufferSize;
     /** @private */ buffer;
     /** @private */ position;
+    /** @private */ endOfLastRow;
     /** @private */ hasTable;
     /** @private */ hasSymbols;
     /** @private */ hasColumns;
@@ -39,18 +40,23 @@ class Sender {
     constructor(bufferSize, jwk = undefined) {
         this.jwk = jwk;
         this.resize(bufferSize);
+        this.reset();
     }
 
     /**
-     * Reinitializes the buffer of the sender. <br>
+     * Extends the size of the sender's buffer. <br>
      * Can be used to increase the size of buffer if overflown.
+     * The buffer's content is copied into the new buffer.
      *
      * @param {number} bufferSize - New size of the buffer used by the sender, provided in bytes.
      */
     resize(bufferSize) {
         this.bufferSize = bufferSize;
-        this.buffer = Buffer.alloc(this.bufferSize + 1, 0, 'utf8');
-        this.reset();
+        const newBuffer = Buffer.alloc(this.bufferSize + 1, 0, 'utf8');
+        if (this.buffer) {
+            this.buffer.copy(newBuffer);
+        }
+        this.buffer = newBuffer;
     }
 
     /**
@@ -136,30 +142,32 @@ class Sender {
     }
 
     /**
-     * Sends the buffer's content to the database and clears the buffer.
+     * Sends the buffer's content to the database and compacts the buffer.
+     * If the last row is not finished it stays in the sender's buffer.
+     *
+     * @return {boolean} Returns true if there was data in the buffer to send.
      */
     async flush() {
-        const data = this.toBuffer();
+        const data = this.toBuffer(this.endOfLastRow);
+        if (!data) {
+            return false;
+        }
         return new Promise((resolve, reject) => {
             this.socket.write(data, err => {
-                this.reset();
-                err ? reject(err) : resolve();
+                compact(this);
+                err ? reject(err) : resolve(true);
             });
         });
     }
 
     /**
      * @ignore
-     * @return {Buffer} Returns a cropped buffer ready to send to the server.
+     * @return {Buffer} Returns a cropped buffer ready to send to the server or null if there is nothing to send.
      */
-    toBuffer() {
-        if (this.hasTable) {
-            throw new Error("The buffer's content is invalid, row needs to be closed by calling at() or atNow()");
-        }
-        if (this.position < 1) {
-            throw new Error("The buffer is empty");
-        }
-        return this.buffer.subarray(0, this.position);
+    toBuffer(pos = this.position) {
+        return pos > 0
+            ? this.buffer.subarray(0, pos)
+            : null;
     }
 
     /**
@@ -176,6 +184,7 @@ class Sender {
             throw new Error("Table name has already been set");
         }
         validateTableName(table);
+        checkCapacity(this, [table]);
         writeEscaped(this, table);
         this.hasTable = true;
         return this;
@@ -195,11 +204,13 @@ class Sender {
         if (!this.hasTable || this.hasColumns) {
             throw new Error("Symbol can be added only after table name is set and before any column added");
         }
+        const valueStr = value.toString();
+        checkCapacity(this, [name, valueStr], 2 + name.length + valueStr.length);
         write(this, ',');
         validateColumnName(name);
         writeEscaped(this, name);
         write(this, '=');
-        writeEscaped(this, value.toString());
+        writeEscaped(this, valueStr);
         this.hasSymbols = true;
         return this;
     }
@@ -213,6 +224,7 @@ class Sender {
      */
     stringColumn(name, value) {
         writeColumn(this, name, value, () => {
+            checkCapacity(this, [value], 2 + value.length);
             write(this, '"');
             writeEscaped(this, value, true);
             write(this, '"');
@@ -229,6 +241,7 @@ class Sender {
      */
     booleanColumn(name, value) {
         writeColumn(this, name, value, () => {
+            checkCapacity(this, [], 1);
             write(this, value ? 't' : 'f');
         }, "boolean");
         return this;
@@ -243,7 +256,9 @@ class Sender {
      */
     floatColumn(name, value) {
         writeColumn(this, name, value, () => {
-            write(this, value.toString());
+            const valueStr = value.toString();
+            checkCapacity(this, [valueStr], valueStr.length);
+            write(this, valueStr);
         }, "number");
         return this;
     }
@@ -260,7 +275,9 @@ class Sender {
             throw new Error(`Value must be an integer, received ${value}`);
         }
         writeColumn(this, name, value, () => {
-            write(this, value.toString());
+            const valueStr = value.toString();
+            checkCapacity(this, [valueStr], 1 + valueStr.length);
+            write(this, valueStr);
             write(this, 'i');
         }, "number");
         return this;
@@ -278,7 +295,9 @@ class Sender {
             throw new Error(`Value must be an integer, received ${value}`);
         }
         writeColumn(this, name, value, () => {
-            write(this, value.toString());
+            const valueStr = value.toString();
+            checkCapacity(this, [valueStr], 1 + valueStr.length);
+            write(this, valueStr);
             write(this, 't');
         }, "number");
         return this;
@@ -297,8 +316,10 @@ class Sender {
             throw new Error(`The designated timestamp must be of type string, received ${typeof timestamp}`);
         }
         validateDesignatedTimestamp(timestamp);
+        const timestampStr = timestamp.toString();
+        checkCapacity(this, [], 2 + timestampStr.length);
         write(this, ' ');
-        write(this, timestamp.toString());
+        write(this, timestampStr);
         write(this, '\n');
         startNewRow(this);
     }
@@ -311,6 +332,7 @@ class Sender {
         if (!this.hasSymbols && !this.hasColumns) {
             throw new Error("The row must have a symbol or column set before it is closed");
         }
+        checkCapacity(this, [], 1);
         write(this, '\n');
         startNewRow(this);
     }
@@ -338,11 +360,34 @@ async function authenticate(sender, challenge) {
 }
 
 function startNewRow(sender) {
+    sender.endOfLastRow = sender.position;
     sender.hasTable = false;
     sender.hasSymbols = false;
     sender.hasColumns = false;
 }
 
+function checkCapacity(sender, data, base = 0) {
+    let length = base;
+    for (const str of data) {
+        length += Buffer.byteLength(str, 'utf8');
+    }
+    if (sender.position + length > sender.bufferSize) {
+        let newSize = sender.bufferSize;
+        do {
+            newSize += sender.bufferSize;
+        } while(sender.position + length > newSize);
+        sender.resize(newSize);
+    }
+}
+
+function compact(sender) {
+    if (sender.endOfLastRow > 0) {
+        sender.buffer.copy(sender.buffer, 0, sender.endOfLastRow, sender.position);
+        sender.position = sender.position - sender.endOfLastRow;
+        sender.endOfLastRow = 0;
+    }
+}
+
 function writeColumn(sender, name, value, writeValue, valueType) {
     if (typeof name !== "string") {
         throw new Error(`Column name must be a string, received ${typeof name}`);
@@ -353,6 +398,7 @@ function writeColumn(sender, name, value, writeValue, valueType) {
     if (!sender.hasTable) {
         throw new Error("Column can be set only after table name is set");
     }
+    checkCapacity(sender, [name], 2 + name.length);
     write(sender, sender.hasColumns ? ',' : ' ');
     validateColumnName(name);
     writeEscaped(sender, name);
diff --git a/src/validation.js b/src/validation.js
@@ -132,20 +132,8 @@ function validateDesignatedTimestamp(timestamp) {
     }
     for (let i = 0; i < len; i++) {
         let ch = timestamp[i];
-        switch (ch) {
-            case '0':
-            case '1':
-            case '2':
-            case '3':
-            case '4':
-            case '5':
-            case '6':
-            case '7':
-            case '8':
-            case '9':
-                break;
-            default:
-                throw new Error(`Invalid character in designated timestamp: ${ch}`);
+        if (ch < '0' || ch > '9') {
+            throw new Error(`Invalid character in designated timestamp: ${ch}`);
         }
     }
 }
diff --git a/test/sender.test.js b/test/sender.test.js
@@ -335,20 +335,21 @@ describe('Sender message builder test suite (anything not covered in client inte
         ).toThrow("Symbol can be added only after table name is set and before any column added");
     });
 
-    it('throws exception if preparing an empty buffer for send', function () {
+    it('returns null if preparing an empty buffer for send', function () {
         const sender = new Sender(1024);
-        expect(
-            () => sender.toBuffer()
-        ).toThrow("The buffer is empty");
+        expect(sender.toBuffer()).toBe(null);
     });
 
-    it('throws exception if preparing a buffer with an unclosed row for send', function () {
+    it('ignores unfinished rows when preparing a buffer for send', function () {
         const sender = new Sender(1024);
+        sender.table("tableName")
+            .symbol("name", "value")
+            .at("1234567890");
+        sender.table("tableName")
+            .symbol("name", "value2");
         expect(
-            () => sender.table("tableName")
-                .symbol("name", "value")
-                .toBuffer()
-        ).toThrow("The buffer's content is invalid, row needs to be closed by calling at() or atNow()");
+            sender.toBuffer(sender.endOfLastRow).toString()
+        ).toBe("tableName,name=value 1234567890\n");
     });
 
     it('throws exception if a float is passed as integer field', function () {
@@ -394,31 +395,35 @@ describe('Sender message builder test suite (anything not covered in client inte
         ).toThrow("The row must have a symbol or column set before it is closed");
     });
 
-    it('throws exception if buffer overflows', function () {
-        const sender = new Sender(16);
-        expect(
-            () => sender.table("tableName")
-                .intColumn("intField", 123)
-        ).toThrow("Buffer overflow [position=17, bufferSize=16]");
-    });
-
-    it('can fix buffer overflows by calling resize()', function () {
-        const sender = new Sender(16);
-        expect(
-            () => sender.table("tableName")
-                .floatColumn("floatField", 123.34)
-        ).toThrow("Buffer overflow [position=17, bufferSize=16]");
+    it('extends the size of the buffer if data does not fit', function () {
+        const sender = new Sender(8);
+        expect(sender.bufferSize).toBe(8);
+        expect(sender.position).toBe(0);
+        sender.table("tableName");
+        expect(sender.bufferSize).toBe(16);
+        expect(sender.position).toBe("tableName".length);
+        sender.intColumn("intField", 123);
+        expect(sender.bufferSize).toBe(32);
+        expect(sender.position).toBe("tableName intField=123i".length);
+        sender.atNow();
+        expect(sender.bufferSize).toBe(32);
+        expect(sender.position).toBe("tableName intField=123i\n".length);
+        expect(sender.toBuffer().toString()).toBe(
+            "tableName intField=123i\n"
+        );
 
-        sender.resize(1024);
-        sender.table("tableName")
-            .floatColumn("floatField", 123.34)
+        sender.table("table2")
+            .intColumn("intField", 125)
+            .stringColumn("strField", "test")
             .atNow();
+        expect(sender.bufferSize).toBe(64);
+        expect(sender.position).toBe("tableName intField=123i\ntable2 intField=125i,strField=\"test\"\n".length);
         expect(sender.toBuffer().toString()).toBe(
-            "tableName floatField=123.34\n"
+            "tableName intField=123i\ntable2 intField=125i,strField=\"test\"\n"
         );
     });
 
-    it('is possible to reuse the buffer by calling reset()', function () {
+    it('is possible to clear the buffer by calling reset()', function () {
         const sender = new Sender(1024);
         sender.table("tableName")
             .booleanColumn("boolCol", true)

Original file line number	Diff line number	Diff line change
`@@ -132,20 +132,8 @@ function validateDesignatedTimestamp(timestamp) {`
`132`	`132`	`}`
`133`	`133`	`for (let i = 0; i < len; i++) {`
`134`	`134`	`let ch = timestamp[i];`
`135`		`- switch (ch) {`
`136`		`- case '0':`
`137`		`- case '1':`
`138`		`- case '2':`
`139`		`- case '3':`
`140`		`- case '4':`
`141`		`- case '5':`
`142`		`- case '6':`
`143`		`- case '7':`
`144`		`- case '8':`
`145`		`- case '9':`
`146`		`- break;`
`147`		`- default:`
`148`		- throw new Error(`Invalid character in designated timestamp: ${ch}`);
	`135`	`+ if (ch < '0' \|\| ch > '9') {`
	`136`	+ throw new Error(`Invalid character in designated timestamp: ${ch}`);
`149`	`137`	`}`
`150`	`138`	`}`
`151`	`139`	`}`