|
|
@@ -5,6 +5,7 @@ import (
|
|
|
"git.sxidc.com/go-framework/baize/framework/core/infrastructure"
|
|
|
"git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue"
|
|
|
"git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/common"
|
|
|
+ "git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/mqtt"
|
|
|
"git.sxidc.com/go-framework/baize/framework/core/infrastructure/message_queue/redis"
|
|
|
"github.com/pkg/errors"
|
|
|
"sync"
|
|
|
@@ -20,6 +21,16 @@ func TestRedisMessageQueue(t *testing.T) {
|
|
|
testRedisMessageQueue(t, redisMessageQueue)
|
|
|
}
|
|
|
|
|
|
+func TestMqttMessageQueue(t *testing.T) {
|
|
|
+ mqttMessageQueue, err := mqtt.New("localhost:1883", "admin", "mtyzxhc")
|
|
|
+ if err != nil {
|
|
|
+ t.Fatalf("%+v\n", err)
|
|
|
+ }
|
|
|
+ defer mqtt.Destroy(mqttMessageQueue)
|
|
|
+
|
|
|
+ testMqttMessageQueue(t, mqttMessageQueue)
|
|
|
+}
|
|
|
+
|
|
|
func TestMessageQueueInfrastructure(t *testing.T) {
|
|
|
i := infrastructure.NewInfrastructure(infrastructure.Config{
|
|
|
MessageQueueConfig: infrastructure.MessageQueueConfig{
|
|
|
@@ -30,11 +41,17 @@ func TestMessageQueueInfrastructure(t *testing.T) {
|
|
|
MaxLen: 1000,
|
|
|
ConsumerNum: 1,
|
|
|
},
|
|
|
+ Mqtt: &infrastructure.MqttConfig{
|
|
|
+ Address: "localhost:1883",
|
|
|
+ UserName: "admin",
|
|
|
+ Password: "mtyzxhc",
|
|
|
+ },
|
|
|
},
|
|
|
})
|
|
|
defer infrastructure.DestroyInfrastructure(i)
|
|
|
|
|
|
testMessageQueue(t, i.RedisMessageQueue())
|
|
|
+ testMessageQueue(t, i.MqttMessageQueue())
|
|
|
}
|
|
|
|
|
|
func testRedisMessageQueue(t *testing.T, redisMessageQueue *redis.MessageQueue) {
|
|
|
@@ -47,7 +64,7 @@ func testRedisMessageQueue(t *testing.T, redisMessageQueue *redis.MessageQueue)
|
|
|
t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
}
|
|
|
|
|
|
- fmt.Println("test1 consumed")
|
|
|
+ fmt.Println("redis test1 consumed")
|
|
|
|
|
|
wg.Done()
|
|
|
})
|
|
|
@@ -61,7 +78,7 @@ func testRedisMessageQueue(t *testing.T, redisMessageQueue *redis.MessageQueue)
|
|
|
t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
}
|
|
|
|
|
|
- fmt.Println("test2 consumed")
|
|
|
+ fmt.Println("redis test2 consumed")
|
|
|
|
|
|
wg.Done()
|
|
|
})
|
|
|
@@ -77,11 +94,51 @@ func testRedisMessageQueue(t *testing.T, redisMessageQueue *redis.MessageQueue)
|
|
|
wg.Wait()
|
|
|
}
|
|
|
|
|
|
-func testMessageQueue(t *testing.T, redisMessageQueue common.MessageQueue) {
|
|
|
+func testMqttMessageQueue(t *testing.T, mqttMessageQueue *mqtt.MessageQueue) {
|
|
|
+ wg := sync.WaitGroup{}
|
|
|
+ wg.Add(2)
|
|
|
+
|
|
|
+ err := mqttMessageQueue.Subscribe("test1", "test-mqtt",
|
|
|
+ func(queue common.MessageQueue, topic string, data string) {
|
|
|
+ if string(data) != "test-message" {
|
|
|
+ t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
+ }
|
|
|
+
|
|
|
+ fmt.Println("mqtt test1 consumed")
|
|
|
+
|
|
|
+ wg.Done()
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ t.Fatalf("%+v\n", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ err = mqttMessageQueue.Subscribe("test2", "test-mqtt",
|
|
|
+ func(queue common.MessageQueue, topic string, data string) {
|
|
|
+ if string(data) != "test-message" {
|
|
|
+ t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
+ }
|
|
|
+
|
|
|
+ fmt.Println("mqtt test2 consumed")
|
|
|
+
|
|
|
+ wg.Done()
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ t.Fatalf("%+v\n", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ err = mqttMessageQueue.Publish("test-mqtt", "test-message")
|
|
|
+ if err != nil {
|
|
|
+ t.Fatalf("%+v\n", err)
|
|
|
+ }
|
|
|
+
|
|
|
+ wg.Wait()
|
|
|
+}
|
|
|
+
|
|
|
+func testMessageQueue(t *testing.T, messageQueue common.MessageQueue) {
|
|
|
wg := sync.WaitGroup{}
|
|
|
wg.Add(2)
|
|
|
|
|
|
- err := message_queue.Subscribe(redisMessageQueue, "test1", "test-message-queue",
|
|
|
+ err := message_queue.Subscribe(messageQueue, "test1", "test-message-queue",
|
|
|
func(queue common.MessageQueue, topic string, data string) {
|
|
|
if string(data) != "test-message" {
|
|
|
t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
@@ -95,7 +152,7 @@ func testMessageQueue(t *testing.T, redisMessageQueue common.MessageQueue) {
|
|
|
t.Fatalf("%+v\n", err)
|
|
|
}
|
|
|
|
|
|
- err = message_queue.Subscribe(redisMessageQueue, "test2", "test-message-queue",
|
|
|
+ err = message_queue.Subscribe(messageQueue, "test2", "test-message-queue",
|
|
|
func(queue common.MessageQueue, topic string, data string) {
|
|
|
if string(data) != "test-message" {
|
|
|
t.Fatalf("%+v\n", errors.New("消息数据不一致"))
|
|
|
@@ -109,7 +166,7 @@ func testMessageQueue(t *testing.T, redisMessageQueue common.MessageQueue) {
|
|
|
t.Fatalf("%+v\n", err)
|
|
|
}
|
|
|
|
|
|
- err = message_queue.Publish(redisMessageQueue, "test-message-queue", "test-message")
|
|
|
+ err = message_queue.Publish(messageQueue, "test-message-queue", "test-message")
|
|
|
if err != nil {
|
|
|
t.Fatalf("%+v\n", err)
|
|
|
}
|
