Consume goroutine in some cases never process channel messages

[pricelessrabbit]

Hi all. I found a very strange behaviour in export service if adding multiple routes. Seems that sometimes the workers goroutines remains waiting also if there is data in the channel. This issue appears randomly, but i managed to reproduce it with workers = 1 and GOMAXPROCS= 1 (but the issue can appear also with default settings). This is my setup

[[routes]]
mqtt_topic = "channels/8f99dca5-7e83-4d57-bc4f-c3356d4e73d7/messages"
nats_topic = "measures"
subtopic = ""
type = "default"
workers = 1

[[routes]]
mqtt_topic = "channels/8f99dca5-7e83-4d57-bc4f-c3356d4e73d7/messages"
nats_topic = "measures2"
subtopic = ""
type = "default"
workers = 1

i also have an external process that publishes on nats "measures" topic.

• if i remove the second route, it works

• if i switch the 2 routes, it works

• if the first route defined is the "measures" one , then the r.Consume() worker goroutine remains waiting on the for msg := range r.Messages, and never process messages also if the channel is full / partially full

Seems a strange concurrency issue, but i cannot understand it. Tried some workarounds to better understand the issue that works:

• change the for := range with a select and added a default case with a sleep. In that case it works
• subscribe to nats with a callback instead of channel, and manually insert values in the channel:

for _, r := range e.consumers {
        //_, err := nc.ChanQueueSubscribe(r.NatsTopic, exportGroup, r.Messages)
        _, err := nc.QueueSubscribe(r.NatsTopic, exportGroup,func(msg *nats.Msg){
            r.Messages <- msg
        })
        if err != nil {
            e.logger.Error(fmt.Sprintf("Failed to subscribe to NATS %s: %s", r.NatsTopic, err))
        }
        for i := 0; i < r.Workers; i++ {
            go r.Consume()
        }
    }