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

import java.io.*;
import java.net.*;

import org.litesoft.commonfoundation.typeutils.*;
import org.litesoft.logger.*;
import org.litesoft.peertopeer.*;
import org.litesoft.peertopeer.nonpublic.peermanagement.*;
import org.litesoft.util.*;

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

    private static final byte BCA_BYTE = (byte) 255;
    private static final byte[] BROADCAST_ADDRESS = new byte[]{BCA_BYTE, BCA_BYTE, BCA_BYTE, BCA_BYTE};

    private String mOurName, mAppGroupID, mAppGroupVersion, mMessagePrefixFilter;
    private AppGroupVersionCompatibilityChecker mCompatibilityChecker;

    public BroadcastMessageManager( String pOurName, String pAppGroupID, String pAppGroupVersion, AppGroupVersionCompatibilityChecker pCompatibilityChecker )
    {
        mOurName = assertNotEmptyAndNoCommas( "OurName", pOurName );
        mAppGroupID = assertNotEmptyAndNoCommas( "AppGroupID", pAppGroupID );
        mAppGroupVersion = assertNotEmptyAndNoCommas( "AppGroupVersion", pAppGroupVersion );
        mCompatibilityChecker = pCompatibilityChecker;
        mMessagePrefixFilter = mAppGroupID + ",";
    }

    public String getOurName()
    {
        return mOurName;
    }

    public synchronized void assert_joinGroup_notCalled()
    {
        if ( mOurServer != null )
        {
            throw new IllegalStateException( "BroadcastMessageManager has already joined (or attempted to join) Group: " + mAppGroupID );
        }
    }

    public synchronized void joinGroup( int pServerPort, PeerConnectionHandlerFactory pConnectionHandlerFactory )
            throws IOException
    {
        P2Putils.validatePort( pServerPort );
        Objects.assertNotNull( "PeerConnectionHandlerFactory", pConnectionHandlerFactory );
        assert_joinGroup_notCalled();

        mOurServer = new OurServer( getClass().getName(), pConnectionHandlerFactory );

        // get a datagram socket
        DatagramSocket socket = new DatagramSocket();

        try
        {
            socket.setBroadcast( true );

            mOurServer.startListening();

            byte[] buf = createAnnouceMessage( pServerPort );
            InetAddress address = InetAddress.getByAddress( BROADCAST_ADDRESS );
            DatagramPacket packet = new DatagramPacket( buf, buf.length, address, DISCOVERY_PORT );
            socket.send( packet );
        }
        catch ( NoRouteToHostException e )
        {
            System.err.println( "*** Unable to Broadcast Our Existence - Proceeding Stand Alone ***" );
            mOurServer.stopListening();
        }
        catch ( IOException e )
        {
            mOurServer.stopListening();
            throw e;
        }
        finally
        {
            socket.close();
        }
    }

    private void processAnnounceMessage( String pMessage, InetAddress pInetAddress, PeerConnectionHandlerFactory pConnectionHandlerFactory )
    {
        if ( !pMessage.startsWith( mMessagePrefixFilter ) )
        {
            return;
        }
        int cAt1 = pMessage.indexOf( ',' );
        int cAt2 = pMessage.indexOf( ',', cAt1 + 1 );
        int cAt3 = pMessage.indexOf( ',', cAt2 + 1 );
        if ( (cAt2 != -1) && (cAt3 != -1) )
        {
            String thierAppGroupVersion = pMessage.substring( cAt1 + 1, cAt2 ).trim();
            String thierName = pMessage.substring( cAt2 + 1, cAt3 ).trim();
            String thierServerPort = pMessage.substring( cAt3 + 1 ).trim();
            if ( (thierAppGroupVersion.length() != 0) && (thierName.length() != 0) && (thierServerPort.length() != 0) )
            {
                int zPort = parseServerPort( thierServerPort );
                if ( zPort != 0 )
                {
                    PeerDefinition zNewPeerDef = new PeerDefinition( thierName, pInetAddress, zPort );
                    if ( !mAppGroupVersion.equals( thierAppGroupVersion ) && ((mCompatibilityChecker == null) || !mCompatibilityChecker.areCompatibile( mAppGroupVersion, thierAppGroupVersion )) )
                    {
                        pConnectionHandlerFactory.handleAnnouncedPeerWithIncompatibleAppGroupVersions( zNewPeerDef, mOurName, mAppGroupVersion, thierAppGroupVersion );
                    }
                    else
                    {
                        pConnectionHandlerFactory.handleAnnouncedPeer( zNewPeerDef );
                    }
                    return;
                }
            }
        }
        LOGGER.error.log( "MalformedBroadcastAnnounceMessage from ", pInetAddress, ":", pMessage );
    }

    private byte[] createAnnouceMessage( int pServerPort )
            throws UnsupportedEncodingException
    {
        return createAnnouceMessage( mAppGroupID + "," + mAppGroupVersion + "," + mOurName + "," + pServerPort );
    }

    private static byte[] createAnnouceMessage( String pMessage )
    {
        byte[] zSBytes;
        try
        {
            zSBytes = pMessage.getBytes( FileUtils.UTF_8 );
        }
        catch ( UnsupportedEncodingException e )
        {
            LOGGER.error.log( e ); // shouldn't happen
            zSBytes = new byte[0];
        }
        int rvLength = zSBytes.length + BROADCAST_PREFIX.length;
        if ( rvLength > 255 )
        {
            throw new IllegalArgumentException( "Announce Message too long: " + pMessage );
        }
        byte[] rv = new byte[rvLength];
        int to = 0;
        for ( byte b : BROADCAST_PREFIX )
        {
            rv[to++] = b;
        }
        for ( byte b : zSBytes )
        {
            rv[to++] = b;
        }
        return rv;
    }

    private static String parseAnnouceMessage( DatagramPacket packet )
    {
        int zDataLength = packet.getLength();
        if ( zDataLength < MINIMUM_BROADCAST_MESSAGE_LENGTH )
        {
            return null;
        }
        byte[] zData = packet.getData();
        int zEncodedStringStartsAt = BROADCAST_PREFIX.length;
        for ( int i = 0; i < zEncodedStringStartsAt; i++ )
        {
            if ( zData[i] != BROADCAST_PREFIX[i] )
            {
                return null;
            }
        }
        String zMessage = null;
        try
        {
            zMessage = new String( zData, zEncodedStringStartsAt, zDataLength - zEncodedStringStartsAt, FileUtils.UTF_8 );
        }
        catch ( UnsupportedEncodingException e )
        {
            LOGGER.error.log( e ); // shouldn't happen
        }
        catch ( RuntimeException e )
        {
            // Not Properly Encoded?
            LOGGER.error.log( e );
        }
        return zMessage;
    }

    private OurServer mOurServer = null;

    private class OurServer extends Thread
    {
        private PeerConnectionHandlerFactory mConnectionHandlerFactory;
        private DatagramSocket mSocket = null;

        public OurServer( String pName, PeerConnectionHandlerFactory pConnectionHandlerFactory )
        {
            super( pName );
            mConnectionHandlerFactory = pConnectionHandlerFactory;
            setDaemon( true );
        }

        public synchronized void startListening()
                throws SocketException
        {
            mSocket = new DatagramSocket( null );
            mSocket.setReuseAddress( true );
            mSocket.bind( new InetSocketAddress( (InetAddress) null, DISCOVERY_PORT ) );
            start();
        }

        public synchronized boolean stopListening()
        {
            DatagramSocket zSocket = mSocket;
            mSocket = null;
            if ( zSocket == null )
            {
                return false;
            }
            zSocket.close();
            return true;
        }

        private synchronized DatagramSocket stillListening()
        {
            return mSocket;
        }

        @Override
        public void run()
        {
            for ( DatagramSocket zSocket; null != (zSocket = stillListening()); )
            {
                DatagramPacket packet = new DatagramPacket( new byte[256], 256 );
                try
                {
                    // receive request
                    zSocket.receive( packet );
                }
                catch ( IOException e )
                {
                    if ( stopListening() )
                    {
                        LOGGER.error.log( e );
                    }
                    return;
                }
                try
                {
                    String zMessage = parseAnnouceMessage( packet );
                    if ( zMessage != null )
                    {
                        processAnnounceMessage( zMessage, packet.getAddress(), mConnectionHandlerFactory );
                    }
                }
                catch ( RuntimeException e )
                {
                    LOGGER.error.log( e );
                }
            }
        }
    }

    private static String assertNotEmptyAndNoCommas( String pErrorMessage, String pStringToAssert )
    {
        pStringToAssert = Strings.assertNotNullNotEmpty( pErrorMessage, pStringToAssert );
        if ( pStringToAssert.contains( "," ) )
        {
            Strings.error( "String", pErrorMessage, " is not allowed to contain any Commas (',')" );
        }

        return pStringToAssert;
    }

    private static int parseServerPort( String pThierServerPort )
    {
        try
        {
            return P2Putils.validatePort( Integer.parseInt( pThierServerPort ) );
        }
        catch ( RuntimeException e )
        {
            return 0;
        }
    }
}