Class: SolidObjects::Worker

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(process_registry: ProcessRegistry.new) ⇒ Worker

Returns a new instance of Worker.

RBS:

  • (?process_registry: ProcessRegistry) -> void

Parameters:



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

def initialize(process_registry: ProcessRegistry.new)
  @process_registry = process_registry
  process_record = process_registry.register
  @activation_manager = ActivationManager.new(owner_id: process_record.id)
  @activations = {}
  @stopped = false
  @shutdown_requested = false
end

Instance Attribute Details

#activation_manager ⇒ Object (readonly)

Returns the value of attribute activation_manager.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/worker.rb', line 112

def activation_manager
  @activation_manager
end

#activations ⇒ Object (readonly)

Returns the value of attribute activations.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/worker.rb', line 112

def activations
  @activations
end

#process_registry ⇒ Object (readonly)

Returns the value of attribute process_registry.

Returns:

  • (Object) 


112
113
114
 
# File 'lib/solid_objects/worker.rb', line 112

def process_registry
  @process_registry
end

Instance Method Details

#cached_ready_activation ⇒ Activation?

RBS:

  • () -> Activation?

Returns:



115
116
117
 
# File 'lib/solid_objects/worker.rb', line 115

def cached_ready_activation
  activations.each_value.find(&:ready?)
end

#maintain_cached_activations ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/solid_objects/worker.rb', line 120

def maintain_cached_activations
  activations.each_value do |activation|
    if activation.idle?
      release_activation(activation)
    elsif activation.lease_renewal_due?
      activation.renew_lease
    end
  rescue LostActivation
    release_activation(activation)
  end
end

#release_activation(activation) ⇒ void

This method returns an undefined value.

RBS:

  • (Activation) -> void

Parameters:



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

def release_activation(activation)
  activations.delete(activation.lease.instance_id)
  activation.yield_ready_messages if activation.pass_exhausted?
  activation.deactivate
end

#request_shutdown ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



84
85
86
87
 
# File 'lib/solid_objects/worker.rb', line 84

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

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



72
73
74
75
76
77
78
79
80
81
 
# File 'lib/solid_objects/worker.rb', line 72

def run
  until shutdown_requested?
    processed = run_once
    next if processed.positive?

    SolidObjects.wake_up.wait(timeout: SolidObjects.configuration.polling_interval)
  end
ensure
  stop
end

#run_once ⇒ Integer

RBS:

  • () -> Integer

Returns:

  • (Integer) 


28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/solid_objects/worker.rb', line 28

def run_once
  return 0 if stopped?

  process_registry.heartbeat
  maintain_cached_activations
  activation = cached_ready_activation || activation_manager.claim_next
  return 0 unless activation

  activations[activation.lease.instance_id] = activation
  processed = LeaseRenewer.new(
    activation:,
    process_registry:
  ).around { activation.drain }
  release_activation(activation) if activation.pass_exhausted?
  processed
rescue ActorDestroyed
  release_activation(activation) if activation
  0
rescue StateMigrationError, ApplicationWriteForbidden, LostActivation => error
  release_activation(activation) if activation
  SolidObjects.configuration.logger.error(
    event: "solid_objects.worker.error",
    process_id: process_registry.process_record&.id,
    error_class: error.class.name,
    error_message: error.message
  )
  0
end

#run_until_idle(max_passes: 1_000) ⇒ Integer

RBS:

  • (?max_passes: Integer) -> Integer

Parameters:

  • max_passes: (Integer) (defaults to: 1_000)

Returns:

  • (Integer) 


58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/solid_objects/worker.rb', line 58

def run_until_idle(max_passes: 1_000)
  total = 0

  max_passes.times do
    processed = run_once
    break if processed.zero?

    total += processed
  end

  total
end

#shutdown_requested? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


106
107
108
 
# File 'lib/solid_objects/worker.rb', line 106

def shutdown_requested?
  @shutdown_requested
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



90
91
92
93
94
95
96
97
98
 
# File 'lib/solid_objects/worker.rb', line 90

def stop
  return if stopped?

  @stopped = true
  process_registry.start_draining
  activations.each_value(&:deactivate)
  activations.clear
  process_registry.stop
end

#stopped? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


101
102
103
 
# File 'lib/solid_objects/worker.rb', line 101

def stopped?
  @stopped
end