Merge pull request #152 from calibr/master

add messageQueueSizeLimit option

Author: okkez

README.md

@@ -367,6 +367,10 @@ Set flush interval in milliseconds. This option has no effect in `Message` mode.
 The logger stores emitted events in buffer and flush events for each interval.
 Default `100`.
 
+**messageQueueSizeLimit**
+
+Maximum number of messages that can be in queue at the same time. If a new message is received and it overflows the queue then the oldest message will be removed before adding the new item. This option has effect only in `Message` mode. No limit by default.
+
 **security.clientHostname**
 
 Set hostname of this logger. Use this value for hostname based authentication.

lib/sender.js

@@ -37,6 +37,7 @@ class FluentSender {
     if (this._eventMode === 'Message') {
       this._sendQueue = []; // queue for items waiting for being sent.
       this._flushInterval = 0;
+      this._messageQueueSizeLimit = options.messageQueueSizeLimit || 0;
     } else {
       this._sendQueue = new Map();
       this._flushInterval = options.flushInterval || 100;
@@ -186,6 +187,9 @@ class FluentSender {
       // Message mode
       const item = this._makePacketItem(tag, time, data);
       item.callback = callback;
+      if (this._messageQueueSizeLimit && this._sendQueue.length === this._messageQueueSizeLimit) {
+        this._sendQueue.shift();
+      }
       this._sendQueue.push(item);
     } else {
       // PackedForward mode

test/test.sender.js

@@ -947,6 +947,26 @@ let doTest = (tls) => {
       });
     });
   });
+
+  it('should limit messages stored in queue if server is not available', (done) => {
+    runServer({}, serverOptions, (server, finish) => {
+      finish(() => {
+        const s = new FluentSender('debug', Object.assign({}, clientOptions, {
+          port: server.port,
+          messageQueueSizeLimit: 3
+        }));
+        s.emit('message1', {});
+        s.emit('message2', {});
+        s.emit('message3', {});
+        s.emit('message4', {});
+        expect(s._sendQueue.length).to.be.equal(3);
+        expect(s._sendQueue[0].tag).to.be.equal('debug.message2');
+        expect(s._sendQueue[1].tag).to.be.equal('debug.message3');
+        expect(s._sendQueue[2].tag).to.be.equal('debug.message4');
+        done();
+      });
+    });
+  });
 };
 
 describe('FluentSender', () => {