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

Open

    def receive_lines(lines, tail_watcher)
      es = @receive_handler.call(lines, tail_watcher)
      unless es.empty?
        tag = if @tag_prefix || @tag_suffix
                @tag_prefix + tail_watcher.tag + @tag_suffix

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

              pos = @read_from_head ? 0 : fsize

Avoid deeply nested control flow statements.

Open

                  if adding_bytes > @chunk_limit_size
                    errors << "concatenated/appended a #{adding_bytes} bytes record (nth: #{writing_splits_index}) is larger than buffer chunk limit size (#{@chunk_limit_size})"
                    writing_splits_index += 1
                    next
                  else

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

Open

    def detach_watcher_after_rotate_wait(tw, ino)
      # Call event_loop_attach/event_loop_detach is high-cost for short-live object.
      # If this has a problem with large number of files, use @_event_loop directly instead of timer_execute.
      if @open_on_every_update
        # Detach now because it's already closed, waiting it doesn't make sense.

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                break if merged_col.end_with?(@quote_char)

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

Open

      def check_pong(ri, message)
        @log.debug('checking pong')
        # ['PONG', bool(authentication result), 'reason if authentication failed',
        #  self_hostname, sha512\_hex(salt + self_hostname + nonce + sharedkey)]
        unless message.size == 5 && message[0] == 'PONG'

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    @ignore_list << p if @ignore_repeated_permission_error

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

Open

    def configure(conf)
      super

      unless @as_secondary
        raise Fluent::ConfigError, "This plugin can only be used in the <secondary> section"

• Read up
  Read up

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

Open

    def start
      super

      if @pos_file
        pos_file_dir = File.dirname(@pos_file)

• Read up
  Read up

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

Open

    def handle_parsed(tag, record, t, values)
      if values && @inject_key_prefix
        values = Hash[values.map { |k, v| [@inject_key_prefix + k, v] }]
      end
      r = @hash_value_field ? {@hash_value_field => values} : values

• Read up
  Read up

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

Open

      def tick
        now = Time.now.to_f
        unless available?
          if @failure.hard_timeout?(now)
            @failure.clear

• Read up
  Read up

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

Open

        def enqueued!
          return unless self.staged?

          new_chunk_path = self.class.generate_queued_chunk_path(@path, @unique_id)

• Read up
  Read up

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

Open

    def ack_check(select_interval)
      @ack_handler.collect_response(select_interval) do |chunk_id, node, sock, result|
        @connection_manager.close(sock)

        case result

• Read up
  Read up

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

Open

      def rollback_write(chunk_id, update_retry: true)
        # This API is to rollback chunks explicitly from plugins.
        # 3rd party plugins can depend it on automatic rollback of #try_rollback_write
        @dequeued_chunks_mutex.synchronize do
          @dequeued_chunks.delete_if{ |info| info.chunk_id == chunk_id }

• Read up
  Read up

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

Open

    def self.build(config_path:, encoding: 'utf-8', additional_config: nil, use_v1_config: true, type: nil)
      if type == :guess
        config_file_ext = File.extname(config_path)
        if config_file_ext == '.yaml' || config_file_ext == '.yml'
          type = :yaml

• Read up
  Read up

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

Open

      def handle_stream_simple(tag, es, enqueue: false)
        format_proc = nil
        meta = metadata((@chunk_key_tag ? tag : nil), nil, nil)
        records = es.size
        if @custom_format

• Read up
  Read up

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

Open

      def load(existing_targets = nil)
        compact(existing_targets)

        map = {}
        @file_mutex.synchronize do

• Read up
  Read up

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

Open

    def notify_new_supervisor_that_old_one_has_stopped
      if config[:pid_path]
        new_pid = File.read(config[:pid_path]).to_i
      else
        raise "[BUG] new_supervisor_pid is not saved" unless @new_supervisor_pid

• Read up
  Read up

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

Open

    def lifecycle(desc: false)
      kind_list = if desc
                    [:output, :filter, :output_with_router]
                  else
                    [:output_with_router, :filter, :output]

• Read up
  Read up

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

Open

      def synchronize_keys(*keys)
        return if keys.empty?
        keys = keys.dup

        while key = keys.shift

• Read up
  Read up