Merge pull request segmentio#277 from North-Two-Five/payload-size

flushing when payload max size has reached

.circleci/config.yml

@@ -5,7 +5,6 @@ workflows:
       - circle-lint
   multi-test:
     jobs:
-      - test-node8
       - test-node10
       - test-node12
   test_and_publish:
@@ -43,7 +42,6 @@ workflows:
                 - master
                 - scheduled_e2e_testing
     jobs:
-      - test-node8
       - test-node10
       - test-node12
 
@@ -74,10 +72,6 @@ jobs:
           root: .
           paths: [.]
 
-  test-node8:
-    <<: *node-base-test
-    docker:
-      - image: circleci/node:8-browsers
   test-node10:
     <<: *node-base-test
     docker:

index.js

@@ -45,6 +45,7 @@ class Analytics {
     this.axiosInstance = axiosInstance
     this.timeout = options.timeout || false
     this.flushAt = Math.max(options.flushAt, 1) || 20
+    this.maxQueueSize = options.maxQueueSize || 1024 * 500 // defaults to 500kb
     this.flushInterval = options.flushInterval || 10000
     this.flushed = false
     Object.defineProperty(this, 'enable', {
@@ -208,7 +209,9 @@ class Analytics {
       return
     }
 
-    if (this.queue.length >= this.flushAt) {
+    const hasReachedFlushAt = this.queue.length >= this.flushAt
+    const hasReachedQueueSize = this.queue.reduce((acc, item) => acc + JSON.stringify(item).length, 0) >= this.maxQueueSize
+    if (hasReachedFlushAt || hasReachedQueueSize) {
       this.flush(callback)
     }
 

package.json

@@ -16,6 +16,7 @@
     "circle-lint": ".buildscript/circle.sh",
     "dependencies": "yarn",
     "test": "standard && nyc ava --timeout=20s&& .buildscript/e2e.sh",
+    "coverage": "nyc npm run test",
     "report-coverage": "nyc report --reporter=lcov > coverage.lcov && codecov",
     "np": "np --no-publish",
     "release": "yarn run np"
@@ -46,12 +47,12 @@
     "ava": "^0.25.0",
     "basic-auth": "^2.0.1",
     "body-parser": "^1.17.1",
-    "codecov": "^3.0.0",
+    "codecov": "^3.8.1",
     "commander": "^2.9.0",
     "delay": "^4.2.0",
     "express": "^4.15.2",
     "husky": "^3.0.4",
-    "nyc": "^14.1.1",
+    "nyc": "^15.1.0",
     "pify": "^4.0.1",
     "sinon": "^7.3.2",
     "snyk": "^1.171.1",

sample.js

@@ -0,0 +1,19 @@
+const os = require('os')
+const uuid = require('uuid')
+const Analytics = require('.')
+const analytics = new Analytics('xemyw6oe3n')
+
+for (let i = 0; i < 10; i++) {
+  for (let j = 0; j < 10; j++) {
+    analytics.track({
+      anonymousId: uuid.v4(),
+      userId: os.userInfo().username,
+      event: 'Node Test',
+      properties: {
+        count: i + j
+      }
+    })
+  }
+}
+
+analytics.flush()

test.js

@@ -375,6 +375,38 @@ test('flush - skip when client is disabled', async t => {
   t.false(callback.called)
 })
 
+test('flush - flush when reaches max payload size', async t => {
+  const client = createClient({ flushAt: 1000 })
+  client.flush = spy()
+
+  // each of these messages when stringified to json has 220-ish bytes
+  // to satisfy our default limit of 1024*500 bytes we need less than 2600 of those messages
+  const event = {
+    userId: 1,
+    event: 'event'
+  }
+  for (let i = 0; i < 2600; i++) {
+    client.track(event)
+  }
+
+  t.true(client.flush.called)
+})
+
+test('flush - wont flush when no flush condition has meet', async t => {
+  const client = createClient({ flushAt: 1000, maxQueueSize: 1024 * 1000 })
+  client.flush = spy()
+
+  const event = {
+    userId: 1,
+    event: 'event'
+  }
+  for (let i = 0; i < 150; i++) {
+    client.track(event)
+  }
+
+  t.false(client.flush.called)
+})
+
 test('identify - enqueue a message', t => {
   const client = createClient()
   stub(client, 'enqueue')