Merge pull request #155 from mitsos1os/fix-maultiple-tags-queue-flush

Fix packets on multiple tags get corrput and multiple calls of callbacks on error

Author: okkez

.gitignore

@@ -1 +1,2 @@
 node_modules
+/package-lock.json

lib/sender.js

@@ -159,11 +159,9 @@ class FluentSender {
     }
 
     const packet = [tag, time, data];
-    let options = {};
+    const options = {};
     if (this.requireAckResponse) {
-      options = {
-        chunk: crypto.randomBytes(16).toString('base64')
-      };
+      options.chunk = crypto.randomBytes(16).toString('base64');
       packet.push(options);
     }
     return {
@@ -186,20 +184,25 @@ class FluentSender {
   _push(tag, time, data, callback) {
     if (this._eventMode === 'Message') {
       // Message mode
-      let item = this._makePacketItem(tag, time, data);
+      const item = this._makePacketItem(tag, time, data);
       item.callback = callback;
       this._sendQueue.push(item);
     } else {
       // PackedForward mode
       const eventEntry = this._makeEventEntry(time, data);
       this._sendQueueSize += eventEntry.length;
       if (this._sendQueue.has(tag)) {
-        let eventEntryData = this._sendQueue.get(tag);
+        const eventEntryData = this._sendQueue.get(tag);
         eventEntryData.eventEntries.push(eventEntry);
+        eventEntryData.size += eventEntry.length;
         if (callback) eventEntryData.callbacks.push(callback);
       } else {
         const callbacks = callback ? [callback] : [];
-        this._sendQueue.set(tag, { eventEntries: [eventEntry], callbacks: callbacks });
+        this._sendQueue.set(tag, {
+          eventEntries: [eventEntry],
+          size: eventEntry.length,
+          callbacks: callbacks
+        });
       }
     }
   }
@@ -225,8 +228,8 @@ class FluentSender {
   }
 
   _doConnect(callback) {
-    let addHandlers = () => {
-      let errorHandler = (err) => {
+    const addHandlers = () => {
+      const errorHandler = (err) => {
         if (this._socket) {
           this._disconnect();
           this._handleEvent('error', err);
@@ -260,7 +263,7 @@ class FluentSender {
         });
       }
     } else {
-      let postConnect = () => {
+      const postConnect = () => {
         if (this.security.clientHostname && this.security.sharedKey !== null) {
           this._handshake(callback);
         } else {
@@ -362,7 +365,7 @@ class FluentSender {
 
   _doFlushSendQueue(timeoutId) {
     if (this._eventMode === 'Message') {
-      let item = this._sendQueue.shift();
+      const item = this._sendQueue.shift();
       if (item === undefined) {
         this._flushingSendQueue = false;
         // nothing written;
@@ -374,10 +377,10 @@ class FluentSender {
         this._flushingSendQueue = false;
         return;
       }
-      let first = this._sendQueue.entries().next().value;
-      let tag = first[0];
-      let eventEntryData = first[1];
-      let entries = Buffer.concat(eventEntryData.eventEntries, this._sendQueueSize);
+      const first = this._sendQueue.entries().next().value;
+      const tag = first[0];
+      const eventEntryData = first[1];
+      let entries = Buffer.concat(eventEntryData.eventEntries, eventEntryData.size);
       let size = entries.length;
       this._sendQueue.delete(tag);
       if (this._compressed) {
@@ -387,15 +390,16 @@ class FluentSender {
       const options = {
         chunk: crypto.randomBytes(16).toString('base64'),
         size: size,
-        compressed: this._compressed ? 'gzip' : 'text'
+        compressed: this._compressed ? 'gzip' : 'text',
+        eventEntryDataSize: eventEntryData.size
       };
       const packet = msgpack.encode([tag, entries, options], { codec: codec });
       this._doWrite(packet, options, timeoutId, eventEntryData.callbacks);
     }
   }
 
   _doWrite(packet, options, timeoutId, callbacks) {
-    const sendQueueSize = this._sendQueueSize;
+    const sendPacketSize = (options && options.eventEntryDataSize) || this._sendQueueSize;
     this._socket.write(packet, () => {
       if (this.requireAckResponse) {
         this._socket.once('data', (data) => {
@@ -409,11 +413,12 @@ class FluentSender {
             callbacks.forEach((callback) => {
               this._handleEvent('error', error, callback);
             });
+          } else { // no error on ack
+            callbacks.forEach((callback) => {
+              callback && callback();
+            });
           }
-          this._sendQueueSize -= sendQueueSize;
-          callbacks.forEach((callback) => {
-            callback && callback();
-          });
+          this._sendQueueSize -= sendPacketSize;
           process.nextTick(() => {
             this._waitToWrite();
           });
@@ -425,7 +430,7 @@ class FluentSender {
           });
         }, this.ackResponseTimeout);
       } else {
-        this._sendQueueSize -= sendQueueSize;
+        this._sendQueueSize -= sendPacketSize;
         callbacks.forEach((callback) => {
           callback && callback();
         });
@@ -452,10 +457,10 @@ class FluentSender {
       this._status = null;
       this.internalLogger.error('Fluentd error', error);
       this.internalLogger.info('Fluentd will reconnect after ' + this.reconnectInterval / 1000 + ' seconds');
-      let timeoutId = setTimeout(() => {
+      const timeoutId = setTimeout(() => {
         this.internalLogger.info('Fluentd is reconnecting...');
         this._connect(() => {
-          this._flushSendQueue()
+          this._flushSendQueue();
           this.internalLogger.info('Fluentd reconnection finished!!');
         });
       }, this.reconnectInterval);
@@ -562,7 +567,7 @@ class FluentSender {
     let dataString = '';
     writable._write = (chunk, encoding, callback) => {
       const dataArray = chunk.toString(defaultEncoding).split(/\n/);
-      let next = () => {
+      const next = () => {
         if (dataArray.length) {
           dataString += dataArray.shift();
         }

test/test.sender.js

@@ -316,6 +316,41 @@ let doTest = (tls) => {
     }, 100);
   });
 
+  it('should send messages with different tags correctly in PackedForward', (done) => {
+    runServer({}, serverOptions, (server, finish) => {
+      const s1 = new FluentSender('debug', Object.assign({}, clientOptions, {
+        port: server.port,
+        eventMode: 'PackedForward'
+      }));
+      const emits = [];
+      const total = 4;
+      function emit(messageData) {
+        emits.push((asyncDone) => {
+          if (messageData.number === total) { // end
+            s1.emit(`multi-${messageData.number}`, { text: messageData.text}, asyncDone); // wait for send
+          } else {
+            s1.emit(`multi-${messageData.number}`, { text: messageData.text});
+            asyncDone(); // run immediately do not wait for ack
+          }
+        });
+      }
+      for (let i = 0; i <= total; i++) {
+        emit({ number: i, text: `This is text No ${i}` });
+      }
+      emits.push(() => {
+        finish((data) => {
+          expect(data.length).to.be.equal(5);
+          data.forEach((element, index) => {
+            expect(element.tag).to.be.equal(`debug.multi-${index}`);
+            expect(element.data.text).to.be.equal(`This is text No ${index}`);
+          });
+          done();
+        });
+      });
+      async.series(emits);
+    });
+  });
+
   [
     {
       name: 'tag and record',