diff --git a/protocol/amqp/v2/receiver.go b/protocol/amqp/v2/receiver.go index 7304e8c8e..601d99f92 100644 --- a/protocol/amqp/v2/receiver.go +++ b/protocol/amqp/v2/receiver.go @@ -8,6 +8,7 @@ package amqp import ( "context" "io" + "strings" "github.com/Azure/go-amqp" @@ -15,6 +16,8 @@ import ( "github.com/cloudevents/sdk-go/v2/protocol" ) +const serverDown = "session ended by server" + // receiver wraps an amqp.Receiver as a binding.Receiver type receiver struct{ amqp *amqp.Receiver } @@ -24,6 +27,10 @@ func (r *receiver) Receive(ctx context.Context) (binding.Message, error) { if err == ctx.Err() { return nil, io.EOF } + // handle case when server goes down + if strings.HasPrefix(err.Error(), serverDown) { + return nil, io.EOF + } return nil, err }