71 lines
1.5 KiB
Go
71 lines
1.5 KiB
Go
|
package rabbit
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"errors"
|
||
|
|
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
|
||
|
|
reconnectDelay time.Duration
|
||
|
|
reInitDelay time.Duration
|
||
|
|
resendDelay time.Duration
|
||
|
|
}
|
||
|
|
|
||
|
|
//const (
|
||
|
|
// reconnectDelay = 5 * time.Second
|
||
|
|
// reInitDelay = 2 * time.Second
|
||
|
|
// resendDelay = 5 * time.Second
|
||
|
|
//)
|
||
|
|
|
||
|
|
type Deps struct {
|
||
|
|
ReconnectDelay time.Duration
|
||
|
|
ReInitDelay time.Duration
|
||
|
|
ResendDelay time.Duration
|
||
|
|
QueueName string
|
||
|
|
Addr string
|
||
|
|
}
|
||
|
|
|
||
|
|
var (
|
||
|
|
errNotConnected = errors.New("not connected to a server")
|
||
|
|
errAlreadyClosed = errors.New("already closed: not connected to the server")
|
||
|
|
errShutdown = errors.New("client is shutting down")
|
||
|
|
)
|
||
|
|
|
||
|
|
func NewClient(deps Deps) *Client {
|
||
|
|
client := Client{
|
||
|
|
mutex: &sync.Mutex{},
|
||
|
|
logger: log.New(os.Stdout, "", log.LstdFlags),
|
||
|
|
queueName: deps.QueueName,
|
||
|
|
done: make(chan bool),
|
||
|
|
reconnectDelay: deps.ReconnectDelay,
|
||
|
|
reInitDelay: deps.ReInitDelay,
|
||
|
|
resendDelay: deps.ResendDelay,
|
||
|
|
}
|
||
|
|
|
||
|
|
go client.handleReconnect(deps.Addr)
|
||
|
|
|
||
|
|
return &client
|
||
|
|
}
|
||
|
|
|
||
|
|
func StartConsumer(ctx context.Context, client *Client, chanLen int) chan []byte {
|
||
|
|
msgCh := make(chan []byte, chanLen)
|
||
|
|
go runConsumer(ctx, client)
|
||
|
|
|
||
|
|
return msgCh
|
||
|
|
}
|