# encoding: utf-8
require "logstash/devutils/rspec/spec_helper"
require "logstash/inputs/rabbitmq"
require "thread"
require 'logstash/event'

Thread.abort_on_exception = true

describe LogStash::Inputs::RabbitMQ do
  let(:klass) { LogStash::Inputs::RabbitMQ }
  let(:host) { "localhost" }
  let(:port) { 5672 }
  let(:exchange_type) { "topic" }
  let(:exchange) { "myexchange" }
  let(:queue) { "myqueue" }
  let(:rabbitmq_settings) {
    {
      "host" => host,
      "port" => port,
      "queue" => queue,
      "prefetch_count" => 123
    }
  }
  let(:instance) { klass.new(rabbitmq_settings) }
  let(:hare_info) { instance.instance_variable_get(:@hare_info) }

  context "when connected" do
    let(:connection) { double("MarchHare Connection") }
    let(:channel) { double("Channel") }
    let(:exchange) { double("Exchange") }
    let(:channel) { double("Channel") }
    let(:queue) { double("queue") }

    # Doing this in a before block doesn't give us enough control over scope
    before do
      allow(instance).to receive(:connect!).and_call_original
      allow(::MarchHare).to receive(:connect).and_return(connection)
      allow(connection).to receive(:create_channel).and_return(channel)
      allow(connection).to receive(:on_shutdown)
      allow(connection).to receive(:on_blocked)
      allow(connection).to receive(:on_unblocked)
      allow(channel).to receive(:exchange).and_return(exchange)
      allow(channel).to receive(:queue).and_return(queue)
      allow(channel).to receive(:prefetch=)

      allow(queue).to receive(:build_consumer).with(:on_cancellation => anything)
      allow(queue).to receive(:subscribe_with).with(any_args)
      allow(queue).to receive(:bind).with(any_args)
    end

    it "should default the codec to JSON" do
      expect(instance.codec).to be_a(LogStash::Codecs::JSON)
    end

    describe "#connect!" do
      subject { hare_info }

      context "without an exchange declared" do
        before do
          instance.register
          instance.setup!
        end

        it "should set the queue correctly" do
          expect(subject.queue).to eql(queue)
        end

        it "should set the prefetch value correctly" do
          expect(channel).to have_received(:prefetch=).with(123)
        end
      end

      context "with an exchange declared" do
        let(:exchange) { "exchange" }
        let(:key) { "routing key" }
        let(:rabbitmq_settings) { super.merge("exchange" => exchange, "key" => key, "exchange_type" => "fanout") }

        before do
          allow(instance).to receive(:declare_exchange!)
        end

        context "on run" do
          before do
            instance.register
            instance.setup!
          end

          it "should bind to the exchange" do
            expect(queue).to have_received(:bind).with(exchange, :routing_key => key)
          end

          it "should declare the exchange" do
            expect(instance).to have_received(:declare_exchange!)
          end
        end

        context "but not immediately available" do
          before do
            i = 0
            allow(queue).to receive(:bind).with(any_args) do
              i += 1
              raise "foo" if i == 1
            end
          end

          it "should reconnect" do
            instance.register
            instance.setup!
            expect(queue).to have_received(:bind).with(any_args).twice()
          end
        end

        context "initially unable to subscribe" do
          before do
            i = 0
            allow(queue).to receive(:subscribe_with).with(any_args) do
              i += 1
              raise "sub error" if i == 1
            end

            it "should retry the subscribe" do
              expect(queue).to have_receive(:subscribe_with).twice()
            end
          end
        end
      end
    end
  end
end

describe "with a live server", :integration => true do
  let(:klass) { LogStash::Inputs::RabbitMQ }
  let(:config) { {"host" => "127.0.0.1", "auto_delete" => true, "codec" => "plain", "add_field" => {"[@metadata][foo]" => "bar"} } }
  let(:instance) { klass.new(config) }
  let(:hare_info) { instance.instance_variable_get(:@hare_info) }
  let(:output_queue) { Queue.new }

  # Spawn a connection in the bg and wait up (n) seconds
  def spawn_and_wait(instance)
    instance.register

    output_queue # materialize in this thread

    Thread.new {
      instance.run(output_queue)
    }

    20.times do
      instance.connected? ? break : sleep(0.1)
    end

    # Extra time to make sure the consumer can attach
    # Without this there's a chance the shutdown code will execute
    # before consumption begins. This is tricky to do more elegantly
    sleep 4
  end

  let(:test_connection) { MarchHare.connect(instance.send(:rabbitmq_settings)) }
  let(:test_channel) { test_connection.create_channel }

  before do
    # Materialize the instance in the current thread to prevent dupes
    # If you use multiple threads with lazy evaluation weird stuff happens
    instance
    spawn_and_wait(instance)

    test_channel # Start up the test client as well
  end

  after do
    instance.stop()
    test_channel.close
    test_connection.close
  end

  context "using defaults" do
    it "should start, connect, and stop cleanly" do
      expect(instance.connected?).to be_truthy
    end
  end

  it "should have the correct prefetch value" do
    expect(instance.instance_variable_get(:@hare_info).channel.prefetch).to eql(256)
  end

  describe "receiving a message with a queue + exchange specified" do
    let(:config) { super.merge("queue" => queue_name, "exchange" => exchange_name, "exchange_type" => "fanout", "metadata_enabled" => true) }
    let(:event) { output_queue.pop }
    let(:exchange) { test_channel.exchange(exchange_name, :type => "fanout") }
    let(:exchange_name) { "logstash-input-rabbitmq-#{rand(0xFFFFFFFF)}" }
    #let(:queue) { test_channel.queue(queue_name, :auto_delete => true) }
    let(:queue_name) { "logstash-input-rabbitmq-#{rand(0xFFFFFFFF)}" }

    after do
      exchange.delete
    end

    context "when the message has a payload but no message headers" do
      before do
        exchange.publish(message)
      end

      let(:message) { "Foo Message" }

      it "should process the message and store the payload" do
        expect(event.get("message")).to eql(message)
      end

      it "should save an empty message header hash" do
        expect(event).to include("@metadata")
        expect(event.get("@metadata")).to include("rabbitmq_headers")
        expect(event.get("[@metadata][rabbitmq_headers]")).to eq({})
      end
    end

    context "when message properties are available" do
      before do
        # Don't test every single property but select a few with
        # different characteristics to get sufficient coverage.
        exchange.publish("",
                      :properties => {
                        :app_id    => app_id,
                        :timestamp => Java::JavaUtil::Date.new(epoch * 1000),
                        :priority  => priority,
                      })
      end

      let(:app_id) { "myapplication" }
      # Randomize the epoch we test with but limit its range to signed
      # ints to not assume all protocols and libraries involved use
      # unsigned ints for epoch values.
      let(:epoch) { rand(0x7FFFFFFF) }
      let(:priority) { 5 }

      it "should save message properties into a @metadata field" do
        expect(event).to include("@metadata")
        expect(event.get("@metadata")).to include("rabbitmq_properties")

        props = event.get("[@metadata][rabbitmq_properties")
        expect(props["app-id"]).to eq(app_id)
        expect(props["delivery-mode"]).to eq(1)
        expect(props["exchange"]).to eq(exchange_name)
        expect(props["priority"]).to eq(priority)
        expect(props["routing-key"]).to eq("")
        expect(props["timestamp"]).to eq(epoch)
      end
    end

    context "when message headers are available" do
      before do
        exchange.publish("", :properties => { :headers => headers })
      end

      let (:headers) {
        {
          "arrayvalue"  => [true, 123, "foo"],
          "boolvalue"   => true,
          "intvalue"    => 123,
          "stringvalue" => "foo",
        }
      }

      it "should save message headers into a @metadata field" do
        expect(event).to include("@metadata")
        expect(event.get("@metadata")).to include("rabbitmq_headers")
        expect(event.get("[@metadata][rabbitmq_headers]")).to include(headers)
      end

      it "should properly decorate the event" do
        expect(event.get("[@metadata][foo]")).to eq("bar")
      end
    end
  end

  describe LogStash::Inputs::RabbitMQ do
    it_behaves_like "an interruptible input plugin"
  end
end