OrgGo/eventials.goevents
1
0
Fork 0
mirror of https://github.com/eventials/goevents.git synced 2025-04-26 13:48:59 +08:00
Code Issues Projects Releases Wiki Activity

Remove useless code

Browse Source
This commit is contained in:
skrater 2019-05-01 14:08:44 -03:00
parent 8231c14ac1
commit bf9ff86a03
1 changed files with 18 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
amqp/producer.go
View File

@ -30,7 +30,6 @@ type producer struct {
channel *amqplib.Channel channel *amqplib.Channel
notifyConfirm chan amqplib.Confirmation notifyConfirm chan amqplib.Confirmation
notifyChanClose chan *amqplib.Error notifyChanClose chan *amqplib.Error
closeQueue chan bool
config ProducerConfig config ProducerConfig
internalQueue chan message internalQueue chan message
@ -62,7 +61,6 @@ func NewProducerConfig(c messaging.Connection, exchange string, config ProducerC
config: config, config: config,
internalQueue: make(chan message, 2), internalQueue: make(chan message, 2),
exchangeName: exchange, exchangeName: exchange,
closeQueue: make(chan bool),
} }
err := producer.setupTopology() err := producer.setupTopology()
@ -127,10 +125,7 @@ func (p *producer) Close() {
p.wg.Wait() p.wg.Wait()
p.closeQueue <- true
close(p.internalQueue) close(p.internalQueue)
close(p.closeQueue)
p.channel.Close() p.channel.Close()
@ -333,32 +328,27 @@ func (p *producer) isClosed() bool {
} }
func (p *producer) drainInternalQueue() { func (p *producer) drainInternalQueue() {
for { for m := range p.internalQueue {
select { retry := true
case <-p.closeQueue:
return
case m := <-p.internalQueue:
retry := true
for retry { for retry {
// block until confirmation // block until confirmation
err := p.publishMessage(m.msg, m.action) err := p.publishMessage(m.msg, m.action)
if err != nil { if err != nil {
log.WithFields(log.Fields{ log.WithFields(log.Fields{
"action": m.action, "action": m.action,
"body": m.msg.Body, "body": m.msg.Body,
"message_id": m.msg.MessageId, "message_id": m.msg.MessageId,
"error": err, "error": err,
"type": "goevents", "type": "goevents",
"sub_type": "producer", "sub_type": "producer",
}).Error("Error publishing message to the exchange. Retrying...") }).Error("Error publishing message to the exchange. Retrying...")
time.Sleep(p.config.publishInterval) time.Sleep(p.config.publishInterval)
} else { } else {
p.wg.Done() p.wg.Done()
retry = false retry = false
}
} }
} }
} }