Sha256: cb564f685ae9b8778b7dad994421db7e0cd0bdeee6d4b13be1aa54e707b3fda5
Contents?: true
Size: 1.18 KB
Versions: 13
Compression:
Stored size: 1.18 KB
Contents
# frozen_string_literal: true
module Karafka
module Instrumentation
module Callbacks
# Callback that is triggered when oauth token needs to be refreshed.
class OauthbearerTokenRefresh
include Helpers::ConfigImporter.new(
monitor: %i[monitor]
)
# @param bearer [Rdkafka::Consumer, Rdkafka::Admin] given rdkafka instance. It is needed as
# we need to have a reference to call `#oauthbearer_set_token` or
# `#oauthbearer_set_token_failure` upon the event.
def initialize(bearer)
@bearer = bearer
end
# @param _rd_config [Rdkafka::Config]
# @param bearer_name [String] name of the bearer for which we refresh
def call(_rd_config, bearer_name)
return unless @bearer.name == bearer_name
monitor.instrument(
'oauthbearer.token_refresh',
bearer: @bearer,
caller: self
)
rescue StandardError => e
monitor.instrument(
'error.occurred',
caller: self,
type: 'callbacks.oauthbearer_token_refresh.error',
error: e
)
end
end
end
end
end
Version data entries
13 entries across 13 versions & 1 rubygems