dedup_pipeline.go

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// UQueue is a Slice based list implementation of a Queue
// that ignores enqueue requests for items that are already queued.
type UQueue struct {
	sync.RWMutex
	container []interface{}
}

// NewUQueue creates a UQueue.
func NewUQueue() *UQueue {
	return &UQueue{
		container: make([]interface{}, 0),
	}
}

// Enqueue adds an item at the back of the queue
func (q *UQueue) Enqueue(item interface{}) {
	q.Lock()
	defer q.Unlock()

	for _, e := range q.container {
		if e == item {
			return
		}
	}
	q.container = append(q.container, item)
}

// Dequeue removes and returns the front queue item
func (q *UQueue) Dequeue() interface{} {
	q.Lock()
	defer q.Unlock()

	if len(q.container) > 0 {
		var el interface{}
		el, q.container = q.container[0], q.container[1:]
		return el
	}
	return nil
}

// Size returns the actual Q size
func (q *UQueue) Size() int {
	q.RLock()
	defer q.RUnlock()

	return len(q.container)
}

func generate(q *UQueue, nums ...int) {
	go func() {
		for _, n := range nums {
			r := rand.Intn(500)
			time.Sleep(time.Duration(r) * time.Millisecond)
			fmt.Println("Generator:", n)
			q.Enqueue(n)
		}
	}()
}

func consume(q *UQueue) {
	for q.Size() != 0 {
		n := q.Dequeue()
		fmt.Println("Consumer: Started", n)
		r := rand.Intn(2000)
		time.Sleep(time.Duration(r) * time.Millisecond)
		fmt.Println("Consumer: Completed", n)
	}
}

func main() {
	rand.Seed(time.Now().UTC().UnixNano())

	q := NewUQueue()
	generate(q, 2, 3, 4, 4, 4, 5, 5, 6, 7, 7, 7)
	time.Sleep(time.Duration(500) * time.Millisecond)
	consume(q)
}

dedup_pipeline_test.go

package main

import (
	"reflect"
	"testing"
)

var tests = []struct {
	input    []int // input
	expected []int // expected result
}{
	{[]int{1, 2, 3, 4}, []int{1, 2, 3, 4}},
	{[]int{1, 2, 2, 2, 3, 3, 4}, []int{1, 2, 3, 4}},
	{[]int{1, 1, 2, 2, 2, 3, 3, 4, 4, 4, 4}, []int{1, 2, 3, 4}},
}

func TestQueue(t *testing.T) {
	for _, tt := range tests {
		q := NewUQueue()
		for _, v := range tt.input {
			q.Enqueue(v)
		}
		result := []int{}
		n := q.Dequeue()
		for n != nil {
			result = append(result, n.(int))
			n = q.Dequeue()
		}
		if !reflect.DeepEqual(result, tt.expected) {
			t.Errorf("UQueue(%d): expected %d, actual %d", tt.input, tt.expected, result)
		}
	}
}

Sample output:

Generator: 2
Generator: 3
Consumer: Started 2
Generator: 4
Consumer: Completed 2
Consumer: Started 3
Generator: 4
Generator: 4
Generator: 5
Consumer: Completed 3
Consumer: Started 4
Generator: 5
Generator: 6
Generator: 7
Generator: 7
Generator: 7
Consumer: Completed 4
Consumer: Started 5
Consumer: Completed 5
Consumer: Started 6
Consumer: Completed 6
Consumer: Started 7
Consumer: Completed 7