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

logger fix

Browse source
This commit is contained in:
nquidox 2026-04-08 12:05:39 +03:00
parent a2862350b3
commit 154320d288
3 changed files with 18 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

21
consumer.go
View file

@ -4,7 +4,6 @@ import (
"context" "context"
amqp "github.com/rabbitmq/amqp091-go" amqp "github.com/rabbitmq/amqp091-go"
"golang.org/x/time/rate" "golang.org/x/time/rate"
"log"
"time" "time"
) )
@ -31,7 +30,7 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
deliveries, err := client.consume() deliveries, err := client.consume()
if err != nil { if err != nil {
log.Printf("Could not start consuming: %s\n", err) client.logger.Printf("Could not start consuming: %s\n", err)
return return
} }
@ -55,18 +54,18 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
case <-runCtx.Done(): case <-runCtx.Done():
err = client.Close() err = client.Close()
if err != nil { if err != nil {
log.Printf("Close failed: %s\n", err) client.logger.Printf("Close failed: %s\n", err)
} }
return return
case amqErr := <-chClosedCh: case amqErr := <-chClosedCh:
log.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr) client.logger.Printf("AMQP Channel closed due to: %s Reconnecting...\n", amqErr)
reconnectTimer.Reset(time.Second) reconnectTimer.Reset(time.Second)
case <-reconnectTimer.C: case <-reconnectTimer.C:
deliveries, err = client.consume() deliveries, err = client.consume()
if err != nil { if err != nil {
log.Println("Error trying to consume, will try again. Retry in 5 seconds.") client.logger.Println("Error trying to consume, will try again. Retry in 5 seconds.")
reconnectTimer.Reset(time.Second * 5) reconnectTimer.Reset(time.Second * 5)
continue continue
} }
@ -76,31 +75,31 @@ func runConsumer(ctx context.Context, client *Client, msgCh chan []byte) {
case delivery, ok := <-deliveries: case delivery, ok := <-deliveries:
if !ok { if !ok {
log.Println("Deliveries channel closed unexpectedly") client.logger.Println("Deliveries channel closed unexpectedly")
reconnectTimer.Reset(time.Second) reconnectTimer.Reset(time.Second)
continue continue
} }
if err = limiter.Wait(runCtx); err != nil { if err = limiter.Wait(runCtx); err != nil {
log.Printf("Wait limiter failed: %s\n", err) client.logger.Printf("Wait limiter failed: %s\n", err)
} }
select { select {
case <-runCtx.Done(): case <-runCtx.Done():
if err = delivery.Nack(false, true); err != nil { if err = delivery.Nack(false, true); err != nil {
log.Printf("Error nacking message: %s\n", err) client.logger.Printf("Error nacking message: %s\n", err)
} }
err = client.Close() err = client.Close()
if err != nil { if err != nil {
log.Printf("Close failed: %s\n", err) client.logger.Printf("Close failed: %s\n", err)
} }
return return
case msgCh <- delivery.Body: case msgCh <- delivery.Body:
log.Printf("Received message: %s\n", delivery.Body) client.logger.Printf("Received message: %s\n", delivery.Body)
if err = delivery.Ack(false); err != nil { if err = delivery.Ack(false); err != nil {
log.Printf("Error acknowledging message: %s\n", err) client.logger.Printf("Error acknowledging message: %s\n", err)
} }
} }
} }

6
handler.go
View file

@ -6,6 +6,7 @@ import (
amqp "github.com/rabbitmq/amqp091-go" amqp "github.com/rabbitmq/amqp091-go"
"io" "io"
"log" "log"
"os"
"sync" "sync"
"time" "time"
) )
@ -36,13 +37,15 @@ type options struct {
} }
func NewClient(address Address, queueName string, opts ...Option) (*Client, error) { func NewClient(address Address, queueName string, opts ...Option) (*Client, error) {
l := log.New(os.Stdout, "", log.LstdFlags)
addr, err := address.makeAddr() addr, err := address.makeAddr()
if err != nil { if err != nil {
return nil, errors.Join(errBadAddr, err) return nil, errors.Join(errBadAddr, err)
} }
if queueName == "" { if queueName == "" {
log.Fatal(errNoQueue) l.Fatal(errNoQueue)
} }
client := Client{ client := Client{
@ -50,6 +53,7 @@ func NewClient(address Address, queueName string, opts ...Option) (*Client, erro
queueName: queueName, queueName: queueName,
done: make(chan bool), done: make(chan bool),
connected: make(chan struct{}), connected: make(chan struct{}),
logger: l,
} }
o := options{ o := options{

6
publisher.go
View file

@ -23,15 +23,15 @@ func (p *pubHandler) Start(ctx context.Context, chanLen uint) chan<- []byte {
case <-ctx.Done(): case <-ctx.Done():
for msg := range ch { for msg := range ch {
if err := p.push(msg); err != nil { if err := p.push(msg); err != nil {
log.Printf("Error publishing message (shutdown): %s", err) p.client.logger.Printf("Error publishing message (shutdown): %s", err)
} }
} }
log.Println("Publisher stopped") p.client.logger.Println("Publisher stopped")
return return
case msg := <-ch: case msg := <-ch:
if err := p.push(msg); err != nil { if err := p.push(msg); err != nil {
log.Printf("Error publishing message: %s", err) p.client.logger.Printf("Error publishing message: %s", err)
} }
} }
} }