Class: Artery::ActiveRecord::SubscriptionInfo

Inherits:
ActiveRecord::Base
  • Object
show all
Defined in:
lib/artery/active_record/subscription_info.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.find_for_subscription(subscription) ⇒ Object 



9
10
11
12
13
14
15
16
 
# File 'lib/artery/active_record/subscription_info.rb', line 9

def find_for_subscription(subscription)
  info = find_or_initialize_by(subscriber: subscription.subscriber.to_s,
                               service: subscription.uri.service,
                               model: subscription.uri.model)

  info.save! if info.new_record?
  info
end

Instance Method Details

#lock_for_message(message, &blk) ⇒ Object 



41
42
43
44
45
46
47
 
# File 'lib/artery/active_record/subscription_info.rb', line 41

def lock_for_message(message, &blk)
  if message.has_index? # only 'indexed' messages should lock
    with_lock(&blk)
  else
    yield
  end
end

#synchronization_transaction(&block) ⇒ Object 



19
20
21
 
# File 'lib/artery/active_record/subscription_info.rb', line 19

def synchronization_transaction(&block)
  with_lock(&block)
end

#with_lock ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/artery/active_record/subscription_info.rb', line 23

def with_lock
  self.class.transaction do
    unless (was_locked = @locked) # prevent double lock to reduce selects
      Artery::Instrumentation.instrument(:lock, state: :waiting, latest_index: latest_index)

      Artery::Instrumentation.instrument(:lock, state: :acquired, latest_index: latest_index) do
        reload lock: true # explicitely reload record
      end

      @locked = true
    end

    yield
  ensure
    @locked = false unless was_locked
  end
end