[misc] compress option work only of server accept compression

geometry in bulk working with MariaDB 10.3
improve testing geometry with batch

lib/cmd/batch-bulk.js

@@ -70,7 +70,7 @@ class BatchBulk extends CommonBinary {
     out.startPacket(this);
 
     this.valueIdx = 0;
-    while (this.valueIdx < this.values.length) {
+    while (this.valueIdx < this.values.length && !this.packet.haveErrorResponse) {
       this.valueRow = this.values[this.valueIdx++];
 
       //********************************************

lib/cmd/batch-rewrite.js

@@ -60,7 +60,7 @@ class BatchRewrite extends CommonText {
 
     this.onPacketReceive = this.readResponsePacket;
     this.valueIdx = 0;
-    while (this.valueIdx < this.values.length) {
+    while (this.valueIdx < this.values.length && !this.packet.haveErrorResponse) {
       this.valueRow = this.values[this.valueIdx++];
 
       //********************************************
@@ -144,6 +144,7 @@ class BatchRewrite extends CommonText {
 
   success(val) {
     this.packet.waitingResponseNo--;
+
     if (this.packet.haveErrorResponse) {
       if (this.sendEnded && this.packet.waitingResponseNo === 0) {
         this.packet = null;
@@ -285,6 +286,11 @@ class BatchRewrite extends CommonText {
   registerStreamSendEvent(packet, info) {
     this.paramWritten = function() {
       while (true) {
+        if (this.packet.haveErrorResponse) {
+          this.sendEnded = true;
+          this.emit("send_end");
+          return;
+        }
         if (this.currentParam === this.valueRow.length) {
           // all parameters from row are written.
           packet.writeString(this.parseResults.partList[this.parseResults.partList.length - 2]);

lib/cmd/common-binary-cmd.js

@@ -53,7 +53,7 @@ class CommonBinary extends ResultSet {
           ) {
             const geoBuff = this.getBufferFromGeometryValue(value);
             if (geoBuff) {
-              out.writeInt8(0x00);
+              out.writeInt8(0x00); //Value follow
               out.writeLengthEncodedBuffer(Buffer.concat([Buffer.from([0, 0, 0, 0]), geoBuff]));
             } else {
               out.writeInt8(0x01); //NULL

lib/connection.js

@@ -13,6 +13,7 @@ const ConnectionInformation = require("./misc/connection-information");
 const tls = require("tls");
 const Errors = require("./misc/errors");
 const Utils = require("./misc/utils");
+const Capabilities = require("./const/capabilities");
 
 /*commands*/
 const Handshake = require("./cmd/handshake/handshake");
@@ -754,10 +755,16 @@ function Connection(options) {
   const _authSucceedHandler = (resolve, rejected) => {
     //enable packet compression according to option
     if (opts.compress) {
-      _out.setStream(new CompressionOutputStream(_socket, opts, info));
-      _in = new CompressionInputStream(_in, _receiveQueue, opts, info);
-      _socket.removeAllListeners("data");
-      _socket.on("data", _in.onData.bind(_in));
+      if (info.serverCapabilities & Capabilities.COMPRESS) {
+        _out.setStream(new CompressionOutputStream(_socket, opts, info));
+        _in = new CompressionInputStream(_in, _receiveQueue, opts, info);
+        _socket.removeAllListeners("data");
+        _socket.on("data", _in.onData.bind(_in));
+      } else {
+        console.error(
+          "connection is configured to use packet compression, but the server doesn't have this capability"
+        );
+      }
     }
 
     if (opts.pipelining) _addCommand = _addCommandEnablePipeline;

lib/io/bulk-packet.js

@@ -70,7 +70,7 @@ class BulkPacket {
               "GeometryCollection"
             ].includes(row[r].type)
           ) {
-            if (this.datatypes[r] !== 0xff) return true;
+            if (this.datatypes[r] !== 0xfb) return true;
           } else {
             if (this.datatypes[r] !== 0x0f) return true;
           }
@@ -122,7 +122,7 @@ class BulkPacket {
               "GeometryCollection"
             ].includes(row[r].type)
           ) {
-            this.buf[this.pos++] = 0xff;
+            this.buf[this.pos++] = 0xfb;
           } else {
             this.buf[this.pos++] = 0x0f;
           }
@@ -532,7 +532,7 @@ class BulkPacket {
 
     this.pos = this.markPos;
 
-    if (!this.haveErrorResponse || this.out.cmd.sequenceNo !== -1) {
+    if (!this.haveErrorResponse) {
       this.copyAndFlush(true);
       this.waitingResponseNo++;
     }

lib/io/rewrite-packet.js

@@ -415,7 +415,7 @@ class ReWritePacket {
     this.pos = this.markPos;
     this.writeString(this.endStr);
 
-    if (!this.haveErrorResponse || this.out.cmd.sequenceNo !== -1) {
+    if (!this.haveErrorResponse) {
       this.copyAndFlush(true);
       this.waitingResponseNo++;
     }

lib/misc/connection-information.js

@@ -7,11 +7,12 @@ class ConnectionInformation {
     this.status = null;
     this.serverVersion = null;
     this.lastPackets = new Queue();
+    this.serverCapabilities = -1;
   }
 
   addPacket(msg) {
     this.lastPackets.push(msg);
-    while (this.lastPackets.size() > 16) this.lastPackets.shift();
+    while (this.lastPackets.size() > 32) this.lastPackets.shift();
   }
 
   getLastPackets() {

test/integration/datatype/test-datetime.js

@@ -41,7 +41,7 @@ describe("datetime", () => {
     }
   });
 
-  it("standard date", done => {
+  it("standard date", function(done) {
     //using distant server, time might be different
     if (Conf.baseConfig.host !== "localhost" && Conf.baseConfig.host !== "mariadb.example.com")
       this.skip();