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

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



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/deimos/active_record_consumer.rb', line 68

def initialize
  @klass = self.class.config[:record_class]
  @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
end

Class Method Details

.bulk_import_id_column ⇒ String?

Returns:

  • (String, nil) 


34
35
36
 
# File 'lib/deimos/active_record_consumer.rb', line 34

def bulk_import_id_column
  config[:bulk_import_id_column]
end

.bulk_import_id_generator ⇒ Proc

Returns:

  • (Proc) 


39
40
41
 
# File 'lib/deimos/active_record_consumer.rb', line 39

def bulk_import_id_generator
  config[:bulk_import_id_generator]
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,



56
57
58
 
# File 'lib/deimos/active_record_consumer.rb', line 56

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.



62
63
64
 
# File 'lib/deimos/active_record_consumer.rb', line 62

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

.replace_associations ⇒ Boolean

Returns:

  • (Boolean) 


44
45
46
 
# File 'lib/deimos/active_record_consumer.rb', line 44

def replace_associations
  config[:replace_associations]
end

.save_associations_first ⇒ Boolean

Returns:

  • (Boolean) 


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

def save_associations_first
  config[:save_associations_first]
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.



92
93
94
 
# File 'lib/deimos/active_record_consumer.rb', line 92

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) 


84
85
86
 
# File 'lib/deimos/active_record_consumer.rb', line 84

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