class Concurrent::Agent

def execute_next_job
  job       = synchronize { @queue.first }
  old_value = @current.value
  @caller.value = job.caller # for nested actions
  new_value     = job.action.call(old_value, *job.args)
  @caller.value = nil
  return if new_value == AWAIT_FLAG
  if ns_validate(new_value)
    @current.value = new_value
    observers.notify_observers(Time.now, old_value, new_value)
  else
    handle_error(ValidationError.new)
  end
rescue => error
  handle_error(error)
ensure
  synchronize do
    @queue.shift
    unless failed? || @queue.empty?
      ns_post_next_job
    end
  end
end