magicbane/src/engine/net/AbstractConnection.java

// • ▌ ▄ ·.  ▄▄▄·  ▄▄ • ▪   ▄▄· ▄▄▄▄·  ▄▄▄·  ▐▄▄▄  ▄▄▄ .
// ·██ ▐███▪▐█ ▀█ ▐█ ▀ ▪██ ▐█ ▌▪▐█ ▀█▪▐█ ▀█ •█▌ ▐█▐▌·
// ▐█ ▌▐▌▐█·▄█▀▀█ ▄█ ▀█▄▐█·██ ▄▄▐█▀▀█▄▄█▀▀█ ▐█▐ ▐▌▐▀▀▀
// ██ ██▌▐█▌▐█ ▪▐▌▐█▄▪▐█▐█▌▐███▌██▄▪▐█▐█ ▪▐▌██▐ █▌▐█▄▄▌
// ▀▀  █▪▀▀▀ ▀  ▀ ·▀▀▀▀ ▀▀▀·▀▀▀ ·▀▀▀▀  ▀  ▀ ▀▀  █▪ ▀▀▀
//      Magicbane Emulator Project © 2013 - 2022
//                www.magicbane.com


package engine.net;

import engine.job.JobManager;
import engine.net.client.Protocol;
import engine.net.client.msg.ClientNetMsg;
import org.pmw.tinylog.Logger;

import java.io.IOException;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

public abstract class AbstractConnection implements
        NetMsgHandler {

    protected static final long SOCKET_FULL_WRITE_DELAY = 50L; //wait one second to write on full socket
    private static final String error01 = "A byte buffer is being free()ed that is not of the size stored in ByteBufferPool. Size: %1%";
    private static final String error02 = "(IP=%1%): Socket has reached the maximum outbound buffer length of %2%. Moving on while we wait for data to be sent.";
    private static final int OPCODEHASH = 1016;    //Opcodes are unique modulo this number as of 11/21/10
    protected final SocketChannel sockChan;
    protected final AbstractConnectionManager connMan;
    protected final NetMsgFactory factory;
    protected final AtomicBoolean execTask = new AtomicBoolean(false);
    protected final ReentrantLock writeLock = new ReentrantLock();
    protected final ReentrantLock readLock = new ReentrantLock();
    protected long lastMsgTime = System.currentTimeMillis();
    protected long lastKeepAliveTime = System.currentTimeMillis();
    protected long lastOpcode = -1;
    protected ConcurrentLinkedQueue<ByteBuffer> outbox = new ConcurrentLinkedQueue<>();
    protected ByteBuffer outBuf = null;
    protected ConcurrentHashMap<Long, Byte> cacheList;
    protected long nextWriteTime = 0L;

    //Opcode tracking
    protected Protocol lastProtocol = Protocol.NONE;

    public AbstractConnection(AbstractConnectionManager connMan,
                              SocketChannel sockChan, boolean clientMode) {
        this.connMan = connMan;
        this.sockChan = sockChan;
        this.factory = new NetMsgFactory(this);

    }

    public void disconnect() {
        try {
            this.sockChan.close();
        } catch (IOException e) {
            Logger.error(e.toString());
        }
    }

    /**
     * Serializes AbstractNetMsg <i>msg</i> into a ByteBuffer, then queues that
     * ByteBuffer for sending on this AbstractConnection's SocketChannel.
     *
     * @param msg - AbstractNetMsg to be sent.
     * @return boolean status if queue is successful or not. On false return,
     * the field <i>lastError</i> will be set.
     */
    public final boolean sendMsg(AbstractNetMsg msg) {
        //		Logger.info("Send: " + msg.getSimpleClassName());
        try {
            return this._sendMsg(msg);

        } catch (Exception e) { // Catch-all
            Logger.error(e);
            return false;
        }
    }

    /**
     * Serializes AbstractNetMsg <i>msg</i> into a ByteBuffer, then queues that
     * ByteBuffer for sending on this AbstractConnection's SocketChannel. This
     * internal function is <b>required</b> to be overridden by subclasses.
     *
     * @param msg - AbstractNetMsg to be sent.
     * @return boolean status if queue is successful or not. On false return,
     * the field <i>lastError</i> will be set.
     */
    protected abstract boolean _sendMsg(AbstractNetMsg msg);

    /**
     * Queues ByteBuffer <i>bb</i> for sending on this AbstractConnection's
     * SocketChannel.
     *
     * @param bb - ByteBuffer to be sent.
     * @return boolean status if queue is successful or not. On false return,
     * the field <i>lastError</i> will be set.
     */
    public final boolean sendBB(ByteBuffer bb) {

        if (bb == null)
            return false;
        try {
            return this._sendBB(bb);
        } catch (Exception e) { // Catch-all
            Logger.error(e);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * Queues ByteBuffer <i>bb</i> for sending on this AbstractConnection's
     * SocketChannel. This internal function is designed to be overrideable by
     * subclasses if needed.
     *
     * @param bb - ByteBuffer to be sent.
     * @return boolean status if queue is successful or not. On false return,
     * the field <i>lastError</i> will be set.
     */
    protected boolean _sendBB(ByteBuffer bb) {
        this.outbox.add(bb);
        this.connMan.sendStart(this.sockChan);
        return true;
    }

    /**
     * Move data off the socketChannel's buffer and into the Connection's
     * internal NetMsgFactory.
     */
    // FIXME the int return value on this means nothing! Clean it up!
    protected int read() {

        if (readLock.tryLock())
            try {

                if (this.sockChan.isOpen() == false) {
                    Logger.info("Sock channel closed. Disconnecting " + this.getRemoteAddressAndPortAsString());
                    this.disconnect();
                    return 0;
                }

                // get socket buffer sized buffer from multipool
                ByteBuffer bb = Network.byteBufferPool.getBuffer(16);

                int totalBytesRead = 0;
                int lastRead;
                do {
                    try {
                        bb.clear();
                        lastRead = this.sockChan.read(bb);

                        // On EOF on the SocketChannel, disconnect.
                        if (lastRead <= -1) {
                            Logger.info(" EOF on Socket Channel " + this.getRemoteAddressAndPortAsString());
                            this.disconnect();
                            break;
                        }

                        if (lastRead == 0)
                            continue;

                        synchronized (this.factory) {
                            this.factory.addData(bb);
                        }
                        this.checkInternalFactory();

                        totalBytesRead += lastRead;

                    } catch (NotYetConnectedException e) {
                        Logger.error(e.toString());
                        break;

                    } catch (ClosedChannelException e) {
                        Logger.error(e.toString());
                        this.disconnect();
                        break;

                    } catch (IOException e) {
                        if (!e.getMessage().startsWith(
                                "An existing connection was forcibly closed"))
                            Logger.error(e.toString());
                        this.disconnect();
                        break;

                    } catch (Exception e) {
                        Logger.error(e.toString());
                        break;
                    }
                } while (lastRead > 0);

                // put buffer back into multipool
                Network.byteBufferPool.putBuffer(bb);

                this.checkInternalFactory();
                return totalBytesRead;

            } finally {
                readLock.unlock();
            }
        else {
            Logger.debug("Another thread already has a read lock! Skipping.");
            return 0;
        }
    }

    /**
     * Move data off the Connection's buffer and into the SocketChannel's
     * Buffer.
     */
    protected boolean writeAll() {

        //intentional delay if socket is full. Give the socket a chance to clear out.
        if (System.currentTimeMillis() < this.nextWriteTime)
            return false;

        if (writeLock.tryLock())
            try {
                String s = "";
                int written = 0;

                boolean allSentOK = true, bufferDoubled = false;

                while (this.outbox.peek() != null) {
                    ByteBuffer bb = this.outbox.peek();

                    int toSend = bb.position();

                    try {
                        bb.flip();
                        written = this.sockChan.write(bb);

                        // Logger.debug("Using a BB with cap of: " + bb.capacity()
                        // + ". toSend: " + toSend + ", written: " + written);
                        if (written != toSend)
                            bb.compact(); // Clean up in case not all gets sent

                        if (toSend == written) {
                            // Actually remove it from the queue
                            this.outbox.poll();

                            // Pool it
                            Network.byteBufferPool.putBuffer(bb);
                            continue;

                        } else if (written == 0) {
                            //Socket full, let's delay the next write on socket.
                            this.nextWriteTime = System.currentTimeMillis() + AbstractConnection.SOCKET_FULL_WRITE_DELAY;

                            int currentBufferSize = this.sockChan.socket()
                                    .getSendBufferSize();

                            if (!bufferDoubled && currentBufferSize <= Network.INITIAL_SOCKET_BUFFER_SIZE) {
                                this.doubleSocketSendBufferSize();
                                bufferDoubled = true;
                            } else {

                                //                                    s = error02;
                                //                                    s = s.replaceAll("%1%", this
                                //                                            .getRemoteAddressAndPortAsString() + ":" +
                                //                                            this.printLastOpcodes());
                                //                                    s = s.replaceAll("%2%", currentBufferSize + "");
                                //                                    this.warn(s);

                                allSentOK = false;
                                break;
                            }
                        }

                    } catch (ClosedChannelException e) {
                        // Catches AsynchronousCloseException,
                        // and ClosedByInterruptException
                        Logger.error(e);
                        break;

                    } catch (Exception e) {
                        // Catches NotYetConnectedException
                        // and IOException
                        Logger.error(e);
                        this.disconnect();
                        break;
                    }

                }
                return allSentOK;
            } finally {
                writeLock.unlock();
            }
        else {
            Logger.debug("Another thread already has a write lock! Skipping.");
            return false;
        }
    }

    private boolean doubleSocketSendBufferSize() {
        String s;
        try {
            int currentSize = this.sockChan.socket().getSendBufferSize();
            int newSize = currentSize << 1;

            this.sockChan.socket().setSendBufferSize(newSize);

            //            s = "(IP=" + this.getRemoteAddressAndPortAsString() + "): ";
            //            s += this.printLastOpcodes();
            //            s += "Socket has reached the maximum outbound buffer length of ";
            //            s += currentSize + ". Attempting to double the outbound buffer size.";
            //
            //            this.warn(s);
            return true;
        } catch (SocketException e) {
            Logger.error(e.toString());
            return false;
        }
    }

    public boolean isConnected() {
        return this.sockChan.isConnected();
    }

    public boolean isOpen() {
        return this.sockChan.isOpen();
    }

    /*
     * Getters n Setters
     */
    public SocketChannel getSocketChannel() {
        return this.sockChan;
    }

    public final void checkInternalFactory() {
        CheckNetMsgFactoryJob j = new CheckNetMsgFactoryJob(this);
        JobManager.getInstance().submitJob(j);
    }

    public NetMsgFactory getFactory() {
        return factory;
    }

    public String getRemoteAddressAndPortAsString() {
        String out = "";

        if (this.sockChan == null)
            out += "NotConnected";
        else if (this.sockChan.socket() == null)
            out += "NotConnected";
        else if (this.sockChan.socket().getRemoteSocketAddress() == null)
            out += "NotConnected";
        else
            out += this.sockChan.socket().getRemoteSocketAddress().toString();

        return out;
    }

    public String getLocalAddressAndPortAsString() {
        String out = "";

        if (this.sockChan == null)
            out += "NotConnected";
        else if (this.sockChan.socket() == null)
            out += "NotConnected";
        else if (this.sockChan.socket().getRemoteSocketAddress() == null)
            out += "NotConnected";
        else {
            out += this.sockChan.socket().getLocalSocketAddress().toString();
            out += ":";
            out += this.sockChan.socket().getLocalPort();
        }

        return out;
    }

    /**
     * Gives the Connection a chance to act on a msg prior to sending it to the
     * provided NetMsgHandler
     */
    @Override
    public abstract boolean handleClientMsg(ClientNetMsg msg);

    protected long getLastMsgTime() {
        return this.lastMsgTime;
    }

    protected void setLastMsgTime() {
        // TODO Consider making this a static to latest system time
        this.lastMsgTime = System.currentTimeMillis();
    }

    protected long getLastKeepAliveTime() {
        return this.lastKeepAliveTime;
    }

    protected void setLastKeepAliveTime() {
        this.lastKeepAliveTime = System.currentTimeMillis();
    }

    public long getLastOpcode() {
        return lastOpcode;
    }

    public void setLastOpcode(long lastOpcode) {
        this.lastOpcode = lastOpcode;
    }

}