mt-rabbit/handler.go

package rabbit

import (
	"context"
	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
}

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

func NewClient(addr, queueName string, opts ...Option) *Client {
	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),
	}

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

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

	go client.handleReconnect(addr)

	return &client
}

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

func NewConsumer(ctx context.Context, client *Client, chanLen int) Consumer {
	return &consumeHandler{ctx: ctx, client: client, chanLen: chanLen}
}