/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.bookkeeper.proto;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.bookkeeper.proto.BookieProtocol.FLAG_NONE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import com.google.common.collect.Lists;
import com.google.protobuf.ByteString;
import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.apache.bookkeeper.proto.BookieProtoEncoding.RequestEnDeCoderPreV3;
import org.apache.bookkeeper.proto.BookieProtoEncoding.RequestEnDecoderV3;
import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseDecoder;
import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseEnDeCoderPreV3;
import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseEnDecoderV3;
import org.apache.bookkeeper.proto.BookieProtocol.AddResponse;
import org.apache.bookkeeper.proto.BookkeeperProtocol.AddRequest.Flag;
import org.apache.bookkeeper.proto.BookkeeperProtocol.BKPacketHeader;
import org.apache.bookkeeper.proto.BookkeeperProtocol.OperationType;
import org.apache.bookkeeper.proto.BookkeeperProtocol.ProtocolVersion;
import org.apache.bookkeeper.proto.BookkeeperProtocol.StatusCode;
import org.apache.bookkeeper.util.ByteBufList;
import org.junit.Before;
import org.junit.Test;

/**
 * Unit test {@link BookieProtoEncoding}.
 */
public class BookieProtoEncodingTest {

    private ExtensionRegistry registry;

    @Before
    public void setup() {
        this.registry = ExtensionRegistry.newInstance();
    }

    @Test
    public void testV3ResponseDecoderNoFallback() throws Exception {
        AddResponse v2Resp = AddResponse.create(
            BookieProtocol.CURRENT_PROTOCOL_VERSION,
            BookieProtocol.EOK,
            1L,
            2L);

        BookkeeperProtocol.Response v3Resp = BookkeeperProtocol.Response.newBuilder()
            .setHeader(BKPacketHeader.newBuilder()
                .setVersion(ProtocolVersion.VERSION_THREE)
                .setTxnId(1L)
                .setOperation(OperationType.ADD_ENTRY)
                .build())
            .setStatus(StatusCode.EOK)
            .setAddResponse(BookkeeperProtocol.AddResponse.newBuilder()
                .setStatus(StatusCode.EOK)
                .setLedgerId(1L)
                .setEntryId(2L)
                .build())
            .build();

        List<Object> outList = Lists.newArrayList();
        ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
        when(ctx.fireChannelRead(any())).thenAnswer((iom) -> {
                outList.add(iom.getArgument(0));
                return null;
        });

        ResponseEnDeCoderPreV3 v2Encoder = new ResponseEnDeCoderPreV3(registry);
        ResponseEnDecoderV3 v3Encoder = new ResponseEnDecoderV3(registry);

        ResponseDecoder v3Decoder = new ResponseDecoder(registry, false, false);
        try {
            v3Decoder.channelRead(ctx,
                v2Encoder.encode(v2Resp, UnpooledByteBufAllocator.DEFAULT)
            );
            fail("V3 response decoder should fail on decoding v2 response");
        } catch (InvalidProtocolBufferException e) {
            // expected
        }
        assertEquals(0, outList.size());

        ByteBuf serWithFrameSize = (ByteBuf) v3Encoder.encode(v3Resp, UnpooledByteBufAllocator.DEFAULT);
        ByteBuf ser = serWithFrameSize.slice(4, serWithFrameSize.readableBytes() - 4);
        v3Decoder.channelRead(ctx, ser);
        assertEquals(1, outList.size());
    }

    @Test(expected = IllegalStateException.class)
    public void testV2RequestDecoderThrowExceptionOnUnknownRequests() throws Exception {
        RequestEnDeCoderPreV3 v2ReqEncoder = new RequestEnDeCoderPreV3(registry);
        RequestEnDecoderV3 v3ReqEncoder = new RequestEnDecoderV3(registry);

        BookkeeperProtocol.Request v3Req = BookkeeperProtocol.Request.newBuilder()
            .setHeader(BKPacketHeader.newBuilder()
                .setVersion(ProtocolVersion.VERSION_THREE)
                .setTxnId(1L)
                .setOperation(OperationType.ADD_ENTRY)
                .build())
            .setAddRequest(BookkeeperProtocol.AddRequest.newBuilder()
                .setLedgerId(1L)
                .setEntryId(2L)
                .setMasterKey(ByteString.copyFrom("", UTF_8))
                .setFlag(Flag.RECOVERY_ADD)
                .setBody(ByteString.copyFrom("test", UTF_8)))
            .build();


        v2ReqEncoder.decode((ByteBuf) v3ReqEncoder.encode(v3Req, UnpooledByteBufAllocator.DEFAULT));
    }

    @Test
    public void testV2BatchReadRequest() throws Exception {
        RequestEnDeCoderPreV3 v2ReqEncoder = new RequestEnDeCoderPreV3(registry);
        BookieProtocol.BatchedReadRequest req = BookieProtocol.BatchedReadRequest.create(
                BookieProtocol.CURRENT_PROTOCOL_VERSION, 1L, 1L, FLAG_NONE, null, 1L, 10, 1024L);
        ByteBuf buf = (ByteBuf) v2ReqEncoder.encode(req, UnpooledByteBufAllocator.DEFAULT);
        buf.readInt(); // Skip the frame size.
        BookieProtocol.BatchedReadRequest reqDecoded = (BookieProtocol.BatchedReadRequest) v2ReqEncoder.decode(buf);
        assertEquals(req.ledgerId, reqDecoded.ledgerId);
        assertEquals(req.entryId, reqDecoded.entryId);
        assertEquals(req.maxSize, reqDecoded.maxSize);
        assertEquals(req.maxCount, reqDecoded.maxCount);
        reqDecoded.recycle();
    }

    @Test
    public void testV2BatchReadResponse() throws Exception {
        ResponseEnDeCoderPreV3 v2ReqEncoder = new ResponseEnDeCoderPreV3(registry);
        ByteBuf first = UnpooledByteBufAllocator.DEFAULT.buffer(4).writeInt(10);
        ByteBuf second = UnpooledByteBufAllocator.DEFAULT.buffer(8).writeLong(10L);
        ByteBufList data = ByteBufList.get(first, second);
        BookieProtocol.BatchedReadResponse res = new BookieProtocol.BatchedReadResponse(
                BookieProtocol.CURRENT_PROTOCOL_VERSION, 1, 1L, 1L, 1L, data);
        ByteBuf buf = (ByteBuf) v2ReqEncoder.encode(res, UnpooledByteBufAllocator.DEFAULT);
        buf.readInt(); // Skip the frame size.
        BookieProtocol.BatchedReadResponse resDecoded = (BookieProtocol.BatchedReadResponse) v2ReqEncoder.decode(buf);
        assertEquals(res.ledgerId, resDecoded.ledgerId);
        assertEquals(res.entryId, resDecoded.entryId);
        assertEquals(res.getData().size(), resDecoded.getData().size());
        assertEquals(res.getData().readableBytes(), resDecoded.getData().readableBytes());
    }

}