package com.jscape.util.net.connection;

import com.jscape.util.io.ByteArrayInputStream;
import com.jscape.util.io.ByteArrayOutputStream;
import com.jscape.util.io.DataCodec;
import com.jscape.util.net.TransportAddress;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.jscape.util.Parameters.assertNotNull;
import static com.jscape.util.net.connection.Connection.ConnectionException.wrap;

/**
 * @author Alex Usun
 */
public class AsyncMessageConnectionAsyncRawBase<M>
        implements AsyncMessageConnection<M>,
                   MessageCodecConnection<M>,
                   AsyncRawConnection.Listener {

    private final AsyncRawConnection base;
    private DataCodec<M> messageCodec;
    private final ByteArrayOutputStream writeBuffer;
    private final ByteArrayOutputStream readBuffer;
    private final ByteArrayInputStream in;
    private final AtomicBoolean closed;
    private Listener<M> listener;

    public AsyncMessageConnectionAsyncRawBase(AsyncRawConnection base, DataCodec<M> messageCodec) {
        assertNotNull(base);
        this.base = base;

        assertNotNull(messageCodec);
        this.messageCodec = messageCodec;

        this.writeBuffer = new ByteArrayOutputStream();
        this.readBuffer = new ByteArrayOutputStream();
        this.in = new ByteArrayInputStream(this.readBuffer.buffer());
        this.closed = new AtomicBoolean(false);
    }

    @Override
    public long creationTime() {
        return this.base.creationTime();
    }

    @Override
    public TransportAddress localAddress() {
        return this.base.localAddress();
    }

    @Override
    public TransportAddress remoteAddress() {
        return this.base.remoteAddress();
    }

    @Override
    public Map<Object, Object> attributes() {
        return this.base.attributes();
    }

    @Override
    public boolean closed() {
        return this.closed.get();
    }

    @Override
    public void close() {
        if (this.closed.getAndSet(true)) {
            return;
        }

        this.base.close();

        if (this.listener != null) {
            this.listener.onClosed(this);
        }
    }

    @Override
    public void start(Listener<M> listener)
            throws ConnectionException {
        assertNotNull(listener);
        this.listener = listener;

        this.base.start(this);
    }

    @Override
    public void write(M message)
            throws ConnectionException {
        try {
            this.messageCodec.write(message, this.writeBuffer);
            this.base.write(this.writeBuffer.buffer(), 0, this.writeBuffer.position());
            this.base.flush();
        } catch (Exception e) {
            throw wrap(e);
        } finally {
            this.writeBuffer.reset();
        }
    }

    @Override
    public void onData(AsyncRawConnection connection, byte[] data, int offset, int length) {
        this.readBuffer.write(data, offset, length);
        this.in.put(this.readBuffer.buffer(), 0, this.readBuffer.position());

        try {
            M message;
            while ((message = nextMessage()) != null) {
                this.listener.onMessage(this, message);
            }
        } catch (Exception e) {
            if (!this.closed.get()) {
                this.listener.onError(this, e);
            }
        }
    }

    @Override
    public void onEof(AsyncRawConnection connection) {
        if (!this.closed.get()) {
            this.listener.onEof(this);
        }
    }

    @Override
    public void onError(AsyncRawConnection connection, Throwable error) {
        if (!this.closed.get()) {
            this.listener.onError(this, error);
        }
    }

    @Override
    public void onClosed(AsyncRawConnection connection) {
        if (!this.closed.get()) {
            this.listener.onError(this, new ConnectionClosedException());
        }
    }

    @Override
    public void onIdle(AsyncRawConnection connection) {
        if (!this.closed.get()) {
            this.listener.onIdle(this);
        }
    }

    public AsyncRawConnection base() {
        return this.base;
    }

    public DataCodec<M> messageCodec() {
        return this.messageCodec;
    }

    public void messageCodec(DataCodec<M> value) {
        assertNotNull(value);
        this.messageCodec = value;
    }

    @Override
    public String toString() {
        return this.base.toString();
    }

    private M nextMessage()
            throws IOException {
        try {
            if (this.in.available() == 0) {
                return null;
            }

            M message = this.messageCodec.read(this.in);

            int startPosition = this.readBuffer.position() - this.in.available();
            this.readBuffer.relocate(startPosition);
            this.in.put(this.readBuffer.buffer(), 0, this.readBuffer.position());

            return message;
        } catch (EOFException | java.io.EOFException e) {
            return null;
        }
    }
}
