Class: Proxy::Dynflow::Runner::Dispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/smart_proxy_dynflow/runner/dispatcher.rb

Defined Under Namespace

Classes: RunnerActor

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(clock, logger) ⇒ Dispatcher

Returns a new instance of Dispatcher.



108
109
110
111
112
113
114
115
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 108

def initialize(clock, logger)
  @mutex  = Mutex.new
  @clock  = clock
  @logger = logger
  @ticker = ::Proxy::Dynflow::Ticker.spawn('dispatcher-ticker', @clock, @logger, refresh_interval)
  @runner_actors = {}
  @runner_suspended_actions = {}
end

Instance Attribute Details

#ticker ⇒ Object (readonly)

Returns the value of attribute ticker.



107
108
109
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 107

def ticker
  @ticker
end

Class Method Details

.instance ⇒ Object 



6
7
8
9
10
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 6

def self.instance
  return @instance if @instance
  @instance = new(Proxy::Dynflow::Core.world.clock,
                  Proxy::Dynflow::Core.world.logger)
end

Instance Method Details

#external_event(runner_id, external_event) ⇒ Object 



153
154
155
156
157
158
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 153

def external_event(runner_id, external_event)
  synchronize do
    runner_actor = @runner_actors[runner_id]
    runner_actor&.tell([:external_event, external_event])
  end
end

#finish(runner_id) ⇒ Object 



145
146
147
148
149
150
151
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 145

def finish(runner_id)
  synchronize do
    _finish(runner_id)
  rescue => exception
    _handle_command_exception(runner_id, exception, false)
  end
end

#handle_command_exception(*args) ⇒ Object 



160
161
162
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 160

def handle_command_exception(*args)
  synchronize { _handle_command_exception(*args) }
end

#kill(runner_id) ⇒ Object 



136
137
138
139
140
141
142
143
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 136

def kill(runner_id)
  synchronize do
    runner_actor = @runner_actors[runner_id]
    runner_actor&.tell(:kill)
  rescue => exception
    _handle_command_exception(runner_id, exception, false)
  end
end

#refresh_interval ⇒ Object 



164
165
166
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 164

def refresh_interval
  1
end

#start(suspended_action, runner) ⇒ Object 



121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 121

def start(suspended_action, runner)
  synchronize do
    raise "Actor with runner id #{runner.id} already exists" if @runner_actors[runner.id]
    runner.logger = @logger
    runner_actor = RunnerActor.spawn("runner-actor-#{runner.id}", self, suspended_action, runner, @clock, @logger)
    @runner_actors[runner.id] = runner_actor
    @runner_suspended_actions[runner.id] = suspended_action
    runner_actor.tell(:start_runner)
    return runner.id
  rescue => exception
    _handle_command_exception(runner.id, exception)
    return nil
  end
end

#synchronize(&block) ⇒ Object 



117
118
119
 
# File 'lib/smart_proxy_dynflow/runner/dispatcher.rb', line 117

def synchronize(&block)
  @mutex.synchronize(&block)
end