# encoding: utf-8

require 'spec_helper'


module Cql
  module Protocol
    describe CqlProtocolHandler do
      let :protocol_handler do
        described_class.new(connection, scheduler, 1)
      end

      let :connection do
        double(:connection)
      end

      let :scheduler do
        double(:scheduler)
      end

      let :request do
        Protocol::OptionsRequest.new
      end

      let :buffer do
        CqlByteBuffer.new
      end

      before do
        connection.stub(:on_data) do |&h|
          connection.stub(:data_listener).and_return(h)
        end
        connection.stub(:on_closed) do |&h|
          connection.stub(:closed_listener).and_return(h)
        end
        connection.stub(:on_connected) do |&h|
          connection.stub(:connected_listener).and_return(h)
        end
        scheduler.stub(:schedule_timer).and_return(Promise.new.future)
        protocol_handler
      end

      describe '#initialize' do
        it 'registers as a data listener to the socket handler' do
          connection.data_listener.should_not be_nil
        end
      end

      describe '#[]/#[]=' do
        it 'is thread safe storage for arbitrary data' do
          protocol_handler[:host_id] = 42
          protocol_handler[:host_id].should == 42
        end
      end

      describe '#host' do
        it 'delegates to the connection' do
          connection.stub(:host).and_return('example.com')
          protocol_handler.host.should == 'example.com'
        end
      end

      describe '#port' do
        it 'delegates to the connection' do
          connection.stub(:port).and_return(9042)
          protocol_handler.port.should == 9042
        end
      end

      describe '#send_request' do
        before do
          connection.stub(:write) do |*args, &block|
            if block
              block.call(buffer)
            else
              buffer.append(*args)
            end
          end
          connection.stub(:closed?).and_return(false)
          connection.stub(:connected?).and_return(true)
        end

        it 'encodes a request frame and writes to the socket handler' do
          protocol_handler.send_request(request)
          version, flags, stream_id, opcode, size, body = *buffer.to_s.unpack('C4Na*')
          version.should eq(1)
          flags.should eq(0)
          stream_id.should be_between(0, 127)
          opcode.should eq(5)
          size.should eq(0)
          body.should be_empty
        end

        it 'encodes a frame with the next available stream ID' do
          protocol_handler.send_request(request)
          protocol_handler.send_request(request)
          version1, flags1, stream_id1, opcode1, size1,
          version2, flags2, stream_id2, opcode2, size2, body2 = *buffer.to_s.unpack('C4NC4Na*')
          version2.should eq(version1)
          flags2.should eq(flags1)
          stream_id2.should be_between(0, 127)
          stream_id2.should_not eq(stream_id1)
          opcode2.should eq(opcode1)
          size2.should eq(size1)
          body2.should be_empty
        end

        it 'returns a future' do
          protocol_handler.send_request(request).should be_a(Future)
        end

        it 'succeeds the future when it receives a response frame with the corresponding stream ID' do
          3.times { protocol_handler.send_request(request) }
          buffer.discard(buffer.size)
          future = protocol_handler.send_request(request)
          stream_id = buffer.to_s.getbyte(2)
          connection.data_listener.call([0x81, 0, stream_id, 2, 0].pack('C4N'))
          await(0.1) { future.resolved? }
        end

        it 'handles multiple response frames in the same data packet' do
          futures = Array.new(4) { protocol_handler.send_request(request) }
          stream_ids = buffer.to_s.unpack('C4NC4NC4NC4N').values_at(2, 7, 12, 17)
          connection.data_listener.call([0x81, 0, stream_ids[2], 2, 0].pack('C4N') + [0x81, 0, stream_ids[3], 2, 0].pack('C4N'))
          await(0.1) { futures[2].resolved? && futures[3].resolved? }
        end

        it 'queues the request when there are too many in flight, sending it as soon as a stream is available' do
          futures = Array.new(128) { protocol_handler.send_request(request) }
          buffer.discard(buffer.size)
          delayed = Array.new(2) { protocol_handler.send_request(request) }
          futures.size.times { |i| connection.data_listener.call([0x81, 0, i, 2, 0].pack('C4N')) }
          futures.each { |future| future.should be_resolved }
          delayed.each { |future| future.should_not be_resolved }
          stream_ids = buffer.to_s.unpack('C4NC4N').values_at(2, 7)
          stream_ids.each { |id| connection.data_listener.call([0x81, 0, id, 2, 0].pack('C4N')) }
          delayed.each { |future| future.should be_resolved }
        end

        it 'flushes the request queue before it resolves the future of the just completed request' do
          f = protocol_handler.send_request(request)
          stream_id = buffer.to_s.getbyte(2)
          129.times { protocol_handler.send_request(request) }
          f = f.map do
            connection.should have_received(:write).exactly(129).times
          end
          connection.data_listener.call([0x81, 0, stream_id, 2, 0].pack('C4N'))
          f.value
        end

        context 'when a compressor is specified' do
          let :protocol_handler do
            described_class.new(connection, scheduler, 1, compressor)
          end

          let :compressor do
            double(:compressor)
          end

          let :request do
            Protocol::PrepareRequest.new('SELECT * FROM things')
          end

          before do
            compressor.stub(:compress?).and_return(true)
            compressor.stub(:compress).and_return('FAKECOMPRESSEDBODY')
          end

          it 'compresses request frames' do
            protocol_handler.send_request(request)
            version, flags, stream_id, opcode, size, body = *buffer.to_s.unpack('C4Na*')
            version.should eq(1)
            flags.should eq(1)
            stream_id.should be_between(0, 127)
            opcode.should eq(9)
            size.should eq(18)
            body.should eq('FAKECOMPRESSEDBODY')
          end

          it 'compresses queued request frames' do
            130.times { protocol_handler.send_request(request) }
            compressor.should have_received(:compress).exactly(130).times
          end

          it 'decompresses response frames' do
            id = "\xCAH\x7F\x1Ez\x82\xD2<N\x8A\xF35Qq\xA5/".force_encoding(::Encoding::BINARY)
            compressor.stub(:decompress).with('FAKECOMPRESSEDBODY').and_return("\x00\x00\x00\x04" + "\x00\x10" + id + "\x00\x00\x00\x01\x00\x00\x00\x01\x00\ncql_rb_911\x00\x05users\x00\tuser_name\x00\r")
            f1 = protocol_handler.send_request(request)
            f2 = protocol_handler.send_request(request)
            stream_ids = buffer.to_s.unpack('C4Na18C4N').values_at(2, 8)
            stream_ids.each do |id|
              connection.data_listener.call([0x81, 1, id, 8, 18, 'FAKECOMPRESSEDBODY'].pack('C4Na*'))
            end
            f1.value.should == Protocol::PreparedResultResponse.new(id, [["cql_rb_911", "users", "user_name", :varchar]], nil, nil)
            f2.value.should == Protocol::PreparedResultResponse.new(id, [["cql_rb_911", "users", "user_name", :varchar]], nil, nil)
          end
        end

        context 'when a protocol version is specified' do
          let :protocol_handler do
            described_class.new(connection, scheduler, 7)
          end

          it 'sets the protocol version in the header' do
            protocol_handler.send_request(request)
            buffer.to_s[0].should == "\x07"
          end
        end

        context 'when the protocol handler closes' do
          it 'fails all requests waiting for a reply' do
            futures = Array.new(5) { protocol_handler.send_request(request) }
            connection.closed_listener.call(StandardError.new('Blurgh'))
            futures.should be_all(&:failed?), 'expected all requests to have failed'
          end

          it 'fails all queued requests' do
            futures = Array.new(200) { protocol_handler.send_request(request) }
            connection.closed_listener.call(StandardError.new('Blurgh'))
            futures.should be_all(&:failed?), 'expected all requests to have failed'
          end

          it 'fails all requests with ConnectionClosedError if there is no specific error' do
            protocol_handler.send_request(request)
            future = protocol_handler.send_request(request)
            stream_id = buffer.to_s.getbyte(2)
            connection.data_listener.call([0x81, 0, stream_id, 2, 0].pack('C4N'))
            connection.closed_listener.call(nil)
            begin
              future.value
            rescue => e
              e.should be_a(Cql::Io::ConnectionClosedError)
            else
              fail('No error was raised!')
            end
          end

          it 'passes the error that caused the protocol handler to close to the failed requests' do
            error = nil
            future = protocol_handler.send_request(request)
            future.on_failure { |e| error = e }
            connection.closed_listener.call(StandardError.new('Blurgh'))
            error.should == StandardError.new('Blurgh')
          end
        end

        context 'when the protocol handler has closed' do
          it 'fails all requests with NotConnectedError' do
            connection.stub(:closed?).and_return(true)
            f = protocol_handler.send_request(request)
            expect { f.value }.to raise_error(NotConnectedError)
          end
        end

        context 'when the request times out' do
          let :timer_promise do
            Promise.new
          end

          before do
            scheduler.stub(:schedule_timer).with(3).and_return(timer_promise.future)
          end

          it 'raises a TimeoutError' do
            f = protocol_handler.send_request(request, 3)
            timer_promise.fulfill
            expect { f.value }.to raise_error(TimeoutError)
          end

          it 'does not attempt to fulfill the promise when the request has already timed out' do
            f = protocol_handler.send_request(request, 3)
            stream_id = buffer.to_s.getbyte(2)
            timer_promise.fulfill
            expect { connection.data_listener.call([0x81, 0, stream_id, 2, 0].pack('C4N')) }.to_not raise_error
          end

          it 'never sends a request when it has already timed out' do
            write_count = 0
            connection.stub(:write) do |s, &h|
              write_count += 1
              if h
                h.call(buffer)
              else
                buffer << s
              end
            end
            128.times do
              scheduler.stub(:schedule_timer).with(5).and_return(Promise.new.future)
              protocol_handler.send_request(request)
            end
            scheduler.stub(:schedule_timer).with(5).and_return(timer_promise.future)
            f = protocol_handler.send_request(request, 5)
            timer_promise.fulfill
            128.times { |i| connection.data_listener.call([0x81, 0, i, 2, 0].pack('C4N')) }
            write_count.should == 128
          end

          it 'does not stop sending queued requests even when one has timed out' do
            write_count = 0
            connection.stub(:write) do |s, &h|
              write_count += 1
              if h
                h.call(buffer)
              else
                buffer << s
              end
            end
            128.times do
              protocol_handler.send_request(request)
            end
            scheduler.stub(:schedule_timer).with(5).and_return(timer_promise.future)
            f1 = protocol_handler.send_request(request, 5)
            f2 = protocol_handler.send_request(request)
            timer_promise.fulfill
            connection.data_listener.call([0x81, 0, 0, 2, 0].pack('C4N'))
            write_count.should == 129
          end
        end
      end

      describe '#close' do
        it 'closes the underlying protocol handler' do
          connection.should_receive(:close)
          protocol_handler.close
        end

        it 'returns a future which succeeds when the socket has closed' do
          connection.stub(:close) do
            connection.closed_listener.call(nil)
          end
          protocol_handler.close.value
        end
      end

      describe '#keyspace' do
        before do
          connection.stub(:closed?).and_return(false)
          connection.stub(:connected?).and_return(true)
          connection.stub(:write).and_yield(buffer)
        end

        it 'is not in a keyspace initially' do
          protocol_handler.keyspace.should be_nil
        end

        it 'registers the keyspace it has changed to' do
          f = protocol_handler.send_request(Protocol::QueryRequest.new('USE hello', nil, nil, :one))
          stream_id = buffer.to_s.getbyte(2)
          connection.data_listener.call([0x81, 0, stream_id, 8, 4 + 2 + 5, 3, 5].pack('C4N2n') + 'hello')
          f.value
          protocol_handler.keyspace.should == 'hello'
        end
      end

      [:connected?, :closed?].each do |message|
        describe "##{message}" do
          it 'reflects the underlying protocol handler\'s status' do
            connection.stub(message).and_return(true)
            protocol_handler.send(message).should be_true
            connection.stub(message).and_return(false)
            protocol_handler.send(message).should be_false
          end
        end
      end

      describe '#on_event' do
        it 'calls the callback on events' do
          event = nil
          protocol_handler.on_event do |e|
            event = e
          end
          connection.data_listener.call("\x81\x00\xFF\f\x00\x00\x00+\x00\rSCHEMA_CHANGE\x00\aDROPPED\x00\x0cthe_keyspace\x00\x09the_table")
          event.should == Protocol::SchemaChangeEventResponse.new('DROPPED', 'the_keyspace', 'the_table')
        end

        it 'ignores errors raised by the listener' do
          called = false
          protocol_handler.on_event { |e| raise 'Blurgh' }
          protocol_handler.on_event { |e| called = true }
          connection.data_listener.call("\x81\x00\xFF\f\x00\x00\x00+\x00\rSCHEMA_CHANGE\x00\aDROPPED\x00\x0cthe_keyspace\x00\x09the_table")
          called.should be_true, 'expected all event listeners to have been called'
        end
      end

      describe '#on_closed' do
        it 'calls the callback when the underlying protocol handler closes' do
          called = false
          protocol_handler.on_closed { called = true }
          connection.closed_listener.call(StandardError.new('Blurgh'))
          called.should be_true, 'expected the close listener to have been called'
        end

        it 'passes the error that made the connection close to the listener' do
          error = nil
          protocol_handler.on_closed { |e| error = e }
          connection.closed_listener.call(StandardError.new('Blurgh'))
          error.message.should == 'Blurgh'
        end

        it 'ignores errors raised by the listener' do
          called = false
          protocol_handler.on_closed { |e| raise 'Blurgh' }
          protocol_handler.on_closed { |e| called = true }
          connection.closed_listener.call(StandardError.new('Blurgh'))
          called.should be_true, 'expected all event listeners to have been called'
        end
      end
    end
  end
end
