Module: Artery::Subscription::Synchronization

Extended by:
ActiveSupport::Concern
Included in:
Artery::Subscription
Defined in:
lib/artery/subscription/synchronization.rb

Constant Summary collapse

ALIVE_EDGE = 
2.minutes
HEARTBEAT_INTERVAL = 
30.seconds

Instance Method Summary collapse

Instance Method Details

#receive_all ⇒ Object 



83
84
85
86
87
88
89
90
 
# File 'lib/artery/subscription/synchronization.rb', line 83

def receive_all
  synchronization_in_progress! unless synchronization_in_progress?

  Artery::Instrumentation.instrument(:sync, stage: :receive_all, route: uri.to_route) do
    reset_latest_index!
    while receive_all_once == :continue; end
  end
end

#receive_updates ⇒ Object 



92
93
94
95
96
97
98
 
# File 'lib/artery/subscription/synchronization.rb', line 92

def receive_updates
  synchronization_in_progress!

  Artery::Instrumentation.instrument(:sync, stage: :receive_updates, route: uri.to_route) do
    while receive_updates_once == :continue; end
  end
end

#rewind_message_counter! ⇒ Object

rubocop:disable Naming/PredicateMethod



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/artery/subscription/synchronization.rb', line 100

def rewind_message_counter! # rubocop:disable Naming/PredicateMethod
  return false unless rewindable?

  local_index = latest_message_index
  return false unless local_index.positive?

  remote_index = request_publisher_latest_index!
  return false unless remote_index.positive?
  return false if remote_index >= local_index

  Artery.logger.info(
    "Rewinding #{uri} latest_index from #{local_index} to #{remote_index}"
  )
  info.update!(latest_index: remote_index)
  true
end

#synchronization_alive? ⇒ Boolean

Returns:

  • (Boolean) 


44
45
46
 
# File 'lib/artery/subscription/synchronization.rb', line 44

def synchronization_alive?
  info.synchronization_heartbeat.blank? || (Time.zone.now - info.synchronization_heartbeat) < ALIVE_EDGE
end

#synchronization_in_progress!(val = true) ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/artery/subscription/synchronization.rb', line 48

def synchronization_in_progress!(val = true)
  if val
    Artery.synchronizing_subscriptions << self
    run_synchronization_heartbeat

    info.update! synchronization_in_progress: true, synchronization_heartbeat: Time.zone.now
  else
    Artery.synchronizing_subscriptions.delete self
    stop_synchronization_heartbeat

    info.update! synchronization_in_progress: false, synchronization_heartbeat: nil
  end
end

#synchronization_in_progress? ⇒ Boolean

Returns:

  • (Boolean) 


40
41
42
 
# File 'lib/artery/subscription/synchronization.rb', line 40

def synchronization_in_progress?
  info.synchronization_in_progress? && synchronization_alive?
end

#synchronization_page_update!(page) ⇒ Object 



69
70
71
 
# File 'lib/artery/subscription/synchronization.rb', line 69

def synchronization_page_update!(page)
  info.update! synchronization_page: page
end

#synchronization_per_page ⇒ Object 



23
24
25
 
# File 'lib/artery/subscription/synchronization.rb', line 23

def synchronization_per_page
  options[:synchronize].is_a?(Hash) ? options[:synchronize][:per_page] : nil
end

#synchronization_scope ⇒ Object 



19
20
21
 
# File 'lib/artery/subscription/synchronization.rb', line 19

def synchronization_scope
  options[:synchronize].is_a?(Hash) ? options[:synchronize][:scope] : nil
end

#synchronization_transaction(&blk) ⇒ Object 



62
63
64
65
66
67
 
# File 'lib/artery/subscription/synchronization.rb', line 62

def synchronization_transaction(&blk)
  return unless blk
  return info.synchronization_transaction(&blk) if info.respond_to?(:synchronization_transaction)

  blk.call
end

#synchronize! ⇒ Object 



73
74
75
76
77
78
79
80
81
 
# File 'lib/artery/subscription/synchronization.rb', line 73

def synchronize!
  return if uri.service == Artery.service_name || synchronization_in_progress?

  if !new? && synchronize_updates?
    receive_updates
  elsif new? && synchronize?
    receive_all
  end
end

#synchronize? ⇒ Boolean

Returns:

  • (Boolean) 


11
12
13
 
# File 'lib/artery/subscription/synchronization.rb', line 11

def synchronize?
  options[:synchronize]
end

#synchronize_updates? ⇒ Boolean

Returns:

  • (Boolean) 


15
16
17
 
# File 'lib/artery/subscription/synchronization.rb', line 15

def synchronize_updates?
  options[:synchronize_updates]
end

#synchronize_updates_autoenrich? ⇒ Boolean

Returns:

  • (Boolean) 


36
37
38
 
# File 'lib/artery/subscription/synchronization.rb', line 36

def synchronize_updates_autoenrich?
  options[:synchronize_updates].is_a?(Hash) ? options[:synchronize_updates][:autoenrich] : false
end

#synchronize_updates_per_page ⇒ Object 



31
32
33
34
 
# File 'lib/artery/subscription/synchronization.rb', line 31

def synchronize_updates_per_page
  (options[:synchronize_updates].is_a?(Hash) && options[:synchronize_updates][:per_page]) ||
    synchronization_per_page
end

#synchronize_updates_scope ⇒ Object 



27
28
29
 
# File 'lib/artery/subscription/synchronization.rb', line 27

def synchronize_updates_scope
  (options[:synchronize_updates].is_a?(Hash) && options[:synchronize_updates][:scope]) || synchronization_scope
end