package com.paritytrading.nassau.soupbintcp;

import static com.paritytrading.nassau.soupbintcp.SoupBinTCP.*;
import static org.jvirtanen.nio.ByteBuffers.*;

import com.paritytrading.nassau.util.Clock;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;

/**
 * The base for both the client and server side of the protocol.
 */
public abstract class SoupBinTCPSession implements Closeable {

    private static final long RX_HEARTBEAT_TIMEOUT_MILLIS  = 15000;
    private static final long TX_HEARTBEAT_INTERVAL_MILLIS =  1000;

    private Clock clock;

    private SocketChannel channel;

    /*
     * This variable is written on data reception and read on session
     * keep-alive. These two functions can run on different threads
     * without locking.
     */
    private volatile long lastRxMillis;

    /*
     * This variable is written on data transmission and read on session
     * keep-alive. These two functions can run on different threads but
     * require locking.
     */
    private long lastTxMillis;

    private ByteBuffer rxBuffer;

    private   ByteBuffer txHeader;
    protected ByteBuffer txPayload;

    private ByteBuffer[] txBuffers;

    private ByteBuffer txHeartbeat;

    protected SoupBinTCPSession(Clock clock, SocketChannel channel, int maxPayloadLength,
            byte heartbeatPacketType) {
        this.clock   = clock;
        this.channel = channel;

        this.lastRxMillis = clock.currentTimeMillis();
        this.lastTxMillis = clock.currentTimeMillis();

        this.rxBuffer = ByteBuffer.allocate(3 + Math.min(maxPayloadLength, MAX_PACKET_LENGTH - 1));

        /*
         * The built-in payload transmit buffer is used for Login Accepted,
         * Login Rejected, End of Session, Login Request and Logout Request
         * packets.
         */
        this.txHeader  = ByteBuffer.allocate(3);
        this.txPayload = ByteBuffer.allocate(46);

        this.txBuffers = new ByteBuffer[2];

        this.txHeartbeat = ByteBuffer.allocate(3);

        this.txHeader.order(ByteOrder.BIG_ENDIAN);
        this.txPayload.order(ByteOrder.BIG_ENDIAN);
        this.txHeartbeat.order(ByteOrder.BIG_ENDIAN);

        this.txBuffers[0] = txHeader;

        putUnsignedShort(this.txHeartbeat, 1);
        this.txHeartbeat.put(heartbeatPacketType);
    }

    /**
     * Get the underlying socket channel.
     *
     * @return the underlying socket channel
     */
    public SocketChannel getChannel() {
        return channel;
    }

    /**
     * Receive data from the underlying socket channel. For each packet
     * received, invoke the corresponding listener if applicable.
     *
     * @return The number of bytes read, possibly zero, or <code>-1</code>
     *   if the channel has reached end-of-stream
     * @throws IOException if an I/O error occurs
     */
    public int receive() throws IOException {
        int bytes = channel.read(rxBuffer);

        if (bytes <= 0)
            return bytes;

        rxBuffer.flip();

        while (parse());

        rxBuffer.compact();

        receivedData();

        return bytes;
    }

    private boolean parse() throws IOException {
        if (rxBuffer.remaining() < 2)
            return false;

        rxBuffer.mark();

        rxBuffer.order(ByteOrder.BIG_ENDIAN);

        int packetLength = getUnsignedShort(rxBuffer);
        if (packetLength > rxBuffer.capacity() - 2)
            throw new SoupBinTCPException("Packet length exceeds buffer capacity");

        if (rxBuffer.remaining() < packetLength) {
            rxBuffer.reset();
            return false;
        }

        byte packetType = rxBuffer.get();

        int limit = rxBuffer.limit();

        rxBuffer.limit(rxBuffer.position() + packetLength - 1);

        packet(packetType, rxBuffer);

        rxBuffer.position(rxBuffer.limit());
        rxBuffer.limit(limit);

        return true;
    }

    /**
     * Keep the session alive.
     *
     * <p>If the heartbeat interval duration has passed since the last packet
     * was sent, send a Heartbeat packet. If the heartbeat timeout duration
     * has passed since the last packet was received, invoke the corresponding
     * method on the status listener.</p>
     *
     * @throws IOException if an I/O error occurs
     */
    public void keepAlive() throws IOException {
        long currentTimeMillis = clock.currentTimeMillis();

        if (currentTimeMillis - lastRxMillis > RX_HEARTBEAT_TIMEOUT_MILLIS)
            handleHeartbeatTimeout();
        else if (currentTimeMillis - lastTxMillis > TX_HEARTBEAT_INTERVAL_MILLIS)
            sendHeartbeat();
    }

    /**
     * Close the underlying socket channel.
     *
     * @throws IOException if an I/O error occurs
     */
    @Override
    public void close() throws IOException {
        channel.close();
    }

    protected abstract void heartbeatTimeout() throws IOException;

    protected abstract void packet(byte packetType, ByteBuffer payload) throws IOException;

    protected void send(byte packetType) throws IOException {
        txPayload.clear();
        txPayload.flip();

        send(packetType, txPayload);
    }

    protected void send(byte packetType, ByteBuffer payload) throws IOException {
        int packetLength = payload.remaining() + 1;

        if (packetLength > MAX_PACKET_LENGTH)
            throw new SoupBinTCPException("Packet length exceeds maximum packet length");

        txHeader.clear();
        putUnsignedShort(txHeader, packetLength);
        txHeader.put(packetType);
        txHeader.flip();

        txBuffers[1] = payload;

        int remaining = txHeader.remaining() + payload.remaining();

        do {
            remaining -= channel.write(txBuffers);
        } while (remaining > 0);

        sentData();
    }

    protected void unexpectedPacketType(byte packetType) throws SoupBinTCPException {
        throw new SoupBinTCPException("Unexpected packet type: " + (char)packetType);
    }

    private void sendHeartbeat() throws IOException {
        txHeartbeat.flip();

        do {
            channel.write(txHeartbeat);
        } while (txHeartbeat.remaining() > 0);

        sentData();
    }

    private void handleHeartbeatTimeout() throws IOException {
        heartbeatTimeout();

        receivedData();
    }

    private void receivedData() {
        lastRxMillis = clock.currentTimeMillis();
    }

    private void sentData() {
        lastTxMillis = clock.currentTimeMillis();
    }

}

@@ -1274,30 +1274,32 @@
 ate 
-ByteBuffer txHeartbeat
+byte heartbeatPacketType
 ;%0A%0A 
@@ -1836,21 +1836,69 @@
 on, 
-Login Request
+Server Heartbeat, Login Request,%0A         * Client Heartbeat,
  and
@@ -1912,27 +1912,16 @@
  Request
-%0A         *
  packets
@@ -2083,60 +2083,8 @@
 %5D;%0A%0A
-        this.txHeartbeat = ByteBuffer.allocate(3);%0A%0A
     
@@ -2185,62 +2185,8 @@
 AN);
-%0A        this.txHeartbeat.order(ByteOrder.BIG_ENDIAN);
 %0A%0A  
@@ -2234,41 +2234,22 @@
     
-putUnsignedShort(
 this.
-txH
+h
 eartbeat
 , 1)
@@ -2248,43 +2248,21 @@
 beat
-, 1);%0A        this.txHeartbeat.put(
+PacketType = 
 hear
@@ -2268,33 +2268,32 @@
 rtbeatPacketType
-)
 ;%0A    %7D%0A%0A    /**
@@ -4575,25 +4575,26 @@
     send
-H
+(h
 eartbeat
 ();%0A    
@@ -4585,17 +4585,26 @@
 eartbeat
-(
+PacketType
 );%0A    %7D
@@ -5952,219 +5952,8 @@
  %7D%0A%0A
-    private void sendHeartbeat() throws IOException %7B%0A        txHeartbeat.flip();%0A%0A        do %7B%0A            channel.write(txHeartbeat);%0A        %7D while (txHeartbeat.remaining() %3E 0);%0A%0A        sentData();%0A    %7D%0A%0A
     

Clean up SoupBinTCP session keep-alive