optimizely · mjc1283 · Aug 29, 2019 · Aug 27, 2019 · Aug 29, 2019
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]