package com.jscape.inet.ssh.protocol.v2.connection;

import com.jscape.inet.ssh.protocol.messages.Message;
import com.jscape.inet.ssh.protocol.v2.messages.*;
import com.jscape.util.net.TransportAddress;
import com.jscape.util.net.connection.MessageConnection;
import com.jscape.util.net.connection.SyncMessageConnection;
import com.jscape.util.net.connection.SyncRawConnection;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

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

/**
 * @author Alex Usun
 */
public class ChannelConnection
        implements SyncRawConnection,
                   SshMsgGlobalRequestUnknown.Handler,
                   SshMsgChannelOpenUnknown.Handler,
                   SshMsgChannelWindowAdjust.Handler,
                   SshMsgChannelData.Handler,
                   SshMsgChannelExtendedData.Handler,
                   SshMsgChannelEof.Handler,
                   SshMsgChannelClose.Handler,
                   SshMsgChannelRequestUnknown.Handler,
                   SshMsgChannelSuccess.Handler,
                   SshMsgChannelFailure.Handler {

    private static final String LANGUAGE_TAG = "en";

    protected final SyncMessageConnection<Message> base;
    protected final int localId;
    protected final int remoteId;
    protected final Window localWindow;
    protected final long localWindowMinSize;
    protected final int localMaxPacketSize;
    protected final Window remoteWindow;
    protected final int remoteMaxPacketSize;
    protected final boolean closeConfirmationRequired;
    protected final byte[] oneByteReadingBuffer;
    protected final ReadingBuffer readingBuffer;
    protected final ReadingBuffer stderrReadingBuffer;
    protected final byte[] oneByteWritingBuffer;
    protected final Lock receiveLock;
    protected final Lock localWindowLock;
    protected final Logger logger;
    protected final AtomicBoolean closed;
    protected final AtomicBoolean closeRequested;
    protected Boolean requestSuccess;

    public ChannelConnection(SyncMessageConnection<Message> base,
                             int localId,
                             int remoteId,
                             long localInitialWindowSize,
                             int localMaxPacketSize,
                             long remoteInitialWindowSize,
                             int remoteMaxPacketSize,
                             boolean closeConfirmationRequired) {
        assertNotNull(base);
        this.base = base;

        this.localId = localId;
        this.remoteId = remoteId;

        this.localWindow = new Window(localInitialWindowSize);
        this.localWindowMinSize = localInitialWindowSize / 3;

        assertGreater(localMaxPacketSize, 0, "Bad local max packet size value.");
        this.localMaxPacketSize = localMaxPacketSize;

        this.remoteWindow = new Window(remoteInitialWindowSize);

        assertGreater(remoteMaxPacketSize, 0, "Bad remote max packet size value.");
        this.remoteMaxPacketSize = remoteMaxPacketSize;

        this.closeConfirmationRequired = closeConfirmationRequired;
        this.oneByteReadingBuffer = new byte[1];
        this.readingBuffer = new ReadingBuffer();
        this.stderrReadingBuffer = new ReadingBuffer();
        this.oneByteWritingBuffer = new byte[1];
        this.receiveLock = new ReentrantLock();
        this.localWindowLock = new ReentrantLock();
        this.closed = new AtomicBoolean(false);
        this.closeRequested = new AtomicBoolean(false);

        this.logger = Logger.getLogger("com.jscape.inet.ssh");
    }

    @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.closeRequested.getAndSet(true)) {
            return;
        }

        sendSafe(new SshMsgChannelEof(this.remoteId));
        sendSafe(new SshMsgChannelClose(this.remoteId));

        if (this.closeConfirmationRequired) {
            awaitCloseConfirmation();
        }

        this.closed.set(true);
    }

    @Override
    public int read()
            throws ConnectionException {
        int read = read(this.oneByteReadingBuffer, 0, 1);
        return read != -1 ? this.oneByteReadingBuffer[0] & 0xFF : read;
    }

    @Override
    public int read(byte[] data, int offset, int length)
            throws ConnectionException {
        return readBuffer(this.readingBuffer, data, offset, length);
    }

    @Override
    public int availableBytes() {
        return 0;
    }

    public int readStderr(byte[] data, int offset, int length)
            throws ConnectionException {
        return readBuffer(this.stderrReadingBuffer, data, offset, length);
    }

    @Override
    public void write(byte data)
            throws ConnectionException {
        this.oneByteWritingBuffer[0] = data;
        write(this.oneByteWritingBuffer, 0, 1);
    }

    @Override
    public void write(byte[] data, int offset, int length)
            throws ConnectionException {
        while (length > 0) {
            int written = sendData(data, offset, length);
            offset += written;
            length -= written;
        }
    }

    @Override
    public void flush() {
        // empty
    }

    @Override
    public void handle(SshMsgGlobalRequestUnknown message, MessageConnection<Message> connection) {
        if (message.wantReply) {
            sendSafe(new SshMsgRequestFailure());
        }
    }

    @Override
    public void handle(SshMsgChannelOpenUnknown message, MessageConnection<Message> connection) {
        sendSafe(new SshMsgChannelOpenFailure(message.senderChannel, ChannelOpenFailureReason.SSH_OPEN_UNKNOWN_CHANNEL_TYPE, "Unsupported channel type.", LANGUAGE_TAG));
    }

    @Override
    public void handle(SshMsgChannelWindowAdjust message, MessageConnection<Message> connection) {
        this.remoteWindow.increase(message.bytesToAdd);
    }

    @Override
    public void handle(SshMsgChannelData message, MessageConnection<Message> connection) {
        this.localWindowLock.lock();

        try {
            if (this.localWindow.size() > 0) {
                this.readingBuffer.put(message.data);
                this.localWindow.decrease(message.length);
            }
        } finally {
            this.localWindowLock.unlock();
        }
    }

    @Override
    public void handle(SshMsgChannelExtendedData message, MessageConnection<Message> connection) {
        if (message.dataTypeCode != DataType.SSH_EXTENDED_DATA_STDERR.code) {
            return;
        }

        this.localWindowLock.lock();

        try {
            if (this.localWindow.size() > 0) {
                this.stderrReadingBuffer.put(message.data);
                this.localWindow.decrease(message.data.length);
            }
        } finally {
            this.localWindowLock.unlock();
        }
    }

    @Override
    public void handle(SshMsgChannelEof message, MessageConnection<Message> connection) {
        this.readingBuffer.close();
        this.stderrReadingBuffer.close();
    }

    @Override
    public void handle(SshMsgChannelClose message, MessageConnection<Message> connection) {
        this.closed.set(true);
        this.readingBuffer.close();
        this.stderrReadingBuffer.close();
    }

    @Override
    public void handle(SshMsgChannelRequestUnknown message, MessageConnection<Message> connection) {
        if (message.wantReply) {
            sendSafe(new SshMsgChannelFailure(message.recipientChannel));
        }
    }

    @Override
    public void handle(SshMsgChannelSuccess message, MessageConnection<Message> connection) {
        this.requestSuccess = Boolean.TRUE;
    }

    @Override
    public void handle(SshMsgChannelFailure message, MessageConnection<Message> connection) {
        this.requestSuccess = Boolean.FALSE;
    }

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

    protected int readBuffer(ReadingBuffer buffer, byte[] data, int offset, int length)
            throws ConnectionException {
        try {
            while (buffer.empty() && !buffer.closed()) {
                this.receiveLock.lock();

                try {
                    if (buffer.empty() && !buffer.closed()) {
                        receive();
                    }
                } finally {
                    this.receiveLock.unlock();
                }
            }

            int read = buffer.read(data, offset, length);
            adjustLocalWindow();

            return read;
        } catch (Exception e) {
            return processReadingError(e);
        }
    }

    protected int sendData(byte[] data, int offset, int length)
            throws ConnectionException {
        while (this.remoteWindow.size() == 0 && !this.closed.get()) {
            if (!this.receiveLock.tryLock()) {
                continue;
            }

            try {
                if (this.remoteWindow.size() == 0 && !this.closed.get()) {
                    receive();
                }
            } finally {
                this.receiveLock.unlock();
            }
        }

        assertNotClosed();

        int bytesToSend = Math.min(this.remoteMaxPacketSize, (int) Math.min(this.remoteWindow.size(), length));
        send(new SshMsgChannelData(this.remoteId, data, offset, bytesToSend));
        this.remoteWindow.decrease(bytesToSend);

        return bytesToSend;
    }

    protected void adjustLocalWindow()
            throws ConnectionException {
        this.localWindowLock.lock();

        try {
            if (this.localWindow.size() > this.localWindowMinSize) {
                return;
            }

            int available = this.readingBuffer.available() + this.stderrReadingBuffer.available();
            long bytesToAdd = this.localWindow.space() - available;

            if (bytesToAdd > 0) {
                send(new SshMsgChannelWindowAdjust(this.remoteId, bytesToAdd));
                this.localWindow.increase(bytesToAdd);
            }
        } finally {
            this.localWindowLock.unlock();
        }
    }

    protected int processReadingError(Exception e)
            throws ConnectionException {
        if (this.closed.get()) {
            return -1;
        }

        throw wrap(e);
    }

    protected void processRequestResult(boolean wantReply)
            throws ConnectionException {
        if (wantReply) {
            awaitRequestResult();
            assertRequestSuccess();
            this.requestSuccess = null;
        }
    }

    protected void awaitRequestResult()
            throws ConnectionException {
        this.requestSuccess = null;

        while (this.requestSuccess == null) {
            this.receiveLock.lock();

            try {
                receive();
            } finally {
                this.receiveLock.unlock();
            }
        }
    }

    protected void awaitCloseConfirmation() {
        try {
            while (!this.closed.get()) {
                this.receiveLock.lock();

                try {
                    receive();
                } finally {
                    this.receiveLock.unlock();
                }
            }
        } catch (Exception e) {
            logErrorClosingChannel(e);
        }
    }

    protected void assertNotClosed()
            throws ConnectionException {
        if (this.closed.get()) {
            throw new ConnectionException("Connection is closed.");
        }
    }

    protected void assertRequestSuccess()
            throws ConnectionException {
        if (this.requestSuccess != null && !this.requestSuccess) {
            throw new ConnectionException("Channel request failure.");
        }
    }

    protected void send(Message message)
            throws ConnectionException {
        this.base.write(message);
    }

    protected void sendSafe(Message message) {
        try {
            send(message);
        } catch (Exception e) {
            logErrorSendMessage(e);
        }
    }

    @SuppressWarnings("unchecked")
    protected void receive()
            throws ConnectionException {
        Message message = this.base.read();

        try {
            message.accept(this, this.base);
        } catch (Exception e) {
            logUnsupportedMessage(message);
        }
    }

    protected void logErrorSendMessage(Throwable error) {
        if (this.logger.isLoggable(Level.WARNING)) {
            this.logger.log(Level.WARNING, "Error sending message.", error);
        }
    }

    protected void logUnsupportedMessage(Message message) {
        if (this.logger.isLoggable(Level.WARNING)) {
            this.logger.log(Level.WARNING,
                            String.format("Unsupported message at [%s <-> %s]: %s",
                                          localAddress(),
                                          remoteAddress(),
                                          message));
        }
    }

    private void logErrorClosingChannel(Throwable error) {
        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.log(Level.FINE, "Error closing channel.", error);
        }
    }
}
