require 'march_hare'

require_relative 'base'


module Anschel
  class Input
    class RabbitMQ < Base
      def initialize output, config, stats, log
        connection_defaults = {
          heartbeat_interval: 30,
          connection_timeout: 10,
          automatically_recover: false
        }

        exchange_defaults = {
          type: 'x-consistent-hash',
          durable: true
        }

        queue_defaults = {
          exclusive: false,
          auto_delete: false,
          durable: true
        }

        binding_defaults = {
          routing_key: '100'
        }

        subscription_defaults = {
          block: true,
          ack: true,
          manual_ack: true
        }

        connection = ::MarchHare.connect \
          connection_defaults.merge(config[:connection] || {})

        handle_errors connection, log

        exchange_name = config[:exchange].delete(:name)

        @threads = config[:queues].map do |queue_name, queue_config|
          Thread.new do
            channel = connection.create_channel
            channel.prefetch = config[:prefetch] || 256

            exchange = channel.exchange exchange_name, \
              exchange_defaults.merge(config[:exchange])

            subscription = subscription_defaults.merge \
              (config[:subscription] || {})

            queue = channel.queue queue_name.to_s, \
              queue_defaults.merge(queue_config)

            queue.bind exchange, \
              binding_defaults.merge(config[:binding] || {})

            log.debug \
              event: 'input-rabbitmq-connecting-queue',
              queue: queue_name

            queue.subscribe(subscription) do |meta, message|
              output << message
              stats.inc 'input'
              channel.ack meta.delivery_tag if subscription[:manual_ack]
            end
          end
        end
      end

      def stop
        return if @stopped
        @threads.map(&:kill)
        @stopped = true
      end

    private

      # Ensure broker errors results in logs and exits appropriately. Seems like
      # the March Hare is catching exceptions under the hood, so we can't just
      # raise within these callbacks, and we're stuck with this jankiness
      def handle_errors connection, log
        shutdown, blocked = false, false

        connection.on_blocked { |reason| blocked = reason }

        connection.on_unblocked { blocked = nil }

        connection.on_shutdown { |_, reason| shutdown = reason }

        Thread.new do
          loop do
            sleep 1

            if shutdown
              log.fatal \
                event: 'input-rabbitmq-shutdown',
                reason: shutdown
              raise RuntimeError
            end

            if blocked
              log.warn \
                event: 'input-rabbitmq-blocked',
                reason: blocked

            elsif blocked.nil?
              log.warn \
                event: 'input-rabbitmq-unblocked'
            end

            blocked = false
          end
        end
      end
    end
  end
end