Class: SolidObjects::BroadcastExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_objects/broadcast_executor.rb,
sig/generated/lib/solid_objects/broadcast_executor.rbs

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(process_registry: ProcessRegistry.new, database_adapter: SolidObjects.database_adapter) ⇒ BroadcastExecutor

Returns a new instance of BroadcastExecutor.

RBS:

  • (?process_registry: ProcessRegistry, ?database_adapter: DatabaseAdapter) -> void

Parameters:

  • process_registry: (ProcessRegistry) (defaults to: ProcessRegistry.new)
  • database_adapter: (DatabaseAdapter) (defaults to: SolidObjects.database_adapter) 


14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/solid_objects/broadcast_executor.rb', line 14

def initialize(
  process_registry: ProcessRegistry.new,
  database_adapter: SolidObjects.database_adapter
)
  @process_registry = process_registry
  @database_adapter = database_adapter
  process_registry.register(kind: "broadcast")
  @stopped = false
  @shutdown_requested = false
  @polling_backoff = PollingBackoff.new(
    minimum_interval: SolidObjects.configuration.polling_interval,
    maximum_interval: SolidObjects.configuration.idle_polling_interval,
    on_change: ->(transition) do
      SolidObjects.instrument(
        :"polling.interval_changed",
        role: "broadcasts",
        **transition
      )
    end
  )
end

Instance Attribute Details

#database_adapter ⇒ Object (readonly)

Returns the value of attribute database_adapter.

Returns:

  • (Object) 


108
109
110
 
# File 'lib/solid_objects/broadcast_executor.rb', line 108

def database_adapter
  @database_adapter
end

#polling_backoff ⇒ Object (readonly)

Returns the value of attribute polling_backoff.

Returns:

  • (Object) 


108
109
110
 
# File 'lib/solid_objects/broadcast_executor.rb', line 108

def polling_backoff
  @polling_backoff
end

#process_registry ⇒ Object (readonly)

Returns the value of attribute process_registry.

Returns:

  • (Object) 


108
109
110
 
# File 'lib/solid_objects/broadcast_executor.rb', line 108

def process_registry
  @process_registry
end

Instance Method Details

#broadcast_adapter ⇒ Proc

RBS:

  • () -> Proc | ActionCableBroadcastAdapter

Returns:

  • (Proc) 


133
134
135
136
 
# File 'lib/solid_objects/broadcast_executor.rb', line 133

def broadcast_adapter
  SolidObjects.configuration.broadcast_adapter ||
    ActionCableBroadcastAdapter.new
end

#claim_next ⇒ Broadcast?

RBS:

  • () -> Broadcast?

Returns:



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/solid_objects/broadcast_executor.rb', line 111

def claim_next
  database_adapter.transaction do
    now = database_adapter.database_now
    stale_at = now - SolidObjects.configuration.process_alive_threshold
    relation = Broadcast
      .where(status: "pending", available_at: ..now)
      .or(Broadcast.where(status: "processing", claimed_at: ..stale_at))
      .order(:available_at, :id)
    broadcast = database_adapter.lock_candidates(relation).first
    next unless broadcast

    broadcast.update!(
      status: "processing",
      attempt_count: broadcast.attempt_count + 1,
      claimed_by: process_registry.process_record.id,
      claimed_at: now
    )
    broadcast
  end
end

#complete(broadcast) ⇒ void

This method returns an undefined value.

RBS:

  • (Broadcast) -> void

Parameters:



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/solid_objects/broadcast_executor.rb', line 139

def complete(broadcast)
  database_adapter.transaction do
    locked_broadcast = Broadcast.lock.find(broadcast.id)
    verify_claim!(locked_broadcast)
    locked_broadcast.update!(
      status: "delivered",
      error: nil,
      delivered_at: database_adapter.database_now,
      claimed_by: nil,
      claimed_at: nil
    )
  end
  SolidObjects.instrument(
    :"broadcast.delivered",
    broadcast_id: broadcast.broadcast_id,
    message_id: broadcast.message_id,
    actor_type: broadcast.instance.actor_type,
    actor_id: broadcast.instance.actor_id,
    observable_name: broadcast.observable_name,
    attempt: broadcast.attempt_count
  )
end

#current_polling_interval ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


102
103
104
 
# File 'lib/solid_objects/broadcast_executor.rb', line 102

def current_polling_interval
  polling_backoff.current_interval
end

#fail_broadcast(broadcast, error) ⇒ void

This method returns an undefined value.

RBS:

  • (Broadcast, Exception) -> void

Parameters:



163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/solid_objects/broadcast_executor.rb', line 163

def fail_broadcast(broadcast, error)
  database_adapter.transaction do
    locked_broadcast = Broadcast.lock.find(broadcast.id)
    verify_claim!(locked_broadcast)
    dead = locked_broadcast.attempt_count >= SolidObjects.configuration.max_attempts
    locked_broadcast.update!(
      status: dead ? "dead" : "pending",
      available_at: database_adapter.database_now +
        SolidObjects.configuration.retry_delay.call(locked_broadcast.attempt_count),
      error: {
        "class" => error.class.name,
        "message" => error.message.to_s.byteslice(0, 8_192),
        "backtrace" => Array(error.backtrace).first(50)
      },
      claimed_by: nil,
      claimed_at: nil
    )
  end
rescue ActiveRecord::RecordNotFound, LostActivation
  nil
end

#request_shutdown ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



86
87
88
89
 
# File 'lib/solid_objects/broadcast_executor.rb', line 86

def request_shutdown
  @shutdown_requested = true
  SolidObjects.wake_up.signal
end

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/solid_objects/broadcast_executor.rb', line 62

def run
  ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up

  until shutdown_requested?
    wake_up = SolidObjects.wake_up
    watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up
    worked = run_once
    if worked
      polling_backoff.reset(:work)
      next
    end

    notified = watch.wait(timeout: current_polling_interval)
    if notified == false
      polling_backoff.record_idle
    else
      polling_backoff.reset(:wake_up)
    end
  end
ensure
  stop
end

#run_once ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/solid_objects/broadcast_executor.rb', line 37

def run_once
  return false if stopped?

  process_registry.heartbeat
  broadcast = claim_next
  return false unless broadcast

  broadcast_adapter.call(broadcast)
  complete(broadcast)
  true
rescue => error
  fail_broadcast(broadcast, error) if broadcast
  false
end

#shutdown_requested? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


97
98
99
 
# File 'lib/solid_objects/broadcast_executor.rb', line 97

def shutdown_requested?
  @shutdown_requested
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



53
54
55
56
57
58
59
 
# File 'lib/solid_objects/broadcast_executor.rb', line 53

def stop
  return if stopped?

  @stopped = true
  process_registry.start_draining
  process_registry.stop
end

#stopped? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


92
93
94
 
# File 'lib/solid_objects/broadcast_executor.rb', line 92

def stopped?
  @stopped
end

#verify_claim!(broadcast) ⇒ void

This method returns an undefined value.

RBS:

  • (Broadcast) -> void

Parameters:



186
187
188
189
190
191
 
# File 'lib/solid_objects/broadcast_executor.rb', line 186

def verify_claim!(broadcast)
  return if broadcast.status == "processing" &&
    broadcast.claimed_by == process_registry.process_record.id

  raise LostActivation, "broadcast claim changed"
end