go-mq/mq_test.go at main · darwinOrg/go-mq · 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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package dgmq_test

import (
	"log"
	"os"
	"testing"
	"time"

	dgctx "github.com/darwinOrg/go-common/context"
	dgsys "github.com/darwinOrg/go-common/sys"
	dgmq "github.com/darwinOrg/go-mq"
	redisdk "github.com/darwinOrg/go-redis"
)

func TestRedisListAdapter(t *testing.T) {
	redisdk.InitClient("localhost:6379")
	mqAdapter, _ := dgmq.NewMqAdapter(&dgmq.MqAdapterConfig{
		Type:    dgmq.MqAdapterRedisList,
		Timeout: time.Minute,
	})
	defer mqAdapter.Close()

	pubAndSub(mqAdapter, "redis_list_topic")
}

func TestRedisStreamAdapter(t *testing.T) {
	redisdk.InitClient("localhost:6379")
	mqAdapter, _ := dgmq.NewMqAdapter(&dgmq.MqAdapterConfig{
		Type:      dgmq.MqAdapterRedisStream,
		Timeout:   0,
		Group:     "test",
		Consumer:  os.Getenv("HOSTNAME"),
		BatchSize: 10,
	})
	defer mqAdapter.Close()

	pubAndSub(mqAdapter, "redis_stream_topic")
}

func TestNatsAdapter(t *testing.T) {
	mqAdapter, err := dgmq.NewMqAdapter(&dgmq.MqAdapterConfig{
		Type:      dgmq.MqAdapterNats,
		Host:      "localhost",
		Port:      4222,
		Timeout:   time.Second * 5,
		PoolSize:  2,
		Group:     "test",
		BatchSize: 10,
		Username:  "startrek_mq",
		Password:  "cswjggljrmpypwfccarzpjxG-urepqldkhecvnzxzmngotaqs-bkwdvjgipruectqcowoqb6nj",
	})
	if err != nil {
		panic(err)
	}
	defer mqAdapter.Close()

	pubAndSub(mqAdapter, "nats_topic")
}

func pubAndSub(mqAdapter dgmq.MqAdapter, topic string) {
	ctx := dgctx.SimpleDgContext()
	tag1 := "tag1"
	tag2 := "tag2"

	cb1, err := mqAdapter.SubscribeWithTag(ctx, topic, tag1, func(_ *dgctx.DgContext, message string) error {
		log.Print(message)

		return nil
	})
	if err != nil {
		panic(err)
	}

	cb2, err := mqAdapter.SubscribeWithTag(ctx, topic, tag2, func(_ *dgctx.DgContext, message string) error {
		log.Print(message)

		return nil
	})
	if err != nil {
		panic(err)
	}

	cb3, err := mqAdapter.Subscribe(ctx, topic, func(_ *dgctx.DgContext, message string) error {
		log.Print(message)

		return nil
	})
	if err != nil {
		panic(err)
	}

	dc := dgctx.SimpleDgContext()
	_ = mqAdapter.Publish(dc, topic, "hello")
	_ = mqAdapter.Publish(dc, topic, []byte("world"))
	_ = mqAdapter.Publish(dc, topic, map[string]string{"haha": "hehe"})

	dgsys.HangupApplication()
	cb1()
	cb2()
	cb3()
	_ = mqAdapter.Destroy(dc, topic)
}

func delay(mqAdapter dgmq.MqAdapter, topic string) {
	ctx := dgctx.SimpleDgContext()

	cb, err := mqAdapter.SubscribeDelay(ctx, topic, time.Second, func(_ *dgctx.DgContext, message string) error {
		log.Print(message)

		return nil
	})
	if err != nil {
		panic(err)
	}

	dc := dgctx.SimpleDgContext()
	_ = mqAdapter.PublishDelay(dc, topic, map[string]string{"delay": "later"}, 5*time.Second)

	dgsys.HangupApplication()
	cb()
	_ = mqAdapter.Destroy(dc, topic)
}