require 'thread'

require 'typhoeus'
require 'typhoeus/adapters/faraday'
require 'elasticsearch'

require_relative 'base'


module Anschel
  class Output
    class Elasticsearch < Base
      def initialize config, stats, log
        default_index = config.delete(:default_index) || '.anschel'
        qsize   = config.delete(:queue_size) || 2000
        bsize   = config.delete(:bulk_size) || 500
        timeout = config.delete(:bulk_timeout) || 0.5
        slice   = timeout / bsize
        client  = ::Elasticsearch::Client.new config
        client.transport.reload_connections!

        @queue = SizedQueue.new qsize

        @thread = Thread.new do
          loop do
            events = []
            count  = 0
            start  = Time.now.to_f
            until (Time.now.to_f - start > timeout) || ((count += 1) > bsize)
              begin
                events.push @queue.shift(true)
                count += 1
              rescue # shift returned immediately
                sleep slice
              end
            end

            next if events.empty?

            body = events.map do |e|
              index = e.delete(:_index)
              routing = e.delete(:_routing)

              if index.nil?
                log.error \
                  event: 'elasticsearch-output-error',
                  reason: 'event was not indexed',
                  remediation: "sending to default index '#{default_index}'",
                  raw_event: e
                index = default_index
              end

              item = { _index: index, _type: e[:type], data: e }
              item[:_routing] = routing if routing
              { index: item }
            end

            response = client.bulk body: body

            if response['errors']
              log.error \
                event: 'elasticsearch-output-error',
                reason: 'response contained errors',
                body: body,
                response: response
            end

            stats.inc 'output', body.size
          end
        end
      end
    end
  end
end