Skip to content

Commit

Permalink
Use transformers efficiently for extracting info for metrics in Kafka…
Browse files Browse the repository at this point in the history
… Channel (knative#5505)

* use transformers for extracting event type for metrics

* fix

* fix
  • Loading branch information
skonto committed Jun 16, 2021
1 parent d6973ca commit 91c5291
Showing 1 changed file with 3 additions and 3 deletions.
6 changes: 3 additions & 3 deletions pkg/channel/message_dispatcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ type MessageDispatcher interface {
// DispatchMessageWithRetries dispatches an event to a destination over HTTP.
//
// The destination and reply are URLs.
DispatchMessageWithRetries(ctx context.Context, message cloudevents.Message, additionalHeaders nethttp.Header, destination *url.URL, reply *url.URL, deadLetter *url.URL, config *kncloudevents.RetryConfig) (*DispatchExecutionInfo, error)
DispatchMessageWithRetries(ctx context.Context, message cloudevents.Message, additionalHeaders nethttp.Header, destination *url.URL, reply *url.URL, deadLetter *url.URL, config *kncloudevents.RetryConfig, transformers ...binding.Transformer) (*DispatchExecutionInfo, error)
}

// MessageDispatcherImpl is the 'real' MessageDispatcher used everywhere except unit tests.
Expand Down Expand Up @@ -93,7 +93,7 @@ func (d *MessageDispatcherImpl) DispatchMessage(ctx context.Context, message clo
return d.DispatchMessageWithRetries(ctx, message, additionalHeaders, destination, reply, deadLetter, nil)
}

func (d *MessageDispatcherImpl) DispatchMessageWithRetries(ctx context.Context, message cloudevents.Message, additionalHeaders nethttp.Header, destination *url.URL, reply *url.URL, deadLetter *url.URL, retriesConfig *kncloudevents.RetryConfig) (*DispatchExecutionInfo, error) {
func (d *MessageDispatcherImpl) DispatchMessageWithRetries(ctx context.Context, message cloudevents.Message, additionalHeaders nethttp.Header, destination *url.URL, reply *url.URL, deadLetter *url.URL, retriesConfig *kncloudevents.RetryConfig, transformers ...binding.Transformer) (*DispatchExecutionInfo, error) {
// All messages that should be finished at the end of this function
// are placed in this slice
var messagesToFinish []binding.Message
Expand Down Expand Up @@ -154,7 +154,7 @@ func (d *MessageDispatcherImpl) DispatchMessageWithRetries(ctx context.Context,
return dispatchExecutionInfo, nil
}

ctx, responseResponseMessage, _, dispatchExecutionInfo, err := d.executeRequest(ctx, reply, responseMessage, responseAdditionalHeaders, retriesConfig)
ctx, responseResponseMessage, _, dispatchExecutionInfo, err := d.executeRequest(ctx, reply, responseMessage, responseAdditionalHeaders, retriesConfig, transformers...)
if err != nil {
// If DeadLetter is configured, then send original message with knative error extensions
if deadLetter != nil {
Expand Down

0 comments on commit 91c5291

Please sign in to comment.