(feat): add channel queue (#90)

* add channel queues to fm-beta

* update for lint issues

* add processor tests for ChanQueue

* update license header

* fix typo in processor_test

Author: thomaszurkan-optimizely

optimizely/event/chan_queue.go

@@ -0,0 +1,62 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
+package event
+
+// ChanQueue is a go channel based queue that takes things from the channel and puts them in a in memory queue
+type ChanQueue struct {
+	ch chan interface{}
+	messages Queue
+}
+
+// Get returns queue for given count size
+func (i *ChanQueue) Get(count int) []interface{} {
+	return i.messages.Get(count)
+}
+
+// Add appends item to queue
+func (i *ChanQueue) Add(item interface{}) {
+	i.ch <- item
+}
+
+// Remove removes item from queue and returns elements slice
+func (i *ChanQueue) Remove(count int) []interface{} {
+	return i.messages.Remove(count)
+
+}
+
+// Size returns size of queue
+func (i *ChanQueue) Size() int {
+	return i.messages.Size()
+}
+
+// NewChanQueue returns new go channel based queue with given in memory queueSize
+func NewChanQueue(queueSize int) Queue {
+
+	ch := make(chan interface{})
+
+	i := &ChanQueue{ch:ch, messages: NewInMemoryQueue(queueSize)}
+
+	go func() {
+		for item := range i.ch {
+			i.messages.Add(item)
+		}
+	}()
+
+	return i
+}
+

optimizely/event/chan_queue_test.go

@@ -0,0 +1,53 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
+package event
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestChanQueue_Add_Size_Remove(t *testing.T) {
+	q := NewChanQueue(100)
+
+	impression := BuildTestImpressionEvent()
+	conversion := BuildTestConversionEvent()
+
+	q.Add(impression)
+	q.Add(impression)
+	q.Add(conversion)
+
+	time.Sleep(2000 * time.Millisecond)
+
+	items1 := q.Get(2)
+
+	assert.Equal(t, 2, len(items1))
+
+	q.Remove(1)
+
+	items2 := q.Get(1)
+
+	assert.True(t, len(items2) != 0)
+
+	allItems := q.Remove(3)
+
+	assert.True(t,len(allItems) > 0)
+
+	assert.Equal(t, 0, q.Size())
+}

optimizely/event/processor_test.go

@@ -172,3 +172,56 @@ func TestDefaultEventProcessor_ProcessBatchProjectMismatch(t *testing.T) {
 		assert.Equal(t, 2, len(evs.Event.Visitors))
 	}
 }
+
+func TestChanQueueEventProcessor_ProcessImpression(t *testing.T) {
+	processor:= &QueueingEventProcessor{
+		MaxQueueSize:    100,
+		FlushInterval:   100,
+		Q:               NewChanQueue(100),
+		EventDispatcher: &HTTPEventDispatcher{},
+	}
+	processor.BatchSize = 10
+	processor.StartTicker(context.TODO())
+
+	impression := BuildTestImpressionEvent()
+
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(impression)
+
+	time.Sleep(3000 * time.Millisecond)
+
+	assert.NotNil(t, processor.Ticker)
+
+	assert.Equal(t, 0, processor.EventsCount())
+}
+
+func TestChanQueueEventProcessor_ProcessBatch(t *testing.T) {
+	processor := &QueueingEventProcessor{MaxQueueSize: 100, FlushInterval: 100, Q: NewChanQueue(100), EventDispatcher: &MockDispatcher{}}
+	processor.BatchSize = 10
+	processor.StartTicker(context.TODO())
+
+	impression := BuildTestImpressionEvent()
+	conversion := BuildTestConversionEvent()
+
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(conversion)
+	processor.ProcessEvent(conversion)
+
+	time.Sleep(3000 * time.Millisecond)
+
+	assert.NotNil(t, processor.Ticker)
+
+	assert.Equal(t, 0, processor.EventsCount())
+
+	time.Sleep(3000 * time.Millisecond)
+
+	result, ok := (processor.EventDispatcher).(*MockDispatcher)
+
+	if ok {
+		assert.Equal(t, 1, len(result.Events))
+		evs := result.Events[0]
+		assert.True(t, len(evs.Event.Visitors) >= 1)
+	}
+}