/*************************************************************************
 * PCRaw
 * Author: nori090
 * Original author: ◆e5bW6vDOJ.
 * Copyright: Copyright (C) 2008 nori090
 * Copyright: Copyright (C) 2007 ◆e5bW6vDOJ.
 * Date: February 11, 2007
 * License: 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 02111-1307, USA.
 * 
 * http://mosax.sakura.ne.jp/fswiki.cgi?page=PCRaw
 */
package org.coderepos.nori090.pcraw;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.coderepos.nori090.pcraw.PCRP.PCRPInfo;

/**
 * @author nori090
 * @version $Rev: 138 $ $Date: 2008-08-01 06:28:28 +0900 (Fri, 01 Aug 2008) $
 */
public class PCRaw {
    private static final Log log = LogFactory.getLog( PCRaw.class );

    static final int VERSION = 0x00030106;

    static final String USER_AGENT = "PCRaw/0.3.1.6";

    static final int PEERCAST_DEFAULT_PORT = 7144;

    static final int MARKER = 0xFFAA0055;

    static final int MAX_DATA_SIZE = 1000000;

    static final int NO_COUNT = 123456789;

    static final int TIMEOUT = 60;

    static final int RECONNECT_WAIT = 30;

    static final int MAX_RECONNECT = 3;

    static final int MAX_SYNC_ERROR = 3;

    static final int MAX_SYNC_ERROR_LIMIT = 3;

    static final float MIN_SKIP_AMP = 0.75F;

    static final int RECONNECT_RESET_COUNT = 3;

    static class Global {
        boolean usable = true;

        boolean versionUpNotifyShowed;

        int syncErrorCount;

        int typeErrorCount;

        Object syncObj = new Object();

    }

    class RecvThread
        implements Runnable {
        boolean terminate = false;

        int syncErrorCount;

        int succeedCount;

        int reconnectCount;

        long typeErrorCount;

        long lastSendtime;

        long lastRecvtime;

        void processPCRP( SocketChannel sock )
            throws IOException, AllowException {
            PCRP pcrp = new PCRP( info, sock );
            long sendtime = 0;
            char[] id = pcrp.readId();
            if ( !Arrays.equals( id, PCRP.PCRP_START ) ) {
                throw new IOException( "RecvThread.processPCRP() Invalid data start." );
            }
            for ( int i = 0; i < PCRP.MAX_DATA_NUM; i++ ) {
                id = pcrp.readId();
                if ( Arrays.equals( id, PCRP.PCRP_END ) ) {
                    break;
                }
                else if ( Arrays.equals( id, PCRP.PCRP_LIFETIME ) ) {
                    byte[] tmp = pcrp.readData();
                    if ( tmp.length == 4 ) {
                        long lifetime = 0;
                        lifetime = ( ( tmp[3] & 0xFF ) << 24 );
                        lifetime = lifetime | ( ( tmp[2] & 0xFF ) << 16 );
                        lifetime = lifetime | ( ( tmp[1] & 0xFF ) << 8 );
                        lifetime = lifetime | ( tmp[0] & 0xFF );
                        info.keyMgr.lifeTime = lifetime;
                    }
                }
                else if ( Arrays.equals( id, PCRP.PCRP_SENDTIME ) ) {
                    byte[] tmp = pcrp.readData();
                    if ( tmp.length == 4 ) {
                        sendtime = ( ( tmp[3] & 0xFF ) << 24 );
                        sendtime = sendtime | ( ( tmp[2] & 0xFF ) << 16 );
                        sendtime = sendtime | ( ( tmp[1] & 0xFF ) << 8 );
                        sendtime = sendtime | ( tmp[0] & 0xFF );
                        long t = PCRawTools.time();
                        boolean skip = ( sendtime - lastSendtime ) > ( t - lastRecvtime ) / MIN_SKIP_AMP;
                        if ( lastRecvtime == 0 ) {
                            skip = true;
                        }
                        log.debug( String.format( "Time: %d/%d Diff: %d/%d\n", sendtime, t, sendtime - lastSendtime, t -
                            lastRecvtime ) );
                        lastSendtime = sendtime;
                        lastRecvtime = t;
                        if ( skip ) {
                            log.debug( "データ処理スキップ..." );
                            try {
                                for ( ; i < PCRP.MAX_DATA_NUM; i++ ) {
                                    id = pcrp.readId();
                                    if ( Arrays.equals( id, PCRP.PCRP_END ) ) {
                                        break;
                                    }
                                    else {
                                        pcrp.skipData();
                                    }
                                }
                            }
                            catch ( IOException e ) {
                                throw new AllowException( e );
                            }
                        }
                    }
                }
                else if ( Arrays.equals( id, PCRP.PCRP_KEYN ) ) {
                    byte[] tmp = pcrp.readData();
                    if ( tmp.length == 16 ) {
                        info.keyMgr.add( tmp, 0L );
                    }
                }
                else if ( Arrays.equals( id, PCRP.PCRP_DATA ) ) {
                    byte[] tmp = pcrp.readData();
                    synchronized ( g.syncObj ) {
                        g.syncErrorCount = 0;
                    }
                    syncErrorCount = 0;
                    succeedCount++;
                    if ( succeedCount >= RECONNECT_RESET_COUNT ) {
                        log.debug( "データ格納." );
                        synchronized ( syncObj ) {
                            data = tmp;
                            if ( sendtime != 0 ) {
                                updateTime = sendtime;
                            }
                            else {
                                updateTime = PCRawTools.time();
                            }
                        }
                    }
                    error = 0;
                }
                else {
                    log.debug( new StringBuilder( "データスキップ. " ).append( id ).toString() );
                    pcrp.skipData();
                }
                if ( Arrays.equals( id, PCRP.PCRP_END ) ) {
                    break;
                }
            }
            if ( !Arrays.equals( id, PCRP.PCRP_END ) ) {
                throw new IOException( "RecvThread.processPCRP() Invalid data end." );
            }
        }

        void processStream( SocketChannel sock ) {
            syncErrorCount = 0;
            succeedCount = 0;
            while ( !terminate ) {
                try {
                    if ( syncErrorCount >= MAX_SYNC_ERROR ) {
                        synchronized ( g.syncObj ) {
                            g.syncErrorCount++;
                            if ( typeErrorCount >= MAX_SYNC_ERROR ) {
                                g.typeErrorCount++;
                            }
                        }
                        throw new RuntimeException( "同期エラー限界." );
                    }

                    {
                        int marker = 0;
                        int count = 0;
                        log.debug( "同期マーカを探索中..." );
                        ByteBuffer buff = ByteBuffer.allocate( 1 );
                        do {
                            if ( sock.read( buff ) == -1 ) {
                                throw new IOException( "end of stream!" );
                            }
                            buff.rewind();
                            marker <<= 8;
                            marker |= buff.get();
                            buff.clear();
                            count++;
                        }
                        while ( !terminate && marker != MARKER && count < MAX_DATA_SIZE * 2 );
                        if ( marker != MARKER || count >= MAX_DATA_SIZE * 2 ) {
                            throw new IOException( "同期マーカが見つかりません." );
                        }
                        log.debug( "同期マーカを発見." );
                    }
                    ByteBuffer buff = ByteBuffer.allocate( 1 );
                    if ( sock.read( buff ) == -1 ) {
                        throw new IOException( "end of stream!" );
                    }
                    buff.rewind();
                    if ( buff.get() != PCRP.STREAM_VERSION ) {
                        typeErrorCount++;
                        throw new IOException( "ストリームバージョンが不正." );
                    }
                    processPCRP( sock );
                    if ( succeedCount >= RECONNECT_RESET_COUNT ) {
                        reconnectCount = 0;
                    }
                }
                catch ( IOException e ) {
                    log.debug( e.getMessage() );
                    syncErrorCount++;
                    error = 1;
                }
                catch ( AllowException e ) {
                }
            }
        }

        void processConnect() {
            while ( !terminate ) {
                try {
                    typeErrorCount = 0;
                    lastSendtime = 0;
                    lastRecvtime = 0;
                    if ( reconnectCount >= MAX_RECONNECT ) {
                        throw new RuntimeException( "再接続限界." );
                    }
                    reconnectCount++;
                    if ( error != 0 ) {
                        log.debug( "待機中..." );
                        try {
                            for ( int i = 0; i < RECONNECT_WAIT * 10 && !terminate; i++ ) {
                                Thread.sleep( 100 );
                            }
                        }
                        catch ( InterruptedException e ) {
                            e.printStackTrace();
                        }
                        if ( terminate ) {
                            throw new RuntimeException( "待機中断." );
                        }
                    }
                    log.debug( "接続中..." );
                    SocketChannel sock = SocketChannel.open( new InetSocketAddress( info.localHost, info.localPort ) );
                    sock.configureBlocking( true );
                    sock.socket().setSoTimeout( TIMEOUT * 1000 );
                    synchronized ( syncObj ) {
                        PCRaw.this.sock = sock;
                    }
                    log.debug( "接続成功." );
                    log.debug( "HTTPヘッダ送信中..." );
                    ByteArrayOutputStream buf = new ByteArrayOutputStream();
                    buf.write( getByte( String.format( "GET %s HTTP/1.0\r\n", object ) ) );
                    buf.write( getByte( String.format( "Host: %s\r\n", info.localHost ) ) );
                    buf.write( getByte( String.format( "User-Agent: %s\r\n", USER_AGENT ) ) );
                    buf.write( getByte( "\r\n" ) );
                    sock.write( ByteBuffer.wrap( buf.toByteArray() ) );
                    String line = readLine( sock );
                    Pattern p = Pattern.compile( "HTTP/[0-9.]+ ([0-9]+) .*" );
                    Matcher m = p.matcher( line );
                    if ( m.find() ) {
                        int status = Integer.parseInt( m.group( 1 ) );
                        log.debug( "HTTPレスポンスヘッダスキップ..." );
                        while ( readLine( sock ).length() > 0 && !terminate ) {
                        }
                        if ( status != 200 ) {
                            throw new IOException( String.format( "HTTPレスポンスヘッダエラー. code:%d", status ) );
                        }
                    }
                    else {
                        throw new IOException( "HTTPレスポンスヘッダエラー." );
                    }
                    log.debug( "HTTPレスポンスヘッダ受信完了." );
                    processStream( sock );
                }
                catch ( IOException e ) {
                    error = 2;
                }
                catch ( RuntimeException e ) {
                    e.printStackTrace();
                    break;
                }
            }
        }

        static final String HTTP_ELEMENT_CHARSET = "US-ASCII";

        byte[] getByte( String data ) {
            try {
                return data.getBytes( HTTP_ELEMENT_CHARSET );
            }
            catch ( UnsupportedEncodingException e ) {
                return data.getBytes();
            }
        }

        String readLine( SocketChannel sock )
            throws IOException {
            ByteBuffer bs = ByteBuffer.allocate( 1 );
            ByteArrayOutputStream buf = new ByteArrayOutputStream();
            byte b = 0;
            while ( sock.read( bs ) >= 0 ) {
                bs.rewind();
                b = bs.get();
                if ( '\n' == b || '\r' == b ) {
                    break;
                }
                buf.write( b );
                bs.clear();
            }
            return new String( buf.toByteArray(), HTTP_ELEMENT_CHARSET );
        }

        @Override
        public void run() {
            try {
                processConnect();
                error = -1;
                log.debug( "受信スレッドが正常終了しました." );
            }
            catch ( Exception e ) {
                error = -2;
                log.debug( "受信スレッドが異常終了しました." );
            }
        }
    }

    /* pcraw fields */
    static Global g = new Global();

    PCRPInfo info = new PCRPInfo();

    Object syncObj = new Object();

    long updateTime;

    byte[] data;

    int error;

    SocketChannel sock;

    String object;

    RecvThread thread;

    Thread t;

    boolean closed = false;

    public PCRaw( String url )
        throws Exception {
        log.debug( String.format( "PCRaw.this(%s)\n", url ) );
        if ( !g.usable ) {
            throw new Exception( "PCRaw 使用不可." );
        }
        if ( g.typeErrorCount >= MAX_SYNC_ERROR_LIMIT ) {
            throw new Exception( "PCRaw ストリームタイプエラー多発." );
        }
        if ( g.syncErrorCount >= MAX_SYNC_ERROR_LIMIT ) {
            throw new Exception( "PCRaw 同期エラー多発." );
        }
        Pattern p = Pattern.compile( "http://([^:/]+)(:([0-9]+))?(/[^/]+/([0-9a-fA-F]+)\\?tip=([^:&]+)(:([0-9]+))?)" );
        Matcher m = p.matcher( url );
        if ( m.find() ) {
            info.localHost = m.group( 1 );
            info.localPort = m.group( 3 ) == null ? 80 : Integer.parseInt( m.group( 3 ) );
            info.trackerHost = m.group( 6 );
            info.trackerPort = m.group( 8 ) == null ? 80 : Integer.parseInt( m.group( 8 ) );
            object = m.group( 4 );
        }
        else {
            throw new Exception( "PCRaw URL不正." );
        }
        info.keyMgr = new KeyManager();
        thread = new RecvThread();
        t = new Thread( thread );
        t.start();
        log.debug( String.format( "PCRaw.this(%s)\n", url ) );
    }

    void close()
        throws IOException {
        log.debug( "PCRaw.close()" );
        if ( !closed ) {
            thread.terminate = true;
            synchronized ( syncObj ) {
                if ( sock != null ) {
                    sock.close();
                }
                try {
                    t.join();
                }
                catch ( InterruptedException e ) {
                    e.printStackTrace();
                }
                closed = true;
            }
        }
    }

    int getLength() {
        synchronized ( syncObj ) {
            log.debug( "PCRaw.getLength()" );
            return data.length;
        }
    }

    byte[] getData() {
        synchronized ( syncObj ) {
            log.debug( "PCRaw.getData()" );
            if ( data != null ) {
                byte[] dest = new byte[data.length];
                System.arraycopy( data, 0, dest, 0, data.length );
                return dest;
            }
            return null;
        }
    }

    int getLastError() {
        synchronized ( syncObj ) {
            log.debug( "PCRaw.getLastError()" );
            return error;
        }
    }

    long getUpdateTime() {
        synchronized ( syncObj ) {
            log.debug( "PCRaw.getUpdateTime()" );
            return updateTime;
        }

    }

    static int pcrawCheckVersion( int assumedVersion ) {
        if ( assumedVersion != 0 ) {
            if ( ( VERSION & 0xffff0000 ) > ( assumedVersion & 0xffff0000 ) ) {
                log.debug( "想定するバージョンより新しい." );
                g.usable = false;
                return 1;
            }

            if ( VERSION < assumedVersion ) {
                log.debug( "想定するバージョンより古い." );
                g.usable = false;
                return -1;
            }

            log.debug( "想定するバージョンと適合." );

            g.usable = true;
            return 0;
        }
        return 0;
    }
}
