Merged PR 710571: refactor connection, session and sender handling during reconnect

* logging all errors related to the session and connection close
* closing the sender properly not only in case of an error

Security-concept-update-needed: false.

JIRA Work Item: STACKITALO-61
This commit is contained in:
Lukas Schmitt 2025-01-03 12:19:11 +00:00 committed by Christian Schaible
parent 8b231d9a7b
commit 6041fd105a

View file

@ -2,13 +2,14 @@ package messaging
import ( import (
"context" "context"
"dev.azure.com/schwarzit/schwarzit.stackit-public/audit-go.git/log"
"errors" "errors"
"fmt" "fmt"
"github.com/Azure/go-amqp"
"strings" "strings"
"sync" "sync"
"time" "time"
"dev.azure.com/schwarzit/schwarzit.stackit-public/audit-go.git/log"
"github.com/Azure/go-amqp"
) )
// Default connection timeout for the AMQP connection // Default connection timeout for the AMQP connection
@ -200,6 +201,11 @@ func (a *AmqpApi) trySend(ctx context.Context, topic string, data []byte, conten
if err != nil { if err != nil {
return err return err
} }
defer func() {
if err := sender.Close(ctx); err != nil {
log.AuditLogger.Error("failed to close session sender", err)
}
}()
bytes := [][]byte{data} bytes := [][]byte{data}
message := amqp.Message{ message := amqp.Message{
@ -216,7 +222,6 @@ func (a *AmqpApi) trySend(ctx context.Context, topic string, data []byte, conten
err = sender.Send(ctx, &message, nil) err = sender.Send(ctx, &message, nil)
if err != nil { if err != nil {
_ = sender.Close(ctx)
return err return err
} }
@ -225,9 +230,7 @@ func (a *AmqpApi) trySend(ctx context.Context, topic string, data []byte, conten
// resetConnection closes the current session and connection and reconnects to the messaging system. // resetConnection closes the current session and connection and reconnects to the messaging system.
func (a *AmqpApi) resetConnection(ctx context.Context) error { func (a *AmqpApi) resetConnection(ctx context.Context) error {
_ = a.session.Close(ctx) if err := a.Close(ctx); err != nil {
err := a.connection.Close()
if err != nil {
log.AuditLogger.Error("failed to close audit messaging connection", err) log.AuditLogger.Error("failed to close audit messaging connection", err)
} }
@ -237,7 +240,5 @@ func (a *AmqpApi) resetConnection(ctx context.Context) error {
// Close implements Api.Close // Close implements Api.Close
func (a *AmqpApi) Close(ctx context.Context) error { func (a *AmqpApi) Close(ctx context.Context) error {
log.AuditLogger.Info("close audit messaging connection") log.AuditLogger.Info("close audit messaging connection")
_ = a.session.Close(ctx) return errors.Join(a.session.Close(ctx), a.connection.Close())
err := a.connection.Close()
return err
} }