queue opts
This commit is contained in:
parent
9b9d8a3064
commit
fd495892fa
2 changed files with 29 additions and 11 deletions
17
handler.go
17
handler.go
|
|
@ -13,7 +13,7 @@ import (
|
||||||
|
|
||||||
type Client struct {
|
type Client struct {
|
||||||
mutex *sync.Mutex
|
mutex *sync.Mutex
|
||||||
queueName string
|
queueOptions *QueueOpts
|
||||||
logger *log.Logger
|
logger *log.Logger
|
||||||
connection *amqp.Connection
|
connection *amqp.Connection
|
||||||
Channel *amqp.Channel
|
Channel *amqp.Channel
|
||||||
|
|
@ -36,7 +36,16 @@ type options struct {
|
||||||
logger *log.Logger
|
logger *log.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewClient(address Address, queueName string, opts ...Option) (*Client, error) {
|
type QueueOpts struct {
|
||||||
|
QueueName string
|
||||||
|
Durable bool
|
||||||
|
AutoDelete bool
|
||||||
|
Exclusive bool
|
||||||
|
NoWait bool
|
||||||
|
Args amqp.Table
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewClient(address Address, queueOpts QueueOpts, opts ...Option) (*Client, error) {
|
||||||
l := log.New(os.Stdout, "", log.LstdFlags)
|
l := log.New(os.Stdout, "", log.LstdFlags)
|
||||||
|
|
||||||
addr, err := address.makeAddr()
|
addr, err := address.makeAddr()
|
||||||
|
|
@ -44,13 +53,13 @@ func NewClient(address Address, queueName string, opts ...Option) (*Client, erro
|
||||||
return nil, errors.Join(errBadAddr, err)
|
return nil, errors.Join(errBadAddr, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if queueName == "" {
|
if queueOpts.QueueName == "" {
|
||||||
l.Fatal(errNoQueue)
|
l.Fatal(errNoQueue)
|
||||||
}
|
}
|
||||||
|
|
||||||
client := Client{
|
client := Client{
|
||||||
mutex: &sync.Mutex{},
|
mutex: &sync.Mutex{},
|
||||||
queueName: queueName,
|
queueOptions: &queueOpts,
|
||||||
done: make(chan bool),
|
done: make(chan bool),
|
||||||
connected: make(chan struct{}),
|
connected: make(chan struct{}),
|
||||||
logger: l,
|
logger: l,
|
||||||
|
|
|
||||||
15
service.go
15
service.go
|
|
@ -78,7 +78,16 @@ func (c *Client) init(conn *amqp.Connection) error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
|
//_, err = ch.QueueDeclare(c.queueName, false, false, false, false, nil)
|
||||||
|
_, err = ch.QueueDeclare(
|
||||||
|
c.queueOptions.QueueName,
|
||||||
|
c.queueOptions.Durable,
|
||||||
|
c.queueOptions.AutoDelete,
|
||||||
|
c.queueOptions.Exclusive,
|
||||||
|
c.queueOptions.NoWait,
|
||||||
|
c.queueOptions.Args,
|
||||||
|
)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
@ -120,7 +129,7 @@ func (c *Client) unsafePush(data []byte) error {
|
||||||
return c.Channel.PublishWithContext(
|
return c.Channel.PublishWithContext(
|
||||||
ctx,
|
ctx,
|
||||||
"",
|
"",
|
||||||
c.queueName,
|
c.queueOptions.QueueName,
|
||||||
false,
|
false,
|
||||||
false,
|
false,
|
||||||
amqp.Publishing{
|
amqp.Publishing{
|
||||||
|
|
@ -147,7 +156,7 @@ func (c *Client) consume() (<-chan amqp.Delivery, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.Channel.Consume(
|
return c.Channel.Consume(
|
||||||
c.queueName,
|
c.queueOptions.QueueName,
|
||||||
"",
|
"",
|
||||||
false,
|
false,
|
||||||
false,
|
false,
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue