topic-queue/mq_test.go at master · guanhg/topic-queue · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
package queue

import (
	"fmt"
	"testing"
	"time"
)

func TestQueue(t *testing.T) {
	count, size := int32(0), 74
	topic := "belike"
	q := NewMQueue()
	q.AddTopic(topic, func(name string, batch batchEntry) error {
		for _, e := range batch.entries {
			i := e.Data.(int32)
			count += i
		}
		return fmt.Errorf("[%d] errHandle", count)
	}, stdHandleErr)

	go q.Consume(topic)

	for i := 0; i < size; i++ {
		q.Produce(topic, Entry{Data: int32(1)})
	}

	timer := time.NewTimer(time.Second * 3)
	<-timer.C

	q.Wait()
	if count != int32(size) {
		t.Errorf("count: %d, expected: %d", count, size)
	}

	topic = "unlike"
	q.AddTopic(topic, func(name string, batch batchEntry) error {
		for _, e := range batch.entries {
			i := e.Data.(int32)
			count -= i
		}
		time.Sleep(25 * time.Millisecond)
		return nil
	}, nil)

	go q.Consume(topic)

	for i := 0; i < size; i++ {
		q.Produce(topic, Entry{Data: int32(1)})
	}

	timer = time.NewTimer(time.Second * 3)
	<-timer.C

	q.Wait()
	if count != 0 {
		t.Errorf("count: %d, expected: %d", count, 0)
	}
}