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

import org.litesoft.core.typeutils.Objects;
import org.litesoft.logger.*;
import org.litesoft.peertopeer.*;
import org.litesoft.peertopeer.nonpublic.*;

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

public class PeerConnectionManager implements P2PConstants
{
    public static final Logger LOGGER = LoggerFactory.getLogger( PeerConnectionManager.class );

    private PeerDefinition mOurDefinition = null;
    private PeerDefs mPendingPeers = new PeerDefs();
    private ActivePeers mActivePeers = new ActivePeers();

    private MessageFromPeerHandlerManager mMessageFromPeerHandlerManager;
    private PeerManager mPeerManager;
    private boolean mOKtoRunIfIncompatibleVersionAlreadyRunning;

    private ReceivedThread mReceivedThread = null;

    public PeerConnectionManager( MessageFromPeerHandlerManager pMessageFromPeerHandlerManager, PeerManager pPeerManager, boolean pOKtoRunIfIncompatibleVersionAlreadyRunning )
    {
        Objects.assertNotNull( "MessageFromPeerHandlerManager", //
                mMessageFromPeerHandlerManager = pMessageFromPeerHandlerManager );
        Objects.assertNotNull( "PeerManager", mPeerManager = pPeerManager );
        mOKtoRunIfIncompatibleVersionAlreadyRunning = pOKtoRunIfIncompatibleVersionAlreadyRunning;
    }

    public boolean isOKtoRunIfIncompatibleVersionAlreadyRunning()
    {
        return mOKtoRunIfIncompatibleVersionAlreadyRunning;
    }

    public synchronized void setOurConnectionInfo( String pOurName, InetAddress pOurServerAddress, int pOurServerPort )
    {
        Objects.assertNotNull( "OurServerAddress", pOurServerAddress );
        if ( mOurDefinition != null )
        {
            throw new IllegalStateException( "setOurConnectionInfo() already called with: " + mOurDefinition );
        }
        mOurDefinition = new PeerDefinition( pOurName, pOurServerAddress, pOurServerPort );
    }

    public synchronized void addAnnouncedPeer( PeerDefinition pPeerDef )
    {
        if ( (pPeerDef != null) && !pPeerDef.equals( mOurDefinition ) && !mPendingPeers.contains( pPeerDef ) && !mActivePeers.contains( pPeerDef ) )
        {
            mPendingPeers.add( pPeerDef );
            PeerList zPeerList = new PeerList( mOurDefinition, mActivePeers.getPeerDefs() );
            new PeerConnectionHandlerAnnouncedPeer( this, pPeerDef, zPeerList ).start();
        }
    }

    public void addFromPeer( PeerDefinition[] pPeerDefinitions )
    {
        if ( pPeerDefinitions != null )
        {
            for ( PeerDefinition zPeerDefinition : pPeerDefinitions )
            {
                addAnnouncedPeer( zPeerDefinition );
            }
        }
    }

    /**
     * @return null if we are already talking to this Peer OR we refuse to talk to it (i.e. it is null)?
     */
    public synchronized BlockingQueue<Serializable> startPeeringWith( PeerConnectionHandler pPeerConnectionHandler )
    {
        if ( pPeerConnectionHandler != null )
        {
            PeerDefinition zPeerDef = pPeerConnectionHandler.getPeerDef();
            if ( zPeerDef != null )
            {
                if ( !mActivePeers.contains( zPeerDef ) )
                {
                    mPendingPeers.remove( zPeerDef );
                    mActivePeers.add( zPeerDef, pPeerConnectionHandler );
                    if ( mReceivedThread == null )
                    {
                        mReceivedThread = new ReceivedThread();
                        mReceivedThread.start();
                    }
                    resetWaitForPeersTimer();
                    return mReceivedThread.getReceiveQueue();
                }
                LOGGER.error.log( "Attempt to duplicate Peer w/ " + zPeerDef );
            }
        }
        return null;
    }

    public synchronized void noLongerPeeringWith( PeerDefinition pPeerDef )
    {
        PeerConnectionHandler zPeerConnectionHandler = mActivePeers.remove( pPeerDef );
        if ( zPeerConnectionHandler != null )
        {
            zPeerConnectionHandler.close();
        }
    }

    public synchronized void sendToPeers( Serializable pMessageToPeers )
    {
        mActivePeers.sendToPeers( pMessageToPeers );
    }

    public synchronized void aboutToJoinGroup()
    {
        resetWaitForPeersTimer();
    }

    public synchronized void waitForPeers()
    {
        for ( long toWait; 0 < (toWait = getWaitForPeersTime()); )
        {
            try
            {
                wait( toWait );
            }
            catch ( InterruptedException e )
            {
                // Whatever!
            }
        }
    }

    private long getWaitForPeersTime()
    {
        return (mWaitTillForPeers == null) ? 0 : (mWaitTillForPeers - System.currentTimeMillis());
    }

    private Long mWaitTillForPeers = null;

    private void resetWaitForPeersTimer()
    {
        mWaitTillForPeers = System.currentTimeMillis() + MILLISEC_TIME_OUT_WAITING_FOR_PEER;
    }

    private static class PeerDefs
    {
        private Set<PeerDefinition> mPeerDefinitions = new HashSet<PeerDefinition>();

        public boolean hasAny()
        {
            return !mPeerDefinitions.isEmpty();
        }

        public boolean contains( PeerDefinition pPeerDef )
        {
            return mPeerDefinitions.contains( pPeerDef );
        }

        public void add( PeerDefinition pPeerDef )
        {
            mPeerDefinitions.add( pPeerDef );
        }

        public void remove( PeerDefinition pPeerDef )
        {
            mPeerDefinitions.remove( pPeerDef );
        }
    }

    private static class ActivePeers
    {
        private Map<PeerDefinition, PeerConnectionHandler> mPeers = new HashMap<PeerDefinition, PeerConnectionHandler>();
        private PeerDefinition[] mCachedPeerDefPeers = null;

        public PeerDefinition[] getPeerDefs()
        {
            if ( mCachedPeerDefPeers == null )
            {
                mCachedPeerDefPeers = mPeers.isEmpty() ? //
                                      PeerDefinition.EMPTY_ARRAY : //
                                      mPeers.keySet().toArray( new PeerDefinition[mPeers.keySet().size()] );
            }
            return mCachedPeerDefPeers;
        }

        public boolean contains( PeerDefinition pPeerDef )
        {
            return mPeers.containsKey( pPeerDef );
        }

        public PeerConnectionHandler remove( PeerDefinition pPeerDef )
        {
            PeerConnectionHandler zHandler = mPeers.remove( pPeerDef );
            if ( zHandler != null )
            {
                mCachedPeerDefPeers = null;
                System.out.println( "--- Removed Peer: " + pPeerDef );
            }
            return zHandler;
        }

        public void add( PeerDefinition pPeerDef, PeerConnectionHandler pPeerConnectionHandler )
        {
            mCachedPeerDefPeers = null;
            mPeers.put( pPeerDef, pPeerConnectionHandler );
            System.out.println( "+++++ Added Peer: " + pPeerDef );
        }

        public void sendToPeers( Serializable pMessageToPeers )
        {
            for ( PeerConnectionHandler zHandler : mPeers.values() )
            {
                zHandler.send( pMessageToPeers );
            }
        }
    }

    private class ReceivedThread extends Thread
    {
        private BlockingQueue<Serializable> mReceiveQueue = new LinkedBlockingQueue<Serializable>();

        public ReceivedThread()
        {
            setName( getClass().getName() );
            setDaemon( true );
        }

        public BlockingQueue<Serializable> getReceiveQueue()
        {
            return mReceiveQueue;
        }

        private Serializable getFromQueue()
        {
            while ( true )
            {
                try
                {
                    return mReceiveQueue.take();
                }
                catch ( InterruptedException e )
                {
                    // Whatever...
                }
            }
        }

        @SuppressWarnings({"InfiniteLoopStatement"}) @Override
        public void run()
        {
            while ( true )
            {
                processFromQueue( getFromQueue() );
            }
        }

        private void processFromQueue( Serializable pSerializable )
        {
            MessageFromPeerHandler[] zHandlers = mMessageFromPeerHandlerManager.getHandlers();
            for ( MessageFromPeerHandler zHandler : zHandlers )
            {
                try
                {
                    if ( zHandler.handle( pSerializable, mPeerManager ) )
                    {
                        return;
                    }
                }
                catch ( Exception e )
                {
                    LOGGER.error.log( e );
                }
            }
            LOGGER.info.log( "No MessageFromPeerHandler interested in: ", pSerializable );
        }
    }
}