Class: Deimos::Backends::Db

Inherits:
Base
  • Object
show all
Defined in:
lib/deimos/backends/db.rb

Overview

Backend which saves messages to the database instead of immediately sending them.

Class Method Summary collapse

Methods inherited from Base

publish

Class Method Details

.execute(producer_class:, messages:) ⇒ Object

:nodoc:



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/deimos/backends/db.rb', line 12

def execute(producer_class:, messages:)
  records = messages.map do |m|
    message = Deimos::KafkaMessage.new(
      message: m.encoded_payload ? m.encoded_payload.to_s.b : nil,
      topic: m.topic,
      partition_key: m.partition_key || m.key
    )
    message.key = m.encoded_key.to_s.b unless producer_class.config[:no_keys]
    message
  end
  Deimos::KafkaMessage.import(records)
  Deimos.config.metrics&.increment(
    'db_producer.insert',
    tags: %W(topic:#{producer_class.topic}),
    by: records.size
  )
end