mt-rabbit/handler.go

package rabbit

import (
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
	"os"
	"sync"
	"time"
)

type Client struct {
	mutex           *sync.Mutex
	queueName       string
	logger          *log.Logger
	connection      *amqp.Connection
	Channel         *amqp.Channel
	done            chan bool
	notifyConnClose chan *amqp.Error
	notifyChanClose chan *amqp.Error
	notifyConfirm   chan amqp.Confirmation
	isReady         bool
	opts            options
	connected       chan struct{}
}

type options struct {
	connectTimeout    time.Duration
	reconnectDelay    time.Duration
	reInitDelay       time.Duration
	resendDelay       time.Duration
	consumerRateLimit time.Duration
	consumerBurstSize int
}

func NewClient(addr, queueName string, opts ...Option) (*Client, error) {
	if addr == "" {
		log.Fatal(errNoAddr)
	}

	if queueName == "" {
		log.Fatal(errNoQueue)
	}

	client := Client{
		mutex:     &sync.Mutex{},
		logger:    log.New(os.Stdout, "", log.LstdFlags),
		queueName: queueName,
		done:      make(chan bool),
		connected: make(chan struct{}),
	}

	o := options{
		connectTimeout:    15 * time.Second,
		reconnectDelay:    5,
		reInitDelay:       2,
		resendDelay:       5,
		consumerRateLimit: time.Millisecond * 500,
		consumerBurstSize: 10,
	}

	for _, opt := range opts {
		opt(&o)
	}

	if err := client.connectAndSignal(addr, o.connectTimeout); err != nil {
		return nil, fmt.Errorf("failed to connect: %w", err)
	}

	go client.handleReconnect(addr)

	return &client, nil
}

func NewPublisher(client *Client) Publisher {
	return &pubHandler{client: client}
}

func NewConsumer(client *Client) Consumer {
	return &consumeHandler{client: client}
}

func (c *Client) connectAndSignal(addr string, timeout time.Duration) error {
	type result struct {
		conn *amqp.Connection
		err  error
	}
	resCh := make(chan result, 1)

	go func() {
		conn, err := amqp.Dial(addr)
		resCh <- result{conn, err}
	}()

	select {
	case <-time.After(timeout):
		return fmt.Errorf("connection timeout after %v", timeout)
	case res := <-resCh:
		if res.err != nil {
			return res.err
		}

		c.changeConnection(res.conn)
		if err := c.init(res.conn); err != nil {
			res.conn.Close()
			return fmt.Errorf("init failed: %w", err)
		}

		close(c.connected)
		return nil
	}
}