Module: Deimos::ActiveRecordConsume::MessageConsumption

Includes:
Consume::MessageConsumption
Included in:
Deimos::ActiveRecordConsumer
Defined in:
lib/deimos/active_record_consume/message_consumption.rb,
sig/defs.rbs

Overview

Methods for consuming individual messages and saving them to the database as ActiveRecord instances.

Instance Method Summary collapse

Methods included from Consume::MessageConsumption

#around_consume

Instance Method Details

#assign_key(record, _payload, key) ⇒ void

This method returns an undefined value.

Assign a key to a new record.

@param record

@param _payload

@param key

Parameters:



29
30
31
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 29

def assign_key(record, _payload, key)
  record[record.class.primary_key] = key
end

#consume ⇒ void

This method returns an undefined value.

@param payload — Decoded payloads

@param metadata — Information about batch, including keys.

Parameters:



1871
 
# File 'sig/defs.rbs', line 1871

def consume: ((::Hash[untyped, untyped] | Deimos::SchemaClass::Record) payload, ::Hash[untyped, untyped] metadata) -> void

#consume_message(message) ⇒ Object

Parameters:

  • message (Karafka::Messages::Message) 


34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 34

def consume_message(message)
  unless self.process_message?(message)
    Deimos::Logging.log_debug(
      message: 'Skipping processing of message',
      payload: message.payload.to_h,
      metadata: Deimos::Logging.metadata_log_text(message.metadata)
    )
    return
  end

  klass = self.class.config[:record_class]
  klass = klass.constantize if klass.is_a?(String)
  payload = message.payload
  if payload.is_a?(Hash) || payload.nil?
    payload = payload.to_h.with_indifferent_access
  end
  record = fetch_record(klass, payload, message.key)
  if delete_record?(message)
    destroy_record(record)
    return
  end
  if record.blank?
    record = klass.new
    assign_key(record, message.payload, message.key)
  end

  payload = message.payload || {}
  payload = payload.with_indifferent_access if payload.is_a?(Hash)
  attrs = record_attributes(payload, message.key)
  # don't use attributes= - bypass Rails < 5 attr_protected
  attrs.each do |k, v|
    record.send("#{k}=", v)
  end
  save_record(record)
end

#destroy_record(record) ⇒ void

This method returns an undefined value.

Destroy a record that received a null payload. Override if you need to do something other than a straight destroy (e.g. mark as archived).

@param record

Parameters:

  • record (ActiveRecord::Base) 


82
83
84
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 82

def destroy_record(record)
  record&.destroy
end

#fetch_record(klass, _payload, key) ⇒ ActiveRecord::Base

Find the record specified by the given payload and key. Default is to use the primary key column and the value of the first field in the key.

@param klass

@param _payload

@param key

Parameters:

Returns:

  • (ActiveRecord::Base) 


19
20
21
22
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 19

def fetch_record(klass, _payload, key)
  fetch_key = key.is_a?(Hash) && key.size == 1 ? key.values.first : key
  klass.unscoped.where(klass.primary_key => fetch_key).first
end

#save_record(record) ⇒ void

This method returns an undefined value.

@param record

Parameters:

  • record (ActiveRecord::Base) 


72
73
74
75
76
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 72

def save_record(record)
  record.created_at ||= Time.zone.now if record.respond_to?(:created_at)
  record.updated_at = Time.zone.now if record.respond_to?(:updated_at)
  record.save!
end