Class Job has 28 methods (exceeds 20 allowed). Consider refactoring.

Open

  class Job < AbstractJob
    include ActiveSupport::Callbacks
    define_callbacks :run,
      terminator: ->(_,result) { result == false },
      skip_after_callbacks_if_terminated: true,

Method leveled_each has a Cognitive Complexity of 23 (exceeds 5 allowed). Consider refactoring.

Open

    def leveled_each
      visited = Set.new
      queue = []
      queue.push(*@jobs.select { |j| j.in_comings.empty? })

• Read up
  Read up

Method run has a Cognitive Complexity of 16 (exceeds 5 allowed). Consider refactoring.

Open

      def run
        process = -> do
          if preview
            stdout.puts "Config:\n#{File.read(config)}"
            cmds = [embulk_bin, "preview", *embulk_bundle, config].compact

• Read up
  Read up

Method execute_command has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

      def execute_command(command, env, opts)
        log = "".dup
        result = Open3.popen3(env, *command, opts) do |stdin, out, err, wait_th|
          stdin.close
          until out.eof? && err.eof?

• Read up
  Read up

Method run has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def run(wait_interval = 1)
      promise = Concurrent::Promise.new do
        futures = execute
        until futures.all?(&:complete?)
          yield self if block_given?

• Read up
  Read up

Method [] has 28 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def self.[](job_class)
        raise "Please set Rukawa.config.status_store subclass of ActiveSupport::Cache::Store" unless Rukawa.config.status_store
        @wrapper_classes ||= {}
        return @wrapper_classes[job_class] if @wrapper_classes[job_class]

Method run has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

          def run
            @job_class.include(Hooks) unless @job_class.include?(Hooks)
            @job_class.prepend(HooksForFailure) unless @job_class.include?(HooksForFailure)
            job = @job_class.perform_later

• Read up
  Read up

Method attrs has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def attrs
      if rankdir || size || rotate || ranksep || nodesep || concentrate
        values = to_h.map { |k, v| "#{k} = #{v}" if k != :node && v }.compact
        "graph [#{values.join(",")}];\n"
      else

• Read up
  Read up

Method handle_parameters has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

        def handle_parameters(config:, embulk_bin: nil, embulk_bundle: nil, embulk_vm_options: nil, jvm_options: nil, stdout: nil, stderr: nil, env: nil, chdir: nil, preview: false)
          self.config = config
          self.embulk_bin = embulk_bin if embulk_bin
          self.embulk_bundle = embulk_bundle if embulk_bundle
          self.embulk_vm_options = embulk_vm_options if embulk_vm_options

• Read up
  Read up

Method attrs has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def attrs
      if shape || style
        values = to_h.map { |k, v| "#{k} = #{v}" if v }.compact
        "node [#{values.join(",")}];\n"
      else

• Read up
  Read up

Method handle_error has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def handle_error(e)
      Rukawa.logger.error("Error #{self.class} by #{e}")
      if retry?(e)
        @retry_count += 1
        set_state(:waiting)

• Read up
  Read up

Consider simplifying this complex logical expression.

Open

      if rankdir || size || rotate || ranksep || nodesep || concentrate
        values = to_h.map { |k, v| "#{k} = #{v}" if k != :node && v }.compact
        "graph [#{values.join(",")}];\n"
      else
        ""

Method run has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def run(batch_mode = false, refresh_interval = DEFAULT_REFRESH_INTERVAL)
      displayed_at = Time.at(0)
      promise = @root_job_net.run do
        unless batch_mode
          if Time.now - displayed_at >= refresh_interval

• Read up
  Read up

Method run has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def run
        cmdenv = env || {}
        opts = chdir ? {chdir: chdir} : {}

        if defined?(Bundler)

• Read up
  Read up

Method fetch_condition has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def fetch_condition
        opts = {if_modified_since: if_modified_since, if_unmodified_since: if_unmodified_since}.reject do |_, v|
          v.nil?
        end

• Read up
  Read up

Method formatted_elapsed_time_from has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def formatted_elapsed_time_from(time = Time.now)
      elapsed = elapsed_time_from(time)
      return "N/A" unless elapsed

      hour = elapsed.to_i / 3600

• Read up
  Read up

Method _run has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _run(job_net_name, *job_name)
      load_config
      set_logger
      set_concurrency
      load_job_definitions

• Read up
  Read up

Method skip? has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def skip?
      parent_skip = @parent_job_net ? @parent_job_net.skip? : false
      parent_skip || skip_rules.inject(false) do |cond, rule|
        cond || rule.is_a?(Symbol) ? method(rule).call : rule.call(self)
      end

• Read up
  Read up