package org.maachang.comet.proxy ;

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.nio.channels.CancelledKeyException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.maachang.comet.httpd.HttpConnectionInfo;
import org.maachang.comet.httpd.HttpSocket;
import org.maachang.comet.httpd.HttpdRequest;
import org.maachang.comet.net.HttpCallbackReceive;
import org.maachang.comet.net.HttpInputStream;
import org.maachang.comet.net.HttpSocketManager;
import org.maachang.comet.net.HttpUseQueue;
import org.maachang.comet.net.ReadHttpdRequest;
import org.maachang.util.ArrayBinary;
import org.maachang.util.ConvertParam;
import org.maachang.util.GZIPBinary;
import org.maachang.util.SequenceSync;

/**
 * 処理用プーリングスレッド.
 * 
 * @version 2007/11/17
 * @author  masahito suzuki
 * @since   MaachangComet 1.00
 */
class ProxyPoolThread extends Thread {
    
    private static final Log LOG = LogFactory.getLog( ProxyPoolThread.class ) ;
    
    private HttpUseQueue queue = null ;
    private HttpSocketManager manager = null ;
    private HttpCallbackReceive receive = null ;
    private SequenceSync sequence = null ;
    private volatile boolean stopFlag = true ;
    
    private ProxyPoolThread() {
        
    }
    
    public ProxyPoolThread( HttpUseQueue queue,HttpSocketManager manager,HttpCallbackReceive receive,
        SequenceSync sequence )
        throws Exception {
        this.queue = queue ;
        this.manager = manager ;
        this.receive = receive ;
        this.sequence = sequence ;
        this.stopFlag = false ;
        this.setDaemon( true ) ;
        this.start() ;
    }
    
    public void destroy() {
        setStop( true ) ;
    }
    
    public synchronized boolean isStop() {
        return stopFlag ;
    }
    
    private synchronized void setStop( boolean mode ) {
        this.stopFlag = mode ;
    }
    
    private static final long WAIT_TIME = 30L ;
    
    public void run() {
        ThreadDeath threadDeach = null ;
        boolean errorFlag = false ;
        boolean endFlag = false ;
        for( ;; ) {
            errorFlag = false ;
            HttpSocket conn = null ;
            HttpConnectionInfo info = null ;
            try {
                if( endFlag == true || isStop() == true ) {
                    break ;
                }
                conn = this.queue.getQueue() ;
                if( conn != null ) {
                    info = conn.getConnectionInfo() ;
                    if( info == null || info.isUse() == false ) {
                        continue ;
                    }
                    int seqId = sequence.getId() ;
                    if( LOG.isInfoEnabled() ) {
                        LOG.info( new StringBuilder().append( "[px]receive(" ).append( seqId ).append( ") - [" ).
                            append( info.getSocket().socket().getInetAddress().getHostAddress() ).
                            append( ":" ).append( info.getSocket().socket().getPort() ).
                            append( "]" ).toString() ) ;
                    }
                    this.receive.execution( info,seqId,readHttpdRequest( conn,seqId ) ) ;
                    manager.append( conn ) ;
                }
                else {
                    sleep( WAIT_TIME ) ;
                }
            } catch( InterruptedException ie ) {
                errorFlag = true ;
                endFlag = true ;
            } catch( CancelledKeyException ck ) {
                errorFlag = true ;
                LOG.warn( "## cancell - key" ) ;
            } catch( OutOfMemoryError mem ) {
                errorFlag = true ;
                LOG.error( "## out-of-memory-error",mem ) ;
            } catch( Exception e ) {
                errorFlag = true ;
                LOG.error( "## error",e ) ;
            } catch( ThreadDeath td ) {
                errorFlag = true ;
                endFlag = true ;
                threadDeach = td ;
            } finally {
                if( errorFlag == true && conn != null ) {
                    if( conn.isClosed() == false ) {
                        // 4バイト文字を送る.
                        try {
                            conn.outputStream().write( ConvertParam.convertInt( 4 ) ) ;
                            conn.outputStream().flush() ;
                            manager.append( conn ) ;
                        } catch( Exception e ) {
                            e.printStackTrace() ;
                        }
                    }
                }
            }
        }
        this.queue = null ;
        this.manager = null ;
        this.receive = null ;
        this.sequence = null ;
        //LOG.info( "## Thread:" + this.getName() + " 終了" ) ;
        setStop( true ) ;
        if( threadDeach != null ) {
            throw threadDeach ;
        }
    }
    
    private static final HttpdRequest readHttpdRequest( HttpSocket socket,int seqId )
        throws Exception {
        if( socket == null || socket.isClosed() == true ) {
            return null ;
        }
        BufferedInputStream buf = null ;
        try {
            buf = new BufferedInputStream( new HttpInputStream( socket.inputStream() ) ) ;
            HttpdRequest ret = receive( buf,seqId ) ;
            buf.close() ;
            buf = null ;
            socket.update() ;
            return ret ;
        } finally {
            if( buf != null ) {
                try {
                    buf.close() ;
                } catch( Exception e ) {
                }
                buf = null ;
            }
        }
    }
    
    /**
     * HTTPリクエスト受信.
     * @return ResponseInfo レスポンス情報が返されます.
     * @exception Exception 例外.
     */
    private static final HttpdRequest receive( InputStream inputStream,int seqId) throws Exception {
        int allLen = -1 ;
        ArrayBinary ret = new ArrayBinary() ;
        try {
            //if( inputStream.available() <= 0 ) {
            //    return null ;
            //}
            for( ;; ) {
                if( allLen <= -1 ) {
                    if( ret.length() >= 4 ) {
                        allLen = ( ret.getByte( 0 ) & 0x000000ff ) |
                            ( ( ret.getByte( 1 ) & 0x000000ff ) << 8 ) |
                            ( ( ret.getByte( 2 ) & 0x000000ff ) << 16 ) |
                            ( ( ret.getByte( 3 ) & 0x000000ff ) << 24 ) ;
                        if( allLen <= 4 ) {
                            return null ;
                        }
                    }
                }
                else if( allLen <= ret.length() ) {
                    Object[] params = new Object[ 1 ] ;
                    params[ 0 ] = ret.getBinary() ;
                    ret.destroy() ;
                    return convertHttpdRequest( params ) ;
                }
                int n = inputStream.read() ;
                ret.write( n ) ;
            }
        } catch( SocketTimeoutException se ) {
            return null ;
        } catch( Exception e ) {
            throw e ;
        }
    }
    
    /** 受信内容をHttpdRequestに変換. */
    private static final HttpdRequest convertHttpdRequest( Object[] o )
        throws Exception {
        HttpdRequest ret = new HttpdProxyRequestImpl() ;
        byte[] bin = ( byte[] )o[ 0 ] ;
        o[ 0 ] = null ;
        // ヘッダ全体長を取得.
        int allHeaderLen = ConvertParam.convertInt( 4,bin ) ;
        // Body全体長を取得.
        int allBodyLen = ConvertParam.convertInt( 8+allHeaderLen,bin ) ;
        // Body全体長が存在する場合.
        if( allBodyLen > 0 ) {
            // Body内容を取得.
            byte[] body = new byte[ allBodyLen ] ;
            System.arraycopy( bin,12+allHeaderLen,body,0,allBodyLen ) ;
            ret.setBody( GZIPBinary.getInstance().convertGZIPByBinary( body ) ) ;
        }
        // ヘッダ内容を解凍して取得.
        byte[] header = GZIPBinary.getInstance().convertGZIPByBinary( bin,8,allHeaderLen ) ;
        bin = null ;
        
        // メソッド名
        int pnt = 0 ;
        int len = ConvertParam.convertInt( pnt,header ) ;
        pnt += 4 ;
        ret.setMethod( convertString( header,pnt,len ) ) ;
        pnt += len ;
        
        // URLパス.
        len = ConvertParam.convertInt( pnt,header ) ;
        pnt += 4 ;
        ret.setUrlPath( convertString( header,pnt,len ) ) ;
        pnt += len ;
        
        // Httpバージョンを取得.
        len = ConvertParam.convertInt( pnt,header ) ;
        pnt += 4 ;
        ret.setHttpVersion( convertString( header,pnt,len ) ) ;
        pnt += len ;
        
        // ヘッダ数を取得.
        int headerLen = ConvertParam.convertInt( pnt,header ) ;
        pnt += 4 ;
        
        // ヘッダ群を取得.
        for( int i = 0 ; i < headerLen ; i ++ ) {
            // キー名を取得.
            len = ConvertParam.convertInt( pnt,header ) ;
            pnt += 4 ;
            String key = convertString( header,pnt,len ) ;
            pnt += len ;
            
            // 要素長を取得.
            int lenJ = ConvertParam.convertInt( pnt,header ) ;
            pnt += 4 ;
            for( int j = 0 ; j < lenJ ; j ++ ) {
                // １つの要素を取得.
                len = ConvertParam.convertInt( pnt,header ) ;
                pnt += 4 ;
                ret.getHeader().addHeader( key,convertString( header,pnt,len ) ) ;
                pnt += len ;
            }
        }
        header = null ;
        // URLパラメータを変換.
        ReadHttpdRequest.convertAnalysisQuery( ret ) ;
        // Bodyパラメータを変換.
        byte[] body = ret.getBodyByBinary() ;
        if( body != null ) {
            ReadHttpdRequest.convertBody( ret,body ) ;
        }
        return ret ;
    }
    
    /** １つの内容を文字列に変換. */
    private static final String convertString( byte[] in,int pnt,int length )
        throws Exception {
        byte[] bin = new byte[ length ] ;
        System.arraycopy( in,pnt,bin,0,length ) ;
        return new String( bin,"ISO-8859-1" ) ;
    }
}
