diff --git a/pkg/broker/receiver.go b/pkg/broker/receiver.go index 07f4ef4ccf7..37c213325ec 100644 --- a/pkg/broker/receiver.go +++ b/pkg/broker/receiver.go @@ -126,8 +126,8 @@ func (r *Receiver) serveHTTP(ctx context.Context, event cloudevents.Event, resp // Remove the TTL attribute that is used by the Broker. originalV2 := event.Context.AsV02() - ttl, present := originalV2.Extensions[V02TTLAttribute] - if !present { + ttl, ttlKey := GetTTL(event.Context) + if ttl == nil { // Only messages sent by the Broker should be here. If the attribute isn't here, then the // event wasn't sent by the Broker, so we can drop it. r.logger.Warn("No TTL seen, dropping", zap.Any("triggerRef", triggerRef), zap.Any("event", event)) @@ -136,7 +136,7 @@ func (r *Receiver) serveHTTP(ctx context.Context, event cloudevents.Event, resp // framework returns a 500 to the caller, so the Channel would send this repeatedly. return nil } - delete(originalV2.Extensions, V02TTLAttribute) + delete(originalV2.Extensions, ttlKey) event.Context = originalV2 r.logger.Debug("Received message", zap.Any("triggerRef", triggerRef)) diff --git a/pkg/broker/ttl.go b/pkg/broker/ttl.go index d71f425b8a7..f84e17a6129 100644 --- a/pkg/broker/ttl.go +++ b/pkg/broker/ttl.go @@ -17,6 +17,8 @@ package broker import ( + "strings" + cloudevents "github.com/cloudevents/sdk-go" ) @@ -26,6 +28,18 @@ const ( V02TTLAttribute = "knativebrokerttl" ) +// GetTTL finds the TTL in the EventContext using a case insensitive comparison +// for the key. The second return param, is the case preserved key that matched. +// Depending on the encoding/transport, the extension case could be changed. +func GetTTL(ctx cloudevents.EventContext) (interface{}, string) { + for k, v := range ctx.AsV02().Extensions { + if lower := strings.ToLower(k); lower == V02TTLAttribute { + return v, k + } + } + return nil, V02TTLAttribute +} + // SetTTL sets the TTL into the EventContext. ttl should be a positive integer. func SetTTL(ctx cloudevents.EventContext, ttl interface{}) (cloudevents.EventContext, error) { err := ctx.SetExtension(V02TTLAttribute, ttl)