Class: Artery::Subscription

Inherits:
Object
  • Object
show all
Includes:
Synchronization
Defined in:
lib/artery/subscription.rb,
lib/artery/subscription/synchronization.rb,
lib/artery/subscription/incoming_message.rb

Defined Under Namespace

Modules: Synchronization Classes: IncomingMessage

Constant Summary collapse

DEFAULTS = 
{
  synchronize: false,
  synchronize_updates: true,
  representation: Artery.service_name
}.freeze

Constants included from Synchronization

Synchronization::ALIVE_EDGE, Synchronization::HEARTBEAT_INTERVAL

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Synchronization

#receive_all, #receive_updates, #rewind_message_counter!, #synchronization_alive?, #synchronization_in_progress!, #synchronization_in_progress?, #synchronization_page_update!, #synchronization_per_page, #synchronization_scope, #synchronization_transaction, #synchronize!, #synchronize?, #synchronize_updates?, #synchronize_updates_autoenrich?, #synchronize_updates_per_page, #synchronize_updates_scope

Constructor Details

#initialize(model, uri, handler:, **options) ⇒ Subscription

Returns a new instance of Subscription.



18
19
20
21
22
23
24
25
 
# File 'lib/artery/subscription.rb', line 18

def initialize(model, uri, handler:, **options)
  @uri        = uri
  @subscriber = model
  @handler    = handler
  @options    = DEFAULTS.merge(options)

  Artery.add_subscription self
end

Instance Attribute Details

#handler ⇒ Object

Returns the value of attribute handler.



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

def handler
  @handler
end

#options ⇒ Object

Returns the value of attribute options.



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

def options
  @options
end

#subscriber ⇒ Object

Returns the value of attribute subscriber.



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

def subscriber
  @subscriber
end

#uri ⇒ Object

Returns the value of attribute uri.



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

def uri
  @uri
end

Instance Method Details

#client? ⇒ Boolean

Returns:

  • (Boolean) 


48
49
50
 
# File 'lib/artery/subscription.rb', line 48

def client?
  options[:client]
end

#handle(message) ⇒ Object

rubocop:disable Metrics/AbcSize



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/artery/subscription.rb', line 75

def handle(message) # rubocop:disable Metrics/AbcSize
  request_id = message.reply || SecureRandom.hex(8)
  Artery.logger.tagged(request_id) do
    Artery::Instrumentation.instrument(
      :message, stage: :received, route: message.from, data: message.data, request_id: request_id
    )

    info.lock_for_message(message) do
      if !message.from_updates? && synchronization_in_progress?
        Artery::Instrumentation.instrument(:message, stage: :skipped, reason: 'sync in progress')
        return
      end
      return if !message.from_updates? && !validate_index(message)

      if message.update_by_us?
        Artery::Instrumentation.instrument(:message, stage: :skipped, reason: 'update by us')
        update_info_by_message!(message)
        return
      end

      unless handler.has_block?(message.action) || handler.has_block?(:_default)
        Artery::Instrumentation.instrument(:message, stage: :skipped, reason: 'no listener for action')
        update_info_by_message!(message)
        return
      end

      Artery::Instrumentation.instrument(:message, stage: :handled, route: message.from, request_id: request_id) do
        case message.action
        when :create, :update
          message.enrich_data do |attributes|
            handle_data(message, attributes)
          end
        else
          handle_data(message)
        end
      end
    end
  end
end

#info ⇒ Object 



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

def info
  @info ||= Artery.subscription_info_class.find_for_subscription(self)
end

#latest_message_index ⇒ Object 



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

def latest_message_index
  info.latest_index.to_i
end

#latest_outgoing_message_index ⇒ Object 



56
57
58
59
60
 
# File 'lib/artery/subscription.rb', line 56

def latest_outgoing_message_index
  return unless source?

  Artery.message_class.latest_index(@subscriber.artery_model_name)
end

#new? ⇒ Boolean

Returns:

  • (Boolean) 


62
63
64
 
# File 'lib/artery/subscription.rb', line 62

def new?
  !latest_message_index.positive?
end

#representation_name ⇒ Object 



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

def representation_name
  options[:representation]
end

#reset_info! ⇒ Object 



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

def reset_info!
  @info = nil
  self
end

#rewindable? ⇒ Boolean

Returns:

  • (Boolean) 


52
53
54
 
# File 'lib/artery/subscription.rb', line 52

def rewindable?
  client? && synchronize_updates?
end

#source? ⇒ Boolean

Returns:

  • (Boolean) 


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

def source?
  @subscriber.artery[:source]
end

#update_info_by_message!(message) ⇒ Object 



66
67
68
69
70
71
72
73
 
# File 'lib/artery/subscription.rb', line 66

def update_info_by_message!(message)
  return if !message.has_index? || message.from_updates?

  new_data = {}
  new_data[:latest_index] = message.index if message.index.positive? && (message.index > latest_message_index)

  info.update! new_data
end