git-workflow/autogits
SHA256
21
2
Fork 14
Code Issues 48 Pull Requests 14 Actions Projects 2 Activity

Fix crash on connection #61

Merged
adamm merged 1 commits from adrianSuSE/autogits:fix_crash into main 2025-08-12 11:56:27 +02:00
Conversation 0 Commits 1 Files Changed 1 +12 -12
1 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
common/rabbitmq.go
View File

@@ -86,38 +86,38 @@ func (l *RabbitConnection) ProcessRabbitMQ(msgCh chan<- RabbitMessage) error {
}
defer connection.Close()
ch, err := connection.Channel()
l.ch, err = connection.Channel()
if err != nil {
return fmt.Errorf("Cannot create a channel. Err: %w", err)
}
defer ch.Close()
defer l.ch.Close()
if err = ch.ExchangeDeclarePassive("pubsub", "topic", true, false, false, false, nil); err != nil {
if err = l.ch.ExchangeDeclarePassive("pubsub", "topic", true, false, false, false, nil); err != nil {
return fmt.Errorf("Cannot find pubsub exchange? Err: %w", err)
}
var q rabbitmq.Queue
if len(queueName) == 0 {
q, err = ch.QueueDeclare("", false, true, true, false, nil)
q, err = l.ch.QueueDeclare("", false, true, true, false, nil)
} else {
q, err = ch.QueueDeclarePassive(queueName, true, false, true, false, nil)
q, err = l.ch.QueueDeclarePassive(queueName, true, false, true, false, nil)
if err != nil {
LogInfo("queue not found .. trying to create it:", err)
if ch.IsClosed() {
ch, err = connection.Channel()
if l.ch.IsClosed() {
l.ch, err = connection.Channel()
if err != nil {
return fmt.Errorf("Channel cannot be re-opened. Err: %w", err)
}
}
q, err = ch.QueueDeclare(queueName, true, false, true, false, nil)
q, err = l.ch.QueueDeclare(queueName, true, false, true, false, nil)
if err != nil {
LogInfo("can't create persistent queue ... falling back to temporaty queue:", err)
if ch.IsClosed() {
ch, err = connection.Channel()
if l.ch.IsClosed() {
l.ch, err = connection.Channel()
return fmt.Errorf("Channel cannot be re-opened. Err: %w", err)
}
q, err = ch.QueueDeclare("", false, true, true, false, nil)
q, err = l.ch.QueueDeclare("", false, true, true, false, nil)
}
}
}
@@ -135,7 +135,7 @@ func (l *RabbitConnection) ProcessRabbitMQ(msgCh chan<- RabbitMessage) error {
l.topicSubChanges <- "+" + topic
}
msgs, err := ch.Consume(q.Name, "", true, true, false, false, nil)
msgs, err := l.ch.Consume(q.Name, "", true, true, false, false, nil)
if err != nil {
return fmt.Errorf("Cannot start consumer. Err: %w", err)
}