Class: Deimos::ActiveRecordConsumer

Inherits:
Consumer
  • Object
show all
Includes:
Deimos::ActiveRecordConsume::BatchConsumption, Deimos::ActiveRecordConsume::MessageConsumption
Defined in:
lib/deimos/active_record_consumer.rb

Overview

To configure batch vs. message mode, change the delivery mode of your Phobos listener. Message-by-message -> use ‘delivery: message` or `delivery: batch` Batch -> use `delivery: inline_batch`

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Deimos::ActiveRecordConsume::BatchConsumption

#consume_batch, #record_key

Methods included from Deimos::ActiveRecordConsume::MessageConsumption

#assign_key, #consume, #destroy_record, #fetch_record, #save_record

Methods inherited from Consumer

#decode_key, #decode_message, decoder, key_decoder

Methods included from Consume::BatchConsumption

#around_consume_batch, #consume_batch

Methods included from Consume::MessageConsumption

#around_consume, #consume

Constructor Details

#initialize ⇒ ActiveRecordConsumer

Setup



60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/deimos/active_record_consumer.rb', line 60

def initialize
  @klass = self.class.config[:record_class]
  @association_list = self.class.config[:association_list]
  @bulk_import_id_column = self.class.config[:bulk_import_id_column] || :bulk_import_id
  @converter = ActiveRecordConsume::SchemaModelConverter.new(self.class.decoder, @klass)

  if self.class.config[:key_schema]
    @key_converter = ActiveRecordConsume::SchemaModelConverter.new(self.class.key_decoder, @klass)
  end

  @compacted = self.class.config[:compacted] != false
  @max_db_batch_size = self.class.config[:max_db_batch_size]
end

Class Method Details

.association_list(associations) ⇒ void

This method returns an undefined value.

should save in addition to @klass

Parameters:

  • associations (List<String>) —

    Optional list of associations that the consumer



36
37
38
 
# File 'lib/deimos/active_record_consumer.rb', line 36

def association_list(associations)
  config[:association_list] = Array(associations)
end

.bulk_import_id_column(name) ⇒ Object

Parameters:



41
42
43
 
# File 'lib/deimos/active_record_consumer.rb', line 41

def bulk_import_id_column(name)
  config[:bulk_import_id_column] = name
end

.compacted(val) ⇒ void

This method returns an undefined value.

only the last message for each unique key in a batch is processed.

Parameters:

  • val (Boolean) —

    Turn pre-compaction of the batch on or off. If true,



48
49
50
 
# File 'lib/deimos/active_record_consumer.rb', line 48

def compacted(val)
  config[:compacted] = val
end

.max_db_batch_size(limit) ⇒ void

This method returns an undefined value.

Parameters:

  • limit (Integer) —

    Maximum number of transactions in a single database call.



54
55
56
 
# File 'lib/deimos/active_record_consumer.rb', line 54

def max_db_batch_size(limit)
  config[:max_db_batch_size] = limit
end

.record_class(klass) ⇒ void

This method returns an undefined value.

database.

Parameters:

  • klass (Class<ActiveRecord::Base>) —

    the class used to save to the



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

def record_class(klass)
  config[:record_class] = klass
end

Instance Method Details

#process_message?(_payload) ⇒ Boolean

Override this message to conditionally save records

Parameters:

Returns:

  • (Boolean) —

    if true, record is created/update. If false, record processing is skipped but message offset is still committed.



87
88
89
 
# File 'lib/deimos/active_record_consumer.rb', line 87

def process_message?(_payload)
  true
end

#record_attributes(payload, _key = nil) ⇒ Hash

Override this method (with ‘super`) if you want to add/change the default attributes set to the new/existing record.

Parameters:

Returns:

  • (Hash) 


79
80
81
 
# File 'lib/deimos/active_record_consumer.rb', line 79

def record_attributes(payload, _key=nil)
  @converter.convert(payload)
end