Module: Deimos::ActiveRecordConsume::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

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:



25
26
27
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 25

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

#consume(payload, metadata) ⇒ void

This method returns an undefined value.

@param payload — Decoded payloads

@param metadata — Information about batch, including keys.

Parameters:



32
33
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
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 32

def consume(payload, metadata)
  unless self.process_message?(payload)
    Deimos.config.logger.debug(
      message: 'Skipping processing of message',
      payload: payload,
      metadata: metadata
    )
    return
  end

  key = metadata.with_indifferent_access[:key]
  klass = self.class.config[:record_class]
  record = fetch_record(klass, (payload || {}).with_indifferent_access, key)
  if payload.nil?
    destroy_record(record)
    return
  end
  if record.blank?
    record = klass.new
    assign_key(record, payload, key)
  end

  # for backwards compatibility
  # TODO next major release we should deprecate this
  attrs = if self.method(:record_attributes).parameters.size == 2
            record_attributes(payload.with_indifferent_access, key)
          else
            record_attributes(payload.with_indifferent_access)
          end
  # 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) 


80
81
82
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 80

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) 


15
16
17
18
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 15

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) 


70
71
72
73
74
 
# File 'lib/deimos/active_record_consume/message_consumption.rb', line 70

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