# encoding: utf-8
require "logstash/outputs/base"
require "logstash/namespace"
require "java"

# 自己写的插件类需要继承对应的 Base 类
class LogStash::Outputs::Rocketmq < LogStash::Outputs::Base

  # 设置插件可多线程并发执行
  # 默认为 single，即使配置多个 workers，也只能依次运行 multi_receive 方法（性能上来说变成单线程）
  concurrency :shared

  config_name "rocketmq"

  # 使用 config 定义插件运行所需要的基本参数
  # 本地 Logstash 的路径，必需，如 C:/ELK/logstash、/usr/local/logstash
  config :logstash_path, :validate => :string, :required => true

  # Rocketmq 的 NameServer 地址，必需，如 192.168.10.10:5678
  config :name_server_addr, :validate => :string, :required => true

  # Rocketmq 的 producer group
  config :producer_group, :validate => :string, :default => "defaultProducerGroup"

  # Message 的 topic，必需
  config :topic, :validate => :string, :required => true

  # Message 的 tag
  config :tag, :validate => :string, :default => "defaultTag"

  # Message 的 key
  config :key, :validate => :string, :default => "defaultKey"

  # 发送异常后的重试次数，默认 2 次
  config :retry_times, :validate => :number, :default => 2

  # 注册方法，在 LogStash 启动时会调用，相当于对插件进行初始化
  def register
    load_jar_files

    @stopping = Concurrent::AtomicBoolean.new(false)

    # 创建生产者对象
    @producer = org.apache.rocketmq.client.producer.DefaultMQProducer.new(producer_group)
    @producer.setNamesrvAddr(name_server_addr)
    @producer.start
  end

  # 接收事件方法，接收 events 对象，在其他插件里还有看到 multi_receive_encoded 方法
  # 入参是 events_and_data 对象，这两个方法有什么不同我反正是不懂
  def multi_receive(events)
    return if events.empty?
    events.each do |event|
      retrying_send(event)
    end
  end

  # 到我们指定的目录下加载依赖的 jar 文件
  def load_jar_files
    jarpath = logstash_path + "/vendor/jar/rocketmq/*.jar"
    @logger.info("RocketMq plugin required jar files are loadding... Jar files path: ", path: jarpath)

    jars = Dir[jarpath]
    raise LogStash::ConfigurationError, 'RocketMq plugin init error, no jars found! Please check the jar files path!' if jars.empty?

    jars.each do |jar|
      @logger.trace('RocketMq plugin loaded a jar: ', jar: jar)
      require jar
    end
  end

  # 生产者客户端核心功能方法
  def retrying_send(event)
    sent_times = 0

    begin
      # 配置 message 对象
      mq_message = org.apache.rocketmq.common.message.Message.new
      mq_message.setTopic(topic)
      mq_message.setTags(tag)
      mq_message.setKeys(key)
      mq_message.setBody("#{event}".bytes)
      result = @producer.send(mq_message)

      if result.nil?
        raise "Send message error! Result is null."
      end

      if org.apache.rocketmq.client.producer.SendStatus::SEND_OK != result.getSendStatus
        status_name = result.getSendStatus.name
        raise "Send message error! Result code is #{status_name}"
      end
    rescue => e
      @logger.error('An Exception Occured!!',
                    :message => e.message,
                    :exception => e.class)
      if @stopping.false? and (sent_times < retry_times)
        # 重试
        sent_times += 1
        retry
      else
        # 根据实际需求处理没发送成功的消息
        puts "Message send failed: #{event}"
      end
    end
  end

  # LogStash 停止时会调用该方法
  def close
    @stopping.make_true
    @producer.shutdown
  end
end