Method dump_stacktrace has 35 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def dump_stacktrace(type, backtrace, level)
      return if @level > level

      time = Time.now

Method thread_create has 35 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def thread_create(title)
        raise ArgumentError, "BUG: title must be a symbol" unless title.is_a? Symbol
        raise ArgumentError, "BUG: callback not specified" unless block_given?
        m = Mutex.new
        m.lock

Method initialize has 11 arguments (exceeds 4 allowed). Consider refactoring.

Open

        def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_width, backoff_base, max_interval, secondary, secondary_threshold)

Method tls_verify_result_name has 34 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def self.tls_verify_result_name(code)
        case code
        when OpenSSL::X509::V_OK then 'V_OK'
        when OpenSSL::X509::V_ERR_AKID_SKID_MISMATCH then 'V_ERR_AKID_SKID_MISMATCH'
        when OpenSSL::X509::V_ERR_APPLICATION_VERIFICATION then 'V_ERR_APPLICATION_VERIFICATION'

Method read_ack_from_sock has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def read_ack_from_sock(sock)
        begin
          raw_data = sock.instance_of?(Fluent::PluginHelper::Socket::WrappedSocket::TLS) ? sock.readpartial(@read_length) : sock.recv(@read_length)
        rescue Errno::ECONNRESET, EOFError # ECONNRESET for #recv, #EOFError for #readpartial
          raw_data = ''

Method start_tcp_server has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def start_tcp_server(tls: false)
      octet_count_frame = @frame_type == :octet_count

      delimiter = octet_count_frame ? " " : @delimiter
      delimiter_size = delimiter.size

Method parse has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def parse(text)
        m = REGEXP.match(text)
        unless m
          yield nil, nil
          return

Method prepare_option_parser has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

  def prepare_option_parser
    @parser = OptionParser.new
    @parser.version = Fluent::VERSION
    @parser.banner = <<BANNER
Usage: #{$0} [options] <type> <name>

Method parse_rfc5424_regex has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def parse_rfc5424_regex(text, &block)
        idx = 0
        record = {}

        if @with_priority

Method parse! has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def parse!(allow_include, elem_name = nil, attrs = {}, elems = [])
        while line = @iterator.next
          line.force_encoding('UTF-8')
          @i += 1
          line.lstrip!

Method compat_parameters_parser has 33 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def compat_parameters_parser(conf)
        return unless conf.elements('parse').empty?
        return if PARSER_PARAMS.keys.all?{|k| !conf.has_key?(k) }

        # TODO: warn obsolete parameters if these are deprecated

Method establish_connection has 32 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def establish_connection(sock, ri)
        while ri.state != :established
          begin
            # TODO: On Ruby 2.2 or earlier, read_nonblock doesn't work expectedly.
            # We need rewrite around here using new socket/server plugin helper.

Method install_main_process_signal_handlers has 32 lines of code (exceeds 25 allowed). Consider refactoring.

Open

    def install_main_process_signal_handlers
      # Fluentd worker process (worker of ServerEngine) don't use code in serverengine to set signal handlers,
      # because it does almost nothing.
      # This method is the only method to set signal handlers in Fluentd worker process.

Method server_create_connection has 32 lines of code (exceeds 25 allowed). Consider refactoring.

Open

      def server_create_connection(title, port, proto: nil, bind: '0.0.0.0', shared: true, backlog: nil, tls_options: nil, **socket_options, &block)
        proto ||= (@transport_config && @transport_config.protocol == :tls) ? :tls : :tcp

        raise ArgumentError, "BUG: title must be a symbol" unless title && title.is_a?(Symbol)
        raise ArgumentError, "BUG: port must be an integer" unless port && port.is_a?(Integer)

Method parse has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

      def parse(text)
        parsed_json = @load_proc.call(text)

        if parsed_json.is_a?(Hash)
          time, record = parse_one_record(parsed_json)

• Read up
  Read up

Method setup_watcher has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def setup_watcher(target_info, pe)
      line_buffer_timer_flusher = @multiline_mode ? TailWatcher::LineBufferTimerFlusher.new(log, @multiline_flush_interval, &method(:flush_buffer)) : nil
      read_from_head = !@startup || @read_from_head
      tw = TailWatcher.new(target_info, pe, log, read_from_head, @follow_inodes, method(:update_watcher), line_buffer_timer_flusher, method(:io_handler), @metrics)

• Read up
  Read up

Method connect has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

      def connect(host:, port:, hostname:, ack: nil, &block)
        if @socket_cache
          return connect_keepalive(host: host, port: port, hostname: hostname, ack: ack, &block)
        end

• Read up
  Read up

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

Open

    def run
      batch_num    = (@rate / BIN_NUM).to_i
      residual_num = (@rate % BIN_NUM)
      while thread_current_running?
        current_time = Time.now.to_i

• Read up
  Read up

Method parse_plain has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

      def parse_plain(re, time, text, idx, record, capture_list, &block)
        m = re.match(text, idx)
        if m.nil?
          yield nil, nil
          return

• Read up
  Read up

Method try_rollback_write has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

      def try_rollback_write
        @dequeued_chunks_mutex.synchronize do
          while @dequeued_chunks.first && @dequeued_chunks.first.expired?
            info = @dequeued_chunks.shift
            if @buffer.takeback_chunk(info.chunk_id)

• Read up
  Read up