watercc

package main

import (
	"context"
	"fmt"
	"runtime"
	"time"

	"github.com/peterdev80/watercc"

	"github.com/ThreeDotsLabs/watermill/message"
)

// amqpURI описывает адрес для подключения к RabbitMQ.
var amqpURI = "amqp://guest:guest@localhost:5672/"

func main() {
	// инициализируем подписку на сообщения
	subscriber, err := watercc.NewSubscriber(amqpURI, "a1.a2.a3", false)
	if err != nil {
		panic(err)
	}

	// инициализируем обработку сообщений
	router, err := watercc.NewRouter()
	if err != nil {
		panic(err)
	}

	// регистрируем обработчик сообщений
	router.AddNoPublisherHandler(
		"my_handler", // уникальное название обработчика сообщений
		"new_topic",  // название темы для чтения сообщений (совпадает с темой в примере публикации)
		subscriber,   // подписка на приём сообщений
		// обработчик входящих сообщений, публикующий их в консоль
		func(msg *message.Message) error {
			fmt.Printf("-> [%s] %q\n", msg.UUID, msg.Payload)

			return nil
		},
	)

	// инициализируем контекст для обработки входящих запросов
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*6)
	defer cancel()

	// запускаем обработку входящих запросов в отдельном потоке
	go func() {
		if err := router.Run(ctx); err != nil {
			panic(err)
		}
	}()

	runtime.Gosched() // позволяем запуститься потоку

	// инициализируем новый публикатор сообщений
	publisher, err := watercc.NewPublisher(amqpURI, "a1.a2.a3", false)
	if err != nil {
		panic(err)
	}
	defer publisher.Close()

	// публикуем тестовые сообщения с небольшой задержкой
	for messageCounter := 1; messageCounter < 5; messageCounter++ {
		// задержка перед публикацией
		time.Sleep(time.Second * 1)

		// формируем сообщение
		id := fmt.Sprintf("msg:%02d", messageCounter)            // генерируем идентификатор сообщения
		body := fmt.Sprintf("Hello, number %d!", messageCounter) // формируем тест сообщения
		msg := message.NewMessage(id, []byte(body))              // инициализируем сообщение

		// и публикуем его с использованием темы "new_topic"
		if err := publisher.Publish("new_topic", msg); err != nil {
			panic(err)
		}
	}

	<-ctx.Done() // ожидаем завершения обработки

}

Output:
-> [msg:01] "Hello, number 1!"
-> [msg:02] "Hello, number 2!"
-> [msg:03] "Hello, number 3!"
-> [msg:04] "Hello, number 4!"