Class: Karafka::Pro::ScheduledMessages::Deserializers::Headers

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/pro/scheduled_messages/deserializers/headers.rb

Overview

Converts certain pieces of headers into their integer form for messages

Instance Method Summary collapse

Instance Method Details

#call(metadata) ⇒ Hash

Returns headers.

Parameters:

  • metadata (Karafka::aMessages::Metadata)

Returns:

  • (Hash) —

    headers



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/karafka/pro/scheduled_messages/deserializers/headers.rb', line 23

def call(metadata)
  raw_headers = metadata.raw_headers

  type = raw_headers.fetch('schedule_source_type')

  # tombstone and cancellation events are not operable, thus we do not have to cast any
  # of the headers pieces
  return raw_headers unless type == 'schedule'

  headers = raw_headers.dup
  headers['schedule_target_epoch'] = headers['schedule_target_epoch'].to_i

  # This attribute is optional, this is why we have to check for its existence
  if headers.key?('schedule_target_partition')
    headers['schedule_target_partition'] = headers['schedule_target_partition'].to_i
  end

  headers
end