fix(event processor): Stop accepting events after stop is called (optimizely#355)

mjc1283 · web-flow · commit 8fc6d38f64ef · 2019-08-29T10:08:44.000-07:00
Summary:

Before this change, event processor would continue processing user events after its stop method was called. If the batch size limit was reached after stop was called, it would dispatch a log event as normal. Worse, the internal queue wasn't being emptied when stop was called, which could lead to the same user event being included in two different log events.

With this change, the event processor's internal queue stops accepting events after the event processor is stopped. A warning message is logged, but the event is dropped.

Test plan:

Updated unit tests. Will manually test upon integration of a new version in optimizely-sdk.
diff --git a/packages/event-processor/CHANGELOG.MD b/packages/event-processor/CHANGELOG.MD
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 Changes that have landed but are not yet released.
 
+### Fixed
+- `DefaultEventQueue` no longer enqueues additional events after being stopped. As a result, `AbstractEventProcessor` no longer processes events after being stopped.
+- `DefaultEventQueue` clears its buffer after being stopped. Event duplication, which was previously possible when additional events were enqueued after the stop, is no longer possible.
+
 ## [0.3.0] - August 13, 2019
 
 ### New Features
diff --git a/packages/event-processor/__tests__/eventQueue.spec.ts b/packages/event-processor/__tests__/eventQueue.spec.ts
@@ -266,5 +266,25 @@ describe('eventQueue', () => {
       queue.stop()
 
     })
+
+    it('should not enqueue additional events after stop() is called', () => {
+      const sinkFn = jest.fn()
+      const queue = new DefaultEventQueue<number>({
+        flushInterval: 30000,
+        maxQueueSize: 3,
+        sink: sinkFn,
+        batchComparator: () => true
+      })
+      queue.start()
+      queue.enqueue(1)
+      queue.stop()
+      expect(sinkFn).toHaveBeenCalledTimes(1)
+      expect(sinkFn).toHaveBeenCalledWith([1])
+      sinkFn.mockClear()
+      queue.enqueue(2)
+      queue.enqueue(3)
+      queue.enqueue(4)
+      expect(sinkFn).toBeCalledTimes(0)
+    })
   })
 })
diff --git a/packages/event-processor/__tests__/v1EventProcessor.spec.ts b/packages/event-processor/__tests__/v1EventProcessor.spec.ts
@@ -154,6 +154,7 @@ describe('LogTierV1EventProcessor', () => {
         flushInterval: 100,
         maxQueueSize: 100,
       })
+      processor.start()
 
       const impressionEvent = createImpressionEvent()
       processor.process(impressionEvent)
@@ -183,6 +184,7 @@ describe('LogTierV1EventProcessor', () => {
         flushInterval: 100,
         maxQueueSize: 100,
       })
+      processor.start()
 
       const impressionEvent = createImpressionEvent()
       processor.process(impressionEvent)
@@ -211,6 +213,7 @@ describe('LogTierV1EventProcessor', () => {
         flushInterval: 100,
         maxQueueSize: 100,
       })
+      processor.start()
 
       const impressionEvent1 = createImpressionEvent()
       const impressionEvent2 = createImpressionEvent()
@@ -223,6 +226,41 @@ describe('LogTierV1EventProcessor', () => {
         done()
       })
     })
+
+    it('should stop accepting events after stop is called', () => {
+      const dispatcher = {
+        dispatchEvent: jest.fn((event: EventV1Request, callback: EventDispatcherCallback) => {
+          setTimeout(() => callback({ statusCode: 204 }), 0)
+        })
+      }
+      const processor = new LogTierV1EventProcessor({
+        dispatcher,
+        flushInterval: 100,
+        maxQueueSize: 3,
+      })
+      processor.start()
+
+      const impressionEvent1 = createImpressionEvent()
+      processor.process(impressionEvent1)
+      processor.stop()
+      // calling stop should haver flushed the current batch of size 1
+      expect(dispatcher.dispatchEvent).toBeCalledTimes(1)
+
+      dispatcher.dispatchEvent.mockClear();
+
+      // From now on, subsequent events should be ignored.
+      // Process 3 more, which ordinarily would have triggered
+      // a flush due to the batch size.
+      const impressionEvent2 = createImpressionEvent()
+      processor.process(impressionEvent2)
+      const impressionEvent3 = createImpressionEvent()
+      processor.process(impressionEvent3)
+      const impressionEvent4 = createImpressionEvent()
+      processor.process(impressionEvent4)
+      // Since we already stopped the processor, the dispatcher should
+      // not have been called again.
+      expect(dispatcher.dispatchEvent).toBeCalledTimes(0)
+    })
   })
 
   describe('when maxQueueSize = 1', () => {
diff --git a/packages/event-processor/src/eventQueue.ts b/packages/event-processor/src/eventQueue.ts
@@ -13,8 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+import { getLogger } from '@optimizely/js-sdk-logging'
 // TODO change this to use Managed from js-sdk-models when available
 import { Managed } from './managed'
+
+const logger = getLogger('EventProcessor')
+
 export type EventQueueSink<K> = (buffer: K[]) => Promise<any>
 
 export interface EventQueue<K> extends Managed {
@@ -85,6 +90,7 @@ export class DefaultEventQueue<K> implements EventQueue<K> {
   // batchComparator is called to determine whether two events can be included
   // together in the same batch
   private batchComparator: (eventA: K, eventB: K) => boolean
+  private started: boolean
 
   constructor({
     flushInterval,
@@ -105,19 +111,28 @@ export class DefaultEventQueue<K> implements EventQueue<K> {
       callback: this.flush.bind(this),
       timeout: flushInterval,
     })
+    this.started = false
   }
 
   start(): void {
+    this.started = true
     // dont start the timer until the first event is enqueued
   }
 
   stop(): Promise<any> {
+    this.started = false
     const result = this.sink(this.buffer)
+    this.buffer = []
     this.timer.stop()
     return result
   }
 
   enqueue(event: K): void {
+    if (!this.started) {
+      logger.warn('Queue is stopped, not accepting event')
+      return
+    }
+
     // If new event cannot be included into the current batch, flush so it can
     // be in its own new batch.
     const bufferedEvent: K | undefined = this.buffer[0]
