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

Open

    def read_messages(conn, &block)
      feeder = nil
      serializer = nil
      bytes = 0
      conn.data do |data|

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

Open

    def supervise
      Process.setproctitle("supervisor:#{@system_config.process_name}") if @system_config.process_name
      $log.info "starting fluentd-#{Fluent::VERSION}", pid: Process.pid, ruby: RUBY_VERSION

      fluentd_spawn_cmd = build_spawn_command

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

Open

        def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threshold)
          @title = title

          @start = current_time
          @steps = 0

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

Open

    def self.default_options
      {
        config_path: Fluent::DEFAULT_CONFIG_PATH,
        plugin_dirs: [Fluent::DEFAULT_PLUGIN_DIR],
        log_level: Fluent::Log::LEVEL_INFO,

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

Open

      def mount_proc(path, &block)
        @server.mount_proc(path) { |req, res|
          begin
            code, header, response = block.call(req, res)
          rescue => e

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

Open

      def configure(conf)
        bufconf = CompatOutputUtils.buffer_section(conf)
        config_style = (bufconf ? :v1 : :v0)
        if config_style == :v0
          buf_params = {

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

Open

        def optimized_filter_stream(tag, es)
          new_es = MultiEventStream.new
          es.each(unpacker: Fluent::MessagePackFactory.thread_local_msgpack_unpacker) do |time, record|
            filtered_record = record
            filtered_time = time

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

Open

        def call(request)
          method = request.method
          resp =
            case method
            when HttpServer::Methods::GET

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

Open

      def compat_parameters_inject(conf)
        return unless conf.elements('inject').empty?
        return if INJECT_PARAMS.keys.all?{|k| !conf.has_key?(k) }

        # TODO: warn obsolete parameters if these are deprecated

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

Open

      def purge_chunk(chunk_id)
        metadata = nil
        synchronize do
          chunk = @dequeued.delete(chunk_id)
          return nil unless chunk # purged by other threads

• Read up
  Read up

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

Open

  def try_connect
    begin
      now = Time.now.to_i

      unless @error_history.empty?

• Read up
  Read up

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

Open

    def update_watcher(tail_watcher, pe, new_inode)
      # TODO we should use another callback for this.
      # To supress impact to existing logics, limit the case to `@follow_inodes`.
      # We may not need `@follow_inodes` condition.
      if @follow_inodes && new_inode.nil?

• Read up
  Read up

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

Open

      def collect_response(select_interval)
        now = Fluent::Clock.now
        sockets = []
        results = []
        begin

• Read up
  Read up

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

Open

      def parse_time(record)
        if @time_key && record.respond_to?(:has_key?) && record.has_key?(@time_key)
          src = if @keep_time_key
                  record[@time_key]
                else

• Read up
  Read up

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

Open

    def on_message(msg)
      unless msg.is_a?(Array)
        log.warn "incoming data is broken:", msg: msg
        return
      end

• Read up
  Read up

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

Open

    def configure(conf)
      compat_parameters_convert(conf, :extract, :parser)
      ['parse', 'extract'].each do |subsection_name|
        if subsection = conf.elements(subsection_name).first
          if subsection.has_key?('time_format')

• Read up
  Read up

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

Open

    def write(chunk)
      path = extract_placeholders(@path_template, chunk)
      FileUtils.mkdir_p File.dirname(path), mode: @dir_perm

      writer = case

• Read up
  Read up

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

Open

    def self.parse(str, fname, basepath = Dir.pwd, v1_config = nil, syntax: :v1)
      parser = if fname =~ /\.rb$/ || syntax == :ruby
                 :ruby
               elsif v1_config.nil?
                 case syntax

• Read up
  Read up

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

Open

    def start
      super

      unless @heartbeat_type == :none
        if @heartbeat_type == :udp

• Read up
  Read up

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

Open

      def configure(conf)
        @include_time_key = false
        @localtime = false
        @timezone = nil

• Read up
  Read up