merch-tracker-v2/mt-rabbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

added address, logging + time fixes

Browse source
This commit is contained in:
nquidox 2026-04-03 11:22:13 +03:00
parent 44d99a4ba5
commit a2862350b3
1 changed files with 14 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

23
handler.go
View file

@ -1,10 +1,11 @@
package rabbit package rabbit
import ( import (
"errors"
"fmt" "fmt"
amqp "github.com/rabbitmq/amqp091-go" amqp "github.com/rabbitmq/amqp091-go"
"io"
"log" "log"
"os"
"sync" "sync"
"time" "time"
) )
@ -31,11 +32,13 @@ type options struct {
resendDelay time.Duration resendDelay time.Duration
consumerRateLimit time.Duration consumerRateLimit time.Duration
consumerBurstSize int consumerBurstSize int
logger *log.Logger
} }
func NewClient(addr, queueName string, opts ...Option) (*Client, error) { func NewClient(address Address, queueName string, opts ...Option) (*Client, error) {
if addr == "" { addr, err := address.makeAddr()
log.Fatal(errNoAddr) if err != nil {
return nil, errors.Join(errBadAddr, err)
} }
if queueName == "" { if queueName == "" {
@ -44,7 +47,6 @@ func NewClient(addr, queueName string, opts ...Option) (*Client, error) {
client := Client{ client := Client{
mutex: &sync.Mutex{}, mutex: &sync.Mutex{},
logger: log.New(os.Stdout, "", log.LstdFlags),
queueName: queueName, queueName: queueName,
done: make(chan bool), done: make(chan bool),
connected: make(chan struct{}), connected: make(chan struct{}),
@ -52,18 +54,21 @@ func NewClient(addr, queueName string, opts ...Option) (*Client, error) {
o := options{ o := options{
connectTimeout: 15 * time.Second, connectTimeout: 15 * time.Second,
reconnectDelay: 5, reconnectDelay: 5 * time.Second,
reInitDelay: 2, reInitDelay: 2 * time.Second,
resendDelay: 5, resendDelay: 5 * time.Second,
consumerRateLimit: time.Millisecond * 500, consumerRateLimit: time.Millisecond * 500,
consumerBurstSize: 10, consumerBurstSize: 10,
logger: log.New(io.Discard, "", 0),
} }
for _, opt := range opts { for _, opt := range opts {
opt(&o) opt(&o)
} }
if err := client.connectAndSignal(addr, o.connectTimeout); err != nil { client.logger = o.logger
if err = client.connectAndSignal(addr, o.connectTimeout); err != nil {
return nil, fmt.Errorf("failed to connect: %w", err) return nil, fmt.Errorf("failed to connect: %w", err)
} }