begin
  require 'kafka'
rescue LoadError
  raise 'Gem ruby-kafka is required for logging to Elasticsearch. Please add the gem "ruby-kafka" to your Gemfile.'
end

require 'date'

# Forward all log messages to Apache Kafka.
#
# Example:
#
#   SemanticLogger.add_appender(
#     appender: :kafka,
#
#     # At least one of these nodes must be available:
#     seed_brokers: ["kafka1:9092", "kafka2:9092"],
#
#     # Set an optional client id in order to identify the client to Kafka:
#     client_id: "my-application",
#   )
class SemanticLogger::Appender::Kafka < SemanticLogger::Subscriber
  attr_accessor :seed_brokers, :client_id, :connect_timeout, :socket_timeout,
                :ssl_ca_cert, :ssl_client_cert, :ssl_client_cert_key,
                :delivery_threshold, :delivery_interval,
                :topic, :partition, :partition_key, :key

  # Send log messages to Kafka in JSON format.
  #
  # Kafka Parameters:
  #
  #   seed_brokers: [Array<String>, String]
  #     The list of brokers used to initialize the client. Either an Array of connections,
  #     or a comma separated string of connections.
  #     Connections can either be a string of "port:protocol" or a full URI with a scheme.
  #     If there's a scheme it's ignored and only host/port are used.
  #
  #   client_id: [String]
  #     The identifier for this application.
  #     Default: semantic-logger
  #
  #   topic: [String]
  #     Topic to publish log messages to.
  #     Default: 'log_messages'
  #
  #   partition: [Integer]
  #     The partition that the message should be written to.
  #     Default: nil
  #
  #   partition_key: [String]
  #     The key that should be used to assign a partition.
  #     Default: nil
  #
  #   key: [String]
  #     The message key.
  #     Default: nil
  #
  #   connect_timeout: [Integer]
  #     The timeout setting for connecting to brokers.
  #     Default: nil
  #
  #   socket_timeout: [Integer]
  #     The timeout setting for socket connections.
  #     Default: nil
  #
  #   ssl_ca_cert: [String, Array<String>]
  #     A PEM encoded CA cert, or an Array of PEM encoded CA certs, to use with a SSL connection.
  #     Default: nil
  #
  #   ssl_client_cert: [String]
  #     A PEM encoded client cert to use with a SSL connection.
  #     Must be used in combination with ssl_client_cert_key.
  #     Default: nil
  #
  #   ssl_client_cert_key [String]
  #     A PEM encoded client cert key to use with a SSL connection.
  #     Must be used in combination with ssl_client_cert.
  #     Default: nil
  #
  #    delivery_threshold: [Integer]
  #      Number of messages between triggering a delivery of messages to Apache Kafka.
  #      Default: 100
  #
  #    delivery_interval: [Integer]
  #      Number of seconds between triggering a delivery of messages to Apache Kafka.
  #      Default: 5
  #
  # Semantic Logger Parameters:
  #
  #   level: [:trace | :debug | :info | :warn | :error | :fatal]
  #     Override the log level for this appender.
  #     Default: SemanticLogger.default_level
  #
  #   formatter: [Object|Proc|Symbol|Hash]
  #     An instance of a class that implements #call, or a Proc to be used to format
  #     the output from this appender
  #     Default: :raw_json (See: #call)
  #
  #   filter: [Regexp|Proc]
  #     RegExp: Only include log messages where the class name matches the supplied.
  #     regular expression. All other messages will be ignored.
  #     Proc: Only include log messages where the supplied Proc returns true
  #           The Proc must return true or false.
  #
  #   host: [String]
  #     Name of this host to appear in log messages.
  #     Default: SemanticLogger.host
  #
  #   application: [String]
  #     Name of this application to appear in log messages.
  #     Default: SemanticLogger.application
  def initialize(seed_brokers:, client_id: 'semantic-logger', connect_timeout: nil, socket_timeout: nil,
                 ssl_ca_cert: nil, ssl_client_cert: nil, ssl_client_cert_key: nil,
                 topic: 'log_messages', partition: nil, partition_key: nil, key: nil,
                 delivery_threshold: 100, delivery_interval: 10,
                 level: nil, formatter: nil, filter: nil, application: nil, host: nil, &block)

    @seed_brokers        = seed_brokers
    @client_id           = @client_id
    @connect_timeout     = connect_timeout
    @socket_timeout      = socket_timeout
    @ssl_ca_cert         = ssl_ca_cert
    @ssl_client_cert     = ssl_client_cert
    @ssl_client_cert_key = ssl_client_cert_key
    @topic               = topic
    @partition           = partition
    @partition_key       = partition_key
    @key                 = key
    @delivery_threshold  = delivery_threshold
    @delivery_interval   = delivery_interval

    super(level: level, formatter: formatter, filter: filter, application: application, host: host, &block)
    reopen
  end

  def reopen
    @kafka = ::Kafka.new(
      seed_brokers:        seed_brokers,
      client_id:           client_id,
      connect_timeout:     connect_timeout,
      socket_timeout:      socket_timeout,
      ssl_ca_cert:         ssl_ca_cert,
      ssl_client_cert:     ssl_client_cert,
      ssl_client_cert_key: ssl_client_cert_key,
      logger:              logger
    )

    @producer = @kafka.async_producer(
      delivery_threshold: delivery_threshold,
      delivery_interval:  delivery_interval
    )
  end

  def close
    @producer.shutdown if @producer
    @producer = nil
    @kafka.close if @kafka
    @kafka = nil
  end

  # Forward log messages to Kafka producer thread.
  def log(log)
    json = formatter.call(log, self)
    @producer.produce(json, topic: topic, partition: partition, partition_key: partition_key, key: key)
  end

  # Use JSON Formatter by default.
  def default_formatter
    SemanticLogger::Formatters::Json.new
  end

  # Restart producer thread since there is no other way to flush.
  def flush
    @producer.shutdown
    @producer = @kafka.async_producer(
      delivery_threshold: delivery_threshold,
      delivery_interval:  delivery_interval
    )
  end

end
