/*
 * Copyright (c) 2016-2022 chronicle.software
 *
 *     https://chronicle.software
 *
 * Licensed 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 net.openhft.chronicle.bytes;

import net.openhft.chronicle.core.Jvm;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;

import static org.junit.Assert.assertEquals;
import static org.junit.Assume.assumeFalse;

@RunWith(Parameterized.class)
public class StreamingDataInputTest extends BytesTestCommon {

    private BytesType bytesType;

    public StreamingDataInputTest(BytesType bytesType) {
        this.bytesType = bytesType;
    }

    @Parameterized.Parameters(name = "bytesType={0}")
    public static Object[] params() {
        return Arrays.stream(BytesType.values()).toArray();
    }

    @Test
    public void read() {
        Bytes<?> b = bytesType.createBuffer();
        b.append("0123456789");
        byte[] byteArr = "ABCDEFGHIJKLMNOP".getBytes();
        b.read(byteArr, 2, 6);
        assertEquals("AB012345IJKLMNOP", new String(byteArr, StandardCharsets.ISO_8859_1));
        assertEquals('6', b.readByte());
        b.releaseLast();
    }

    @Test
    public void roundTripWorksOnHeap() {
        assumeFalse(Jvm.isAzulZing());
        Bytes<?> b = bytesType.createBuffer();
        TestObject source = new TestObject(123L, 123, false);
        b.unsafeWriteObject(source, 13);
        TestObject dest = new TestObject();
        b.unsafeReadObject(dest, 13);
        assertEquals(source, dest);
    }

    @Test
    public void readWithLength() {
        int max = 130; // two bytes of length for a stop bit encoded length
        Bytes bytes = Bytes.allocateElasticOnHeap(max + 2);
        Bytes from = Bytes.wrapForRead(new byte[max]);
        Bytes to = Bytes.wrapForRead(new byte[max]);
        for (int len = 0; len <= max; len++) {
            from.readPositionRemaining(0, len);
            bytes.clear();
            bytes.writeWithLength(from);
            assertEquals(len + (len < 128 ? 1 : 2), bytes.readRemaining());
            bytes.readWithLength(to);
            assertEquals(len, to.readRemaining());
        }
    }

    private enum BytesType implements BytesFactory {
        DIRECT {
            @Override
            public Bytes<?> createBuffer() {
                return Bytes.allocateElasticDirect();
            }
        },
        ON_HEAP {
            @Override
            public Bytes<?> createBuffer() {
                return Bytes.allocateElasticOnHeap();
            }
        }
    }

    interface BytesFactory {
        Bytes<?> createBuffer();
    }

    static class TestObject {
        long l1;
        long i1;
        boolean b1;

        public TestObject() {
        }

        public TestObject(long l1, int i1, boolean b1) {
            this.l1 = l1;
            this.i1 = i1;
            this.b1 = b1;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            TestObject that = (TestObject) o;
            return l1 == that.l1 && i1 == that.i1 && b1 == that.b1;
        }

        @Override
        public int hashCode() {
            return Objects.hash(l1, i1, b1);
        }

        @Override
        public String toString() {
            return "TestObject{" +
                    "l1=" + l1 +
                    ", i1=" + i1 +
                    ", b1=" + b1 +
                    '}';
        }
    }
}
@@ -1104,35 +1104,35 @@
 private 
-BytesType bytesType
+Allocator allocator
 ;%0A%0A    p
@@ -1164,65 +1164,65 @@
 est(
-BytesType bytesType) %7B%0A        this.bytesType = bytesType
+Allocator allocator) %7B%0A        this.allocator = allocator
 ;%0A  
@@ -1264,25 +1264,25 @@
 name = %22
-bytesType
+allocator
 =%7B0%7D%22)%0A 
@@ -1347,25 +1347,25 @@
 .stream(
-BytesType
+Allocator
 .values(
@@ -1441,39 +1441,395 @@
 %3C?%3E b = 
-bytesType.createBuffer(
+allocator.elasticBytes(32);%0A        b.append(%220123456789%22);%0A        byte%5B%5D byteArr = %22ABCDEFGHIJKLMNOP%22.getBytes();%0A        b.readPosition(3);%0A        b.read(byteArr);%0A        assertEquals(%223456789HIJKLMNOP%22, new String(byteArr, StandardCharsets.ISO_8859_1));%0A        b.releaseLast();%0A    %7D%0A%0A    @Test%0A    public void readOffset() %7B%0A        Bytes%3C?%3E b = allocator.elasticBytes(32
 );%0A     
@@ -2222,31 +2222,33 @@
 b = 
-bytesType.createBuffer(
+allocator.elasticBytes(32
 );%0A 
@@ -2467,16 +2467,41 @@
  dest);%0A
+        b.releaseLast();%0A
     %7D%0A%0A 
@@ -3137,463 +3137,8 @@
  %7D%0A%0A
-    private enum BytesType implements BytesFactory %7B%0A        DIRECT %7B%0A            @Override%0A            public Bytes%3C?%3E createBuffer() %7B%0A                return Bytes.allocateElasticDirect();%0A            %7D%0A        %7D,%0A        ON_HEAP %7B%0A            @Override%0A            public Bytes%3C?%3E createBuffer() %7B%0A                return Bytes.allocateElasticOnHeap();%0A            %7D%0A        %7D%0A    %7D%0A%0A    interface BytesFactory %7B%0A        Bytes%3C?%3E createBuffer();%0A    %7D%0A%0A
     

test StreamingDataInput#read(byte[])