// This Source Code is in the Public Domain per: http://litesoft.org/License.txt
package org.litesoft.peertopeer.nonpublic.peermanagement;

import java.io.*;
import java.net.*;
import java.util.concurrent.*;

import org.litesoft.logger.*;

public abstract class PeerConnectionHandler extends Thread
{
    public static final int MAX_SERIALIZED_OBJECT_SIZE = 63 * 1024;
    public static final int NEW_OBJECT_COMING = 22; // ASCII_SYN

    public static final Logger LOGGER = LoggerFactory.getLogger( PeerConnectionHandler.class );

    private PeerConnectionManager mConnectionManager;
    private PeerDefinition mPeerDef;
    private Socket mSocket;
    private OutputStream mOutputStream = null;
    private InputStream mInputStream = null;
    private BlockingQueue<Serializable> mSenderQueue = new LinkedBlockingQueue<Serializable>();
    private volatile boolean mClosing = false;

    protected PeerConnectionHandler( PeerConnectionManager pConnectionManager, PeerDefinition pPeerDef, Socket pSocket )
    {
        setName( getClass().getName() );
        mConnectionManager = pConnectionManager;
        mPeerDef = pPeerDef;
        mSocket = pSocket;
    }

    public PeerDefinition getPeerDef()
    {
        return mPeerDef;
    }

    protected void setPeerDef( PeerList pPeerList )
    {
        mPeerDef = pPeerList.getFrom();
    }

    public void send( Serializable pToSend )
    {
        if ( pToSend != null )
        {
            mSenderQueue.add( pToSend );
        }
    }

    public void close()
    {
        mClosing = true;
        interrupt();
    }

    protected void LLsend( Serializable pToSend )
            throws IOException
    {
        ByteOutputStream zByteStream = new ByteOutputStream();
        ObjectOutputStream zObjectStream = new ObjectOutputStream( zByteStream );
        try
        {
            zObjectStream.writeObject( pToSend );
            zObjectStream.flush();
        }
        finally
        {
            zObjectStream.close();
        }

        int count = zByteStream.getCount();
        if ( count > MAX_SERIALIZED_OBJECT_SIZE )
        {
            throw new IOException( "Serialized Object too large (" + count + " > " + MAX_SERIALIZED_OBJECT_SIZE + "), Object: " + pToSend );
        }

        try
        {
            OutputStream zPeerStream = getOutputStream();
            zPeerStream.write( NEW_OBJECT_COMING );
            zPeerStream.write( (count >> 8) & 0xFF ); // Hi Order Byte
            zPeerStream.write( count & 0xFF ); // Low Order Byte
            for ( int i = 0; i < count; i++ )
            {
                zPeerStream.write( zByteStream.readByteAt( i ) );
            }
            zPeerStream.flush();
        }
        catch ( SocketException e )
        {
            if ( isConnectionReset( e ) )
            {
                close();
                return;
            }
            throw e;
        }
    }

    protected Serializable LLread()
            throws IOException, ClassNotFoundException
    {
        try
        {
            InputStream zPeerStream = getInputStream();
            for ( int i; NEW_OBJECT_COMING != (i = zPeerStream.read()); )// Skip anything else as noise!!!
            {
                if ( i == -1 )
                {
                    return null; // EOF
                }
            }
            int zHiOrderByte = zPeerStream.read();
            int zLowOrderByte = zPeerStream.read();
            if ( (zHiOrderByte == -1) || (zLowOrderByte == -1) )
            {
                problem( new IllegalStateException( "EOF in block size" ) );
                return null; // EOF
            }
            int count = ((zHiOrderByte & 0xFF) << 8) + (zLowOrderByte & 0xFF);
            if ( count > MAX_SERIALIZED_OBJECT_SIZE )
            {
                throw new IOException( "De-Serialized Object too large (" + count + " > " + MAX_SERIALIZED_OBJECT_SIZE + ")" );
            }
            byte[] buf = new byte[count];
            for ( int i = 0; i < count; i++ )
            {
                int zByte = zPeerStream.read();
                if ( zByte == -1 )
                {
                    problem( new IllegalStateException( "EOF in block" ) );
                    return null; // EOF
                }
                buf[i] = (byte) zByte;
            }
            ObjectInputStream zObjectStream = new ObjectInputStream( new ByteArrayInputStream( buf ) );
            Object o;
            try
            {
                o = zObjectStream.readObject();
            }
            finally
            {
                zObjectStream.close();
            }
            return (Serializable) o;
        }
        catch ( SocketException e )
        {
            if ( isConnectionReset( e ) )
            {
                return null;
            }
            throw e;
        }
    }

    protected void runAsSenderThread()
    {
        try
        {
            while ( !mClosing )
            {
                try
                {
                    LLsend( mSenderQueue.take() );
                }
                catch ( InterruptedException e )
                {
                    if ( !mClosing )
                    {
                        problem( e );
                    }
                }
            }
        }
        catch ( IOException e )
        {
            problem( e );
        }
        catch ( RuntimeException e )
        {
            problem( e );
        }
    }

    protected boolean startPeering()
            throws IOException
    {
        BlockingQueue<Serializable> zQueue = mConnectionManager.startPeeringWith( this );
        if ( zQueue != null )
        {
            new ReceiverThread( zQueue ).start();
            return true;
        }
        return false;
    }

    protected void notifyConnectionManagerOf( PeerDefinition[] pPeers )
    {
        mConnectionManager.addFromPeer( pPeers );
    }

    protected boolean isOKtoRunIfIncompatibleVersionAlreadyRunning()
    {
        return mConnectionManager.isOKtoRunIfIncompatibleVersionAlreadyRunning();
    }

    protected void addFromPeer( PeerDefinition[] pPeerDefinitions )
    {
        mConnectionManager.addFromPeer( pPeerDefinitions );
    }

    private void problem( Exception e )
    {
        InetAddress zInetAddress = mSocket != null ? mSocket.getInetAddress() : null;
        LOGGER.error.log( e, zInetAddress, " from ", mPeerDef );
    }

    @Override
    protected void finalize()
            throws Throwable
    {
        releaseResources();
        super.finalize();
    }

    private void releaseResources()
    {
        mClosing = true;
        if ( mInputStream != null )
        {
            try
            {
                mInputStream.close();
            }
            catch ( IOException e )
            {
                // Whatever...
            }
            finally
            {
                mInputStream = null;
            }
        }
        if ( mOutputStream != null )
        {
            try
            {
                mOutputStream.close();
            }
            catch ( IOException e )
            {
                // Whatever...
            }
            finally
            {
                mOutputStream = null;
            }
        }
        if ( mSocket != null )
        {
            try
            {
                mSocket.close();
            }
            catch ( IOException e )
            {
                // Whatever...
            }
            finally
            {
                mSocket = null;
            }
        }
        if ( mConnectionManager != null )
        {
            try
            {
                mConnectionManager.noLongerPeeringWith( mPeerDef );
            }
            catch ( Exception e )
            {
                // Whatever...
            }
            mConnectionManager = null;
        }
    }

    protected Socket getSocket()
            throws IOException
    {
        if ( mSocket == null )
        {
            mSocket = new Socket( mPeerDef.getInetAddress(), mPeerDef.getPort() );
        }
        return mSocket;
    }

    protected OutputStream getOutputStream()
            throws IOException
    {
        if ( mOutputStream == null )
        {
            mOutputStream = new BufferedOutputStream( getSocket().getOutputStream() );
        }
        return mOutputStream;
    }

    protected InputStream getInputStream()
            throws IOException
    {
        if ( mInputStream == null )
        {
            mInputStream = new BufferedInputStream( getSocket().getInputStream() );
        }
        return mInputStream;
    }

    @Override
    public void run()
    {
        try
        {
            LLrun();
        }
        catch ( IOException e )
        {
            problem( e );
        }
        catch ( ClassNotFoundException e )
        {
            problem( e );
        }
        catch ( RuntimeException e )
        {
            problem( e );
        }
        finally
        {
            releaseResources();
        }
    }

    abstract protected void LLrun()
            throws IOException, ClassNotFoundException;

    private class ReceiverThread extends Thread
    {
        private BlockingQueue<Serializable> mReceiverQueue;

        public ReceiverThread( BlockingQueue<Serializable> pReceiverQueue )
        {
            mReceiverQueue = pReceiverQueue;
            setDaemon( true );
        }

        @Override
        public void run()
        {
            try
            {
                for ( Serializable zSerializable; null != (zSerializable = LLread()); )
                {
                    mReceiverQueue.add( zSerializable );
                }
            }
            catch ( IOException e )
            {
                problem( e );
            }
            catch ( ClassNotFoundException e )
            {
                problem( e );
            }
            catch ( RuntimeException e )
            {
                problem( e );
            }
            mReceiverQueue = null;
        }
    }

    private static class ByteOutputStream extends ByteArrayOutputStream
    {
        public ByteOutputStream()
        {
            super( 4096 );
        }

        public int getCount()
        {
            return count;
        }

        public byte readByteAt( int pOffset )
        {
            return buf[pOffset];
        }
    }

    private boolean isConnectionReset( SocketException e )
    {
        String zMessage = e.getMessage().toUpperCase();
        return zMessage.contains( "CONNECTION RESET" );
    }
}