launchdarkly
diff --git a/‎index.d.ts‎
Lines changed: 3 additions & 2 deletions b/‎index.d.ts‎
Lines changed: 3 additions & 2 deletions
diff --git a/‎polling.js‎
Lines changed: 16 additions & 19 deletions b/‎polling.js‎
Lines changed: 16 additions & 19 deletions
diff --git a/‎test/LDClient-tls-test.js‎
Lines changed: 7 additions & 5 deletions b/‎test/LDClient-tls-test.js‎
Lines changed: 7 additions & 5 deletions
diff --git a/‎test/event_processor-test.js‎
Lines changed: 4 additions & 3 deletions b/‎test/event_processor-test.js‎
Lines changed: 4 additions & 3 deletions
diff --git a/‎test/polling-test.js‎
Lines changed: 65 additions & 41 deletions b/‎test/polling-test.js‎
Lines changed: 65 additions & 41 deletions
@@ -337,11 +337,12 @@ declare module 'launchdarkly-node-server-sdk' {
     streamInitialReconnectDelay?: number;
 
     /**
-     * Whether you are using the LaunchDarkly relay proxy in daemon mode.
+     * Whether you are using the LaunchDarkly Relay Proxy in daemon mode.
      *
      * In this configuration, the client will not connect to LaunchDarkly to get feature flags,
      * but will instead get feature state from a database (Redis or another supported feature
-     * store integration) that is populated by the relay. By default, this is false.
+     * store integration) that is populated by the Relay Proxy. By default, this is false.
+     * To learn more, read [Using daemon mode](https://docs.launchdarkly.com/home/relay-proxy/using#using-daemon-mode).
      */
     useLdd?: boolean;
 
 
@@ -4,33 +4,31 @@ const dataKind = require('./versioned_data_kind');
 
 function PollingProcessor(config, requestor) {
   const processor = {},
-    featureStore = config.featureStore;
+    featureStore = config.featureStore,
+    intervalMs = config.pollInterval * 1000;
+
   let stopped = false;
 
+  let pollTask;
+
   function poll(maybeCallback) {
     const cb = maybeCallback || function () {};
 
     if (stopped) {
       return;
     }
 
-    const startTime = new Date().getTime();
     config.logger.debug('Polling LaunchDarkly for feature flag updates');
+
     requestor.requestAllData((err, respBody) => {
-      const elapsed = new Date().getTime() - startTime;
-      const sleepFor = Math.max(config.pollInterval * 1000 - elapsed, 0);
-      config.logger.debug('Elapsed: %d ms, sleeping for %d ms', elapsed, sleepFor);
       if (err) {
         if (err.status && !errors.isHttpErrorRecoverable(err.status)) {
           const message = messages.httpErrorMessage(err, 'polling request');
           config.logger.error(message);
           cb(new errors.LDPollingError(message));
+          processor.stop();
         } else {
           config.logger.warn(messages.httpErrorMessage(err, 'polling request', 'will retry'));
-          // Recursively call poll after the appropriate delay
-          setTimeout(() => {
-            poll(cb);
-          }, sleepFor);
         }
       } else {
         if (respBody) {
@@ -40,27 +38,26 @@ function PollingProcessor(config, requestor) {
           initData[dataKind.segments.namespace] = allData.segments;
           featureStore.init(initData, () => {
             cb();
-            // Recursively call poll after the appropriate delay
-            setTimeout(() => {
-              poll(cb);
-            }, sleepFor);
           });
-        } else {
-          // There wasn't an error but there wasn't any new data either, so just keep polling
-          setTimeout(() => {
-            poll(cb);
-          }, sleepFor);
         }
+        // There wasn't an error but there wasn't any new data either, so just keep polling
       }
     });
   }
 
   processor.start = cb => {
-    poll(cb);
+    if (!pollTask && !stopped) {
+      pollTask = setInterval(() => poll(cb), intervalMs);
+      // setInterval always waits for the delay before firing the first time, but we want to do an initial poll right away
+      poll(cb);
+    }
   };
 
   processor.stop = () => {
     stopped = true;
+    if (pollTask) {
+      clearInterval(pollTask);
+    }
   };
 
   processor.close = () => {
 
@@ -2,12 +2,12 @@ import * as LDClient from '../index';
 
 import {
   AsyncQueue,
-  sleepAsync,
   TestHttpHandlers,
   TestHttpServer,
   withCloseable
 } from 'launchdarkly-js-test-helpers';
 import * as stubs from './stubs';
+import { failIfTimeout } from './test_helpers';
 
 describe('LDClient TLS configuration', () => {
   const sdkKey = 'secret';
@@ -36,18 +36,20 @@ describe('LDClient TLS configuration', () => {
     await withCloseable(TestHttpServer.startSecure, async server => {
       server.forMethodAndPath('get', '/sdk/latest-all', TestHttpHandlers.respondJson({}));
 
+      const logCapture = stubs.asyncLogCapture();
       const config = {
         baseUri: server.url,
         sendEvents: false,
         stream: false,
-        logger: stubs.stubLogger(),
+        logger: logCapture.logger,
         diagnosticOptOut: true,
       };
 
       await withCloseable(LDClient.init(sdkKey, config), async client => {
-        await sleepAsync(300); // the client won't signal an unrecoverable error, but it should log a message
-        expect(config.logger.warn.mock.calls.length).toEqual(2);
-        expect(config.logger.warn.mock.calls[1][0]).toMatch(/self.signed/);
+        const message1 = await failIfTimeout(logCapture.warn.take(), 1000);
+        expect(message1).toMatch(/only disable the streaming API/); // irrelevant message due to our use of polling mode
+        const message2 = await failIfTimeout(logCapture.warn.take(), 1000);
+        expect(message2).toMatch(/self.signed/);
       });
     });
   });
 
@@ -1,6 +1,7 @@
 const { DiagnosticsManager, DiagnosticId } = require('../diagnostic_events');
 const EventProcessor = require('../event_processor');
-const { sleepAsync, TestHttpHandlers, TestHttpServer, withCloseable } = require('launchdarkly-js-test-helpers');
+const { TestHttpHandlers, TestHttpServer, withCloseable } = require('launchdarkly-js-test-helpers');
+const { failIfTimeout } = require('./test_helpers');
 
 describe('EventProcessor', () => {
 
@@ -656,8 +657,8 @@ describe('EventProcessor', () => {
       ep.sendEvent({ kind: 'identify', creationDate: 1000, user: user });
 
       // unfortunately we must wait for both the flush interval and the 1-second retry interval
-      await sleepAsync(1500);
-      expect(s.requestCount()).toEqual(2);
+      await failIfTimeout(s.nextRequest(), 500);
+      await failIfTimeout(s.nextRequest(), 1500);
     });
   }));
 
 
@@ -1,8 +1,9 @@
 const InMemoryFeatureStore = require('../feature_store');
 const PollingProcessor = require('../polling');
 const dataKind = require('../versioned_data_kind');
-const { promisify, promisifySingle, sleepAsync } = require('launchdarkly-js-test-helpers');
+const { AsyncQueue, promisify, promisifySingle } = require('launchdarkly-js-test-helpers');
 const stubs = require('./stubs');
+const { failIfTimeout } = require('./test_helpers');
 
 describe('PollingProcessor', () => {
   const longInterval = 100000;
@@ -48,7 +49,8 @@ describe('PollingProcessor', () => {
     };
     processor = PollingProcessor(config, requestor);
 
-    await promisify(processor.start)(); // didn't throw -> success
+    const err = await new Promise(resolve => processor.start(resolve));
+    expect(err).not.toBe(expect.anything());
   });
 
   it('initializes feature store', async () => {
@@ -66,68 +68,90 @@ describe('PollingProcessor', () => {
   });
 
   it('polls repeatedly', async() => {
+    const calls = new AsyncQueue();
     const requestor = {
-      requestAllData: jest.fn(cb => cb(null, jsonData))
+      requestAllData: cb => {
+        calls.add();
+        cb(null, jsonData);
+      }
     };
-    config.pollInterval = 0.1;  // note, pollInterval is in seconds
+    config.pollInterval = 0.05;  // note, pollInterval is in seconds
     processor = PollingProcessor(config, requestor);
 
     processor.start(() => {});
-    await sleepAsync(500);
-
-    expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(4);
+    const startTime = new Date().getTime();
+    for (let i = 0; i < 4; i++) {
+      await failIfTimeout(calls.take(), 500);
+    }
+    expect(new Date().getTime() - startTime).toBeLessThanOrEqual(500);
   });
 
   async function testRecoverableError(err) {
+    const calls = new AsyncQueue();
+    let count = 0;
     const requestor = {
-      requestAllData: jest.fn(cb => cb(err))
+      // The first two calls will return the error; the third will succeed.
+      requestAllData: cb => {
+        calls.add();
+        count++;
+        if (count > 2) {
+          cb(null, jsonData);
+        } else {
+          cb(err);
+        }
+      }
     };
-    config.pollInterval = 0.1;
+    config.pollInterval = 0.05;
     processor = PollingProcessor(config, requestor);
 
     let errReceived;
     processor.start(e => { errReceived = e; });
-    await sleepAsync(300);
 
-    expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(2);
+    await failIfTimeout(calls.take(), 500);
+    await failIfTimeout(calls.take(), 500);
+    await failIfTimeout(calls.take(), 500);
+
     expect(config.logger.error).not.toHaveBeenCalled();
     expect(errReceived).toBeUndefined();
   }
 
-  function testRecoverableHttpError(status) {
+  it.each([400, 408, 429, 500, 503])(
+    'continues polling after error %d',
+    async (status) => {
+      const err = new Error('sorry');
+      err.status = status;
+      await testRecoverableError(err);
+    }
+  );
+
+  it('continues polling after I/O error', async () => await testRecoverableError(new Error('sorry')));
+
+  async function testUnrecoverableError(status) {
     const err = new Error('sorry');
     err.status = status;
-    it('continues polling after error ' + status, async () => await testRecoverableError(err));
-  }
 
-  testRecoverableHttpError(400);
-  testRecoverableHttpError(408);
-  testRecoverableHttpError(429);
-  testRecoverableHttpError(500);
-  testRecoverableHttpError(503);
+    const calls = new AsyncQueue();
+    const requestor = {
+      requestAllData: cb => {
+        calls.add();
+        cb(err);
+      }
+    };
+    config.pollInterval = 0.05;
+    processor = PollingProcessor(config, requestor);
 
-  it('continues polling after I/O error', async () => await testRecoverableError(new Error('sorry')));
+    const result = new AsyncQueue();
+    processor.start(e => result.add(e));
 
-  function testUnrecoverableHttpError(status) {
-    it('stops polling after error ' + status, async () => {
-      const err = new Error('sorry');
-      err.status = status;
-      const requestor = {
-        requestAllData: jest.fn(cb => cb(err))
-      };
-      config.pollInterval = 0.1;
-      processor = PollingProcessor(config, requestor);
-
-      let errReceived;
-      processor.start(e => { errReceived = e; });
-      await sleepAsync(300);
-
-      expect(requestor.requestAllData.mock.calls.length).toEqual(1);
-      expect(config.logger.error).toHaveBeenCalledTimes(1);
-      expect(errReceived).not.toBeUndefined();
-    });
-  }
+    const errReceived = await failIfTimeout(result.take(), 1000);
+    expect(errReceived.message).toMatch(new RegExp('error ' + status + '.*giving up permanently'));
 
-  testUnrecoverableHttpError(401);
-  testUnrecoverableHttpError(403);
+    expect(calls.length()).toEqual(1);
+    expect(config.logger.error).toHaveBeenCalledTimes(1);
+  }
+  
+  it.each([401, 403])(
+    'stops polling after error %d',
+    testUnrecoverableError
+  );  
 });