<commit_before>require 'mqtt'
require 'msgpack'
require 'fluent/plugin/output'

module Fluent::Plugin
  class OutMqtt < Output
    Fluent::Plugin.register_output('mqtt', self)

    helpers :compat_parameters, :inject, :formatter

    DEFAULT_BUFFER_TYPE = "memory"

    config_set_default :include_tag_key, false
    config_set_default :include_time_key, true

    config_param :port, :integer, :default => 1883
    config_param :bind, :string, :default => '127.0.0.1'
    config_param :topic, :string, :default => 'td-agent'
    config_param :format, :string, :default => 'none'
    config_param :client_id, :string, :default => nil
    config_param :username, :string, :default => nil
    config_param :password, :string, :default => nil
    config_param :ssl, :bool, :default => nil
    config_param :ca, :string, :default => nil
    config_param :key, :string, :default => nil
    config_param :cert, :string, :default => nil
    config_param :retain, :string, :default => true

    config_section :buffer do
      config_set_default :@type, DEFAULT_BUFFER_TYPE
      config_set_default :chunk_keys, ['tag']
    end

    config_section :inject do
      config_set_default :time_key, "time"
      config_set_default :time_type, "string"
      config_set_default :time_format, "%Y-%m-%dT%H:%M:%S%z"
    end

    unless method_defined?(:log)
      define_method(:log) { $log }
    end

    def initialize
      super

      @clients = {}
      @connection_options = {}
      @collection_options = {:capped => false}
    end

    def configure(conf)
      compat_parameters_convert(conf, :buffer, :inject, :formatter)
      super
      @bind ||= conf['bind']
      @topic ||= conf['topic']
      @port ||= conf['port']
      @formatter = formatter_create
      if conf.has_key?('buffer_chunk_limit')
        #check buffer_size
        conf['buffer_chunk_limit'] = available_buffer_chunk_limit(conf)
      end
    end

    def start

      $log.debug "start mqtt #{@bind}"
      opts = {host: @bind,
              port: @port}
      opts[:client_id] = @client_id if @client_id<commit_msg>Use log method instead of $log global variable to support plugin log level<commit_after>require 'mqtt'
require 'msgpack'
require 'fluent/plugin/output'

module Fluent::Plugin
  class OutMqtt < Output
    Fluent::Plugin.register_output('mqtt', self)

    helpers :compat_parameters, :inject, :formatter

    DEFAULT_BUFFER_TYPE = "memory"

    config_set_default :include_tag_key, false
    config_set_default :include_time_key, true

    config_param :port, :integer, :default => 1883
    config_param :bind, :string, :default => '127.0.0.1'
    config_param :topic, :string, :default => 'td-agent'
    config_param :format, :string, :default => 'none'
    config_param :client_id, :string, :default => nil
    config_param :username, :string, :default => nil
    config_param :password, :string, :default => nil
    config_param :ssl, :bool, :default => nil
    config_param :ca, :string, :default => nil
    config_param :key, :string, :default => nil
    config_param :cert, :string, :default => nil
    config_param :retain, :string, :default => true

    config_section :buffer do
      config_set_default :@type, DEFAULT_BUFFER_TYPE
      config_set_default :chunk_keys, ['tag']
    end

    config_section :inject do
      config_set_default :time_key, "time"
      config_set_default :time_type, "string"
      config_set_default :time_format, "%Y-%m-%dT%H:%M:%S%z"
    end

    unless method_defined?(:log)
      define_method(:log) { $log }
    end

    def initialize
      super

      @clients = {}
      @connection_options = {}
      @collection_options = {:capped => false}
    end

    def configure(conf)
      compat_parameters_convert(conf, :buffer, :inject, :formatter)
      super
      @bind ||= conf['bind']
      @topic ||= conf['topic']
      @port ||= conf['port']
      @formatter = formatter_create
      if conf.has_key?('buffer_chunk_limit')
        #check buffer_size
        conf['buffer_chunk_limit'] = available_buffer_chunk_limit(conf)
      end
    end

    def start

      log.debug "start mqtt #{@bind}"
      opts = {host: @bind,
              port: @port}
      opts[:client_id] = @client_id if @client_id
{'fraction_non_alphanumeric': 0.1223429381199811, 'fraction_numerical': 0.004723665564478035, 'mean_word_length': 3.3346980552712386, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2987184', 'n_tokens_mistral': 1375, 'n_tokens_neox': 1332, 'n_words': 331}