LspConnection.java

Vá para a documentação deste arquivo.

package lsp;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.atomic.AtomicInteger;

class LspConnection {
    private final short id;
    private final long sockId;
    private final ConnectionTriggers triggers;

    private volatile boolean closed;
    private volatile boolean markClosed;
    private volatile short seqNum;
    private volatile long receivedTime;
    private volatile short receivedSeqNum;
    private final AtomicInteger sendMissing;
    private final Object lock = new Object();

    private volatile InternalPack sentMessage;
    private final SocketAddress sockAddr;
    private final Thread statusThread;

    LspConnection(short id, long sockId, SocketAddress sockAddr, LspParams params, ConnectionTriggers triggers) {
        if (sockAddr == null || params == null)
            throw new NullPointerException("Nenhum parâmetro pode ser nulo");

        this.id = id;
        this.sockId = sockId;
        this.sockAddr = sockAddr;
        this.triggers = triggers;
        this.closed = false;
        this.seqNum = 0;
        this.receivedTime = -1;
        this.receivedSeqNum = -1;
        this.sendMissing = new AtomicInteger(0);

        this.statusThread = new Thread(new StatusChecker(params));
        this.statusThread.setDaemon(true);
        this.statusThread.start();
    }

    LspConnection(short id, SocketAddress sockAddr, LspParams params, ConnectionTriggers triggers) {
        this(id, uniqueSockId(sockAddr), sockAddr, params, triggers);
    }

    short getId() {
        return this.id;
    }

    static long uniqueSockId(SocketAddress sockAddr) {
        final InetSocketAddress addr = (InetSocketAddress) sockAddr;
        final int ip = addr.getAddress().hashCode();
        final int port = addr.getPort();
        return (ip & 0xffff_ffffL) << 16 | (short) port;
    }

    long getSockId() {
        return this.sockId;
    }

    SocketAddress getSockAddr() {
        return this.sockAddr;
    }

    void incSendMissing() {
        this.sendMissing.incrementAndGet();
    }

    int getSendMissing() {
        return this.sendMissing.intValue();
    }

    InternalPack sent() {
        return this.sentMessage;
    }

    InternalPack sent(Pack pack) {
        synchronized (lock) {
            InternalPack p = new InternalPack(this, ++seqNum, pack.getPayload());
            if (this.sentMessage == null) {
                this.sentMessage = p;
                return p;
            }
        }

        return null;
    }

    void ack(short seqNum) {
        // Atualiza o momento de recebimento
        received();

        synchronized (lock) {
            // Marca dados como recebidos, se o número de sequência é igual ao atual
            if (this.seqNum == seqNum) {
                this.sentMessage = null;
            }

            // Diminuição da quantidade de mensagens faltando entregar.
            this.sendMissing.decrementAndGet();
        }
    }

    short receivedSeqNum() {
        return receivedSeqNum;
    }

    void received() {
        this.receivedTime = System.currentTimeMillis();
    }

    void received(short seqNum) {
        // Atualiza o momento de recebimento
        received();
        // Altera o número de sequência atual
        this.receivedSeqNum = seqNum;
    }

    boolean isInterrupted() {
        return this.closed;
    }

    boolean isClosed() {
        return this.closed || this.markClosed;
    }

    void close() {
        close(true);
    }

    void close(boolean interrupt) {
        if (interrupt) {
            this.statusThread.interrupt();
            this.closed = true;
        } else {
            this.markClosed = true;
        }
    }

    private final class StatusChecker implements Runnable {
        private final LspParams params;

        private StatusChecker(LspParams params) {
            this.params = params;
        }

        @Override
        public void run() {
            // Obtém o horário da última mensagem recebida em milisegundos
            long lastTime = receivedTime;

            // Obtém parâmetros da conexão
            int limit = params.getEpochLimit();
            final int epoch = params.getEpoch();

            // Monitora a conexão continuamente até que o limite de épocas seja
            // atingido ou a conexão seja fechada
            while (!closed && limit-- > 0) {
                // ..ou quando a conexão está no estado de encerramento, até que
                // não haja mais mensagens para enviar.
                if (markClosed && sendMissing.get() <= 0) {
                    break;
                }

                try {
                    Thread.sleep(epoch);
                } catch (InterruptedException e) {
                    return;
                }

                // Dispara as ações da época
                triggers.doEpochActions();

                // Reinicia contagem de épocas se houve mensagens recebidas
                // desde a última época
                final long time = receivedTime;
                if (time != lastTime) {
                    lastTime = time;
                    limit = params.getEpochLimit();
                }
            }

            // Encerra formalmente a conexão
            triggers.doCloseConnection();
        }
    }
}