Class: CycloneLariat::Outbox::Repo::ActiveRecord::Messages

Inherits:
Object
  • Object
show all
Defined in:
lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb

Constant Summary collapse

LIMIT = 
1000

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(dataset) ⇒ Messages

Returns a new instance of Messages.



17
18
19
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 17

def initialize(dataset)
  @dataset = dataset
end

Instance Attribute Details

#dataset ⇒ Object (readonly)

Returns the value of attribute dataset.



15
16
17
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 15

def dataset
  @dataset
end

Instance Method Details

#create(msg) ⇒ Object 



21
22
23
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 21

def create(msg)
  dataset.create(Outbox::Mappers::Messages.to_row(msg)).uuid
end

#delete(uuid) ⇒ Object 



25
26
27
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 25

def delete(uuid)
  dataset.where(uuid: uuid).delete_all
end

#each_with_error ⇒ Object 



33
34
35
36
37
38
39
40
41
42
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 33

def each_with_error
  dataset
    .where('sending_error IS NOT NULL')
    .order(created_at: :asc)
    .limit(LIMIT)
    .each do |row|
      msg = build_message_from_ar_row(row)
      yield(msg)
    end
end

#lock(uuid) ⇒ Object 



48
49
50
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 48

def lock(uuid)
  dataset.lock('FOR UPDATE NOWAIT').where(uuid: uuid)
end

#transaction(&block) ⇒ Object 



44
45
46
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 44

def transaction(&block)
  dataset.transaction(&block)
end

#update_error(uuid, error_message) ⇒ Object 



29
30
31
 
# File 'lib/cyclone_lariat/plugins/outbox/repo/active_record/messages.rb', line 29

def update_error(uuid, error_message)
  dataset.where(uuid: uuid).update(sending_error: error_message)
end