package org.apache.bookkeeper.bookie;

import static org.junit.Assert.assertEquals;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Random;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

@RunWith(Parameterized.class)
public class BufferedChannelTest {

    private static final String WRAPPED_FILE_PATH = System.getProperty("user.dir") + "/bktest.txt";
    private static final int BUFFERED_CHANNEL_CAPACITY = 100;
    private static final int RANDOM_DATA_LENGTH = 300;

    private final int readPos;
    private final int readLen;

    @Parameterized.Parameters
    public static Iterable<Object[]> params() {
        return Arrays.asList(
                new Object[][] {
                        { 10, 20 },
                        { 30, 40 },
                        { 70, 230 },
                        { 50, 200 },
                        { 43, 256 },

                        { 256, 43 },
                        //{ 43, 257 },

                        { 289, 10 },
                        //{ 9, 290 },

                        { 299, 0 },
                        //{ 0, 300 },

                        { 259, 40 },
                        //{ 40, 260 },

                        { 259, 30 },
                        //{ 30, 260 },
                }
        );
    }

    public BufferedChannelTest(int readPos, int readLen) {
        this.readPos = readPos;
        this.readLen = readLen;
    }

    @Test
    public void test() throws IOException {
        ByteBuf destBuf = allocBuf();
        String randomData = generateString(RANDOM_DATA_LENGTH);

        try(RandomAccessFile raf = new RandomAccessFile(WRAPPED_FILE_PATH, "rw")) {
            try (BufferedChannel bufChannel = new BufferedChannel(
                    ByteBufAllocator.DEFAULT,
                    raf.getChannel(),
                    BUFFERED_CHANNEL_CAPACITY)) {
                bufChannel.write(allocBuf(randomData));
                bufChannel.read(destBuf, readPos, readLen);
            }
        }

        assertEquals(
                randomData.substring(readPos, readPos + readLen),
                bufContentToString(readLen, destBuf));
    }

    protected static ByteBuf allocBuf(String data) {
        byte[] b = data.getBytes(StandardCharsets.UTF_8);
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
        buf.writeBytes(b);

        return buf;
    }

    protected static ByteBuf allocBuf() {
        return ByteBufAllocator.DEFAULT.buffer();
    }

    protected static String bufContentToString(int size, ByteBuf buf) {
        byte[] b = new byte[size];
        buf.getBytes(0, b);

        return new String(b);
    }

    protected static String concatArray(String[] strs) {
        StringBuilder builder = new StringBuilder();

        for(final String str : strs) {
            builder.append(str);
        }

        return builder.toString();
    }

    protected static String generateString(int len) {
        StringBuilder s = new StringBuilder();

        for(int i = 1; i <= len; ++i) {
            s.append(randomChar());
        }

        return s.toString();
    }

    protected static char randomChar() {
        final Random random = new Random();
        final String alphabet = "qwertyuiopasdfghjklzxcvbnm1234567890";

        return alphabet.charAt(random.nextInt(alphabet.length()));
    }
}