package com.jscape.util.io;

import com.jscape.util.ArrayTools;
import com.jscape.util.Time;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import static com.jscape.util.Parameters.assertNotNull;

/**
 * @author Alex Usun
 */
public class BlockBuffer {

    private static final Time AWAIT_TIMEOUT = Time.millis(50);

    private final BlockingQueue<Block> blocks;
    private volatile boolean closed;
    private Block currentBlock;

    public BlockBuffer(int maxBlocks) {
        this.blocks = new LinkedBlockingQueue<>(maxBlocks);
    }

    public BlockBuffer() {
        this(Integer.MAX_VALUE);
    }

    public int get(byte[] buffer, int offset, int length)
            throws IOException {
        return currentBlockValid() || adjustCurrentBlock() ? this.currentBlock.get(buffer, offset, length) : -1;
    }

    public void put(byte[] data, int offset, int length)
            throws IOException {
        boolean offered = false;

        while (!this.closed && !offered) {
            try {
                Block block = blockFor(data, offset, length);
                offered = this.blocks.offer(block, AWAIT_TIMEOUT.value, AWAIT_TIMEOUT.unit);
            } catch (InterruptedException e) {
                // ignore
            }
        }

        if (this.closed && !offered) {
            throw new IOException("Stream is closed.");
        }
    }

    public void close() {
        this.closed = true;
    }

    private boolean adjustCurrentBlock() {
        if (this.closed && this.blocks.isEmpty()) {
            return false;
        }

        do {
            try {
                this.currentBlock = this.blocks.poll(AWAIT_TIMEOUT.value, AWAIT_TIMEOUT.unit);
            } catch (InterruptedException e) {
                // ignore
            }
        } while (!currentBlockValid() && !this.closed);

        return currentBlockValid();
    }

    private Block blockFor(byte[] data, int offset, int length) {
        return new Block(data, offset, length);
    }

    private boolean currentBlockValid() {
        return this.currentBlock != null && !this.currentBlock.isEmpty();
    }

    private boolean currentBlockNotValid() {
        return this.currentBlock == null || this.currentBlock.isEmpty();
    }

    private void awaitIsEmpty() {
        while (!this.closed && !this.blocks.isEmpty()) {
            AWAIT_TIMEOUT.sleepSafe();
        }
    }

    public static class Block {

        public final byte[] data;
        public int length;
        public int offset;

        private Block(byte[] data, int offset, int length) {
            this.data = ArrayTools.copy(data, offset, new byte[length]);
            this.length = length;
            this.offset = offset;
        }

        public boolean isEmpty() {
            return this.offset >= this.length;
        }

        public int get(byte[] buffer, int offset, int length) {
            int read = Math.min(this.length - this.offset, length);
            System.arraycopy(this.data, this.offset, buffer, offset, read);
            this.offset += read;

            return read;
        }

        @Override
        public String toString() {
            return "Entry {"
                    + "length=" + this.length
                    + ", offset=" + this.offset
                    + '}';
        }
    }

    public static class BufferInputStream
            extends InputStream {

        private final BlockBuffer buffer;
        private byte[] oneByteBuffer;

        public BufferInputStream(BlockBuffer buffer) {
            assertNotNull(buffer);
            this.buffer = buffer;
        }

        @Override
        public int read()
                throws IOException {
            if (this.oneByteBuffer == null) {
                this.oneByteBuffer = new byte[1];
            }

            int read;

            do {
                read = read(this.oneByteBuffer, 0, 1);
            } while (read == 0);

            return read != -1 ? this.oneByteBuffer[0] & 0xFF : -1;
        }

        @Override
        public int read(byte[] buffer, int offset, int length)
                throws IOException {
            return this.buffer.get(buffer, offset, length);
        }

        @Override
        public void close() {
            this.buffer.close();
        }
    }

    public static class BufferOutputStream
            extends OutputStream {

        private final BlockBuffer buffer;
        private byte[] oneByteBuffer;

        public BufferOutputStream(BlockBuffer buffer) {
            assertNotNull(buffer);
            this.buffer = buffer;
        }

        @Override
        public void write(int b)
                throws IOException {
            if (this.oneByteBuffer == null) {
                this.oneByteBuffer = new byte[1];
            }

            this.oneByteBuffer[0] = (byte) b;
            write(this.oneByteBuffer);
        }

        @Override
        public void write(byte[] buffer, int offset, int length)
                throws IOException {
            this.buffer.put(buffer, offset, length);
        }

        @Override
        public void flush()
                throws IOException {
            this.buffer.awaitIsEmpty();
        }

        @Override
        public void close() {
            this.buffer.close();
        }
    }

}
