gt-common/rabbitMQ/consumer_test.go

package rabbitMQ

import (
	"testing"
	"time"
)

func TestNewProducter1(t *testing.T) {
	exchange := Exchange{
		Name:      "gt.dc.event",
		Type:      "fanout", // 广播
		QueueName: "step_change",
		Key:       "UF-4001A",
		Dns:       "amqp://admin:devmq*1120@47.96.12.136:5672/",
	}

	got, err := NewProducter(exchange)

	t.Log(got, err)
	for {
		send := got.Publisher("step_change", []byte(`{"device_code":"UF_4001A"}`))

		_ = got.Publisher("step_change", []byte(`{"device_code":"UF_4001B"}`))

		_ = got.Publisher("step_change_a", []byte(`{"device_code":"UF_4001C"}`))
		_ = got.Publisher("step_change_b", []byte(`{"device_code":"UF_4001D"}`))

		t.Log(send)

		time.Sleep(2 * time.Second)
	}

	if got != nil {
		got.Close()
	}
}

func TestConsumer_Subscribe(t *testing.T) {
	exchange := Exchange{
		Name:      "gt.dc.event",
		Type:      "fanout", // 广播
		QueueName: "step_change",
		Key:       "UF-4001A",
		Dns:       "amqp://admin:devmq*1120@47.96.12.136:5672/",
	}
	c, _ := NewConsumer(exchange)

	go TestNewProducter1(t)

	go c.Handler("step_change", func(body []byte) error {
		t.Logf("new message from  A step_change: %s", body)
		return nil
	})

	go c.Handler("step_change", func(body []byte) error {
		t.Logf("new message from B step_change: %s", body)
		return nil
	})

	c.Handler("step_change_b", func(body []byte) error {
		t.Logf("new message from step_change_b: %s", body)
		return nil
	})
}