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



42
43
44
45
46
47
48
49
50
51
 
# File 'lib/deimos/active_record_consumer.rb', line 42

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

.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,



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

def compacted(val)
  config[:compacted] = val
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.



66
67
68
 
# File 'lib/deimos/active_record_consumer.rb', line 66

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) 


58
59
60
 
# File 'lib/deimos/active_record_consumer.rb', line 58

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