uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 18:16:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

Update AMQP message processing loop

Browse Source 
Previously, msg.Ack(true) was used, which is mostly used for batch
processing. It mistakenly acknowledged all previously unacknowledged
messages, causing errors from the AMQP server. Now, messages are
acknowledged individually after each one is processed.

Messages that ERs failed to process are now rejected and requeued
for future processing attempts.

The reader is now closed immediately if the message delivery
channel closes. Therefore, it prevents an endless loop by avoiding
continuous consumption from empty or closed channels.

Addresses: #4146
This commit is contained in:
ionutboangiu
2023-10-16 09:53:39 -04:00
committed by Dan Christian Bogos
parent d7aabe20e3
commit 55c3730cda
1 changed files with 31 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
ers/amqp.go
View File

@@ -158,21 +158,47 @@ func (rdr *AMQPER) readLoop(msgChan <-chan amqp.Delivery) {
utils.ERs, rdr.dialURL))
rdr.close()
return
case msg := <-msgChan:
case msg, ok := <-msgChan:
if !ok {
utils.Logger.Warning(
fmt.Sprintf("<%s> lost connection to AMQP server at %s, closing reader...",
utils.ERs, rdr.dialURL))
rdr.close()
return
}
go func(msg amqp.Delivery) {
if err := rdr.processMessage(msg.Body); err != nil {
err := rdr.processMessage(msg.Body)
if err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> processing message %s error: %s",
utils.ERs, msg.MessageId, err.Error()))
err = msg.Reject(true)
if err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error negatively acknowledging message %s: %s",
utils.ERs, msg.MessageId, err.Error()))
}
return
}
if rdr.poster != nil { // post it
if err := ees.ExportWithAttempts(rdr.poster, msg.Body, utils.EmptyString); err != nil {
// Post the message if poster is available.
if rdr.poster != nil {
err = ees.ExportWithAttempts(rdr.poster, msg.Body, utils.EmptyString)
if err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> writing message %s error: %s",
utils.ERs, msg.MessageId, err.Error()))
}
}
msg.Ack(true)
err = msg.Ack(false)
if err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error acknowledging message %s: %s",
utils.ERs, msg.MessageId, err.Error()))
}
if rdr.Config().ConcurrentReqs != -1 {
rdr.cap <- struct{}{}
}