package org.maachang.comet.net.nio ;

import java.io.IOException;
import java.io.InputStream;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.maachang.comet.net.HttpdRequestImpl;
import org.maachang.comet.net.ReadHttpdRequest;
import org.maachang.comet.net.RequestBuffer;
import org.maachang.util.SequenceSync;
import org.maachang.util.thread.LoopThread;

/**
 * 処理用プーリングスレッド.
 * 
 * @version 2008/05/28
 * @author  masahito suzuki
 * @since   MaachangComet 1.1B
 */
class NioPoolThread extends LoopThread {
    
    /**
     * ログ.
     */
    private static final Log LOG = LogFactory.getLog( NioPoolThread.class ) ;
    
    /**
     * 受信タイムアウト値.
     */
    private static final long TIMEOUT = 15000L ;
    
    /**
     * コールバックオブジェクト.
     */
    private HttpReceiveCallback call = null ;
    
    /**
     * キューオブジェクト.
     */
    private ReceiveLinkQueue queue = null ;
    
    /**
     * シーケンスID発行オブジェクト.
     */
    private SequenceSync sequence = null ;
    
    /**
     * 実行中フラグ.
     */
    private volatile boolean executionFlag = false ;
    
    /**
     * 実行同期.
     */
    private final Object execSync = new Object() ;
    
    /**
     * リクエストバッファ.
     */
    private final RequestBuffer requestBuffer = new RequestBuffer() ;
    
    /**
     * コネクション情報.
     */
    private final ConnectionInfoImpl connectionInfo = new ConnectionInfoImpl() ;
    
    /**
     * リクエスト情報.
     */
    private final HttpdRequestImpl request = new HttpdRequestImpl() ;
    
    /**
     * コンストラクタ.
     */
    private NioPoolThread() {
        
    }
    
    /**
     * コンストラクタ.
     * @param call コールバックオブジェクトを設定します.
     * @param queue キューオブジェクトを設定します.
     * @param sequence シーケンスID割り当てオブジェクトを設定します.
     * @param threadId 対象のスレッドIDを設定します.
     * @exception Exception 例外.
     */
    public NioPoolThread( HttpReceiveCallback call,ReceiveLinkQueue queue,SequenceSync sequence,int threadId )
        throws Exception {
        if( call == null || queue == null || sequence == null ) {
            throw new IllegalArgumentException( "引数は不正です" ) ;
        }
        this.call = call ;
        this.queue = queue ;
        this.sequence = sequence ;
        this.threadId = threadId ;
        this.startThread() ;
    }
    
    /**
     * デストラクタ.
     */
    protected void finalize() throws Exception {
        this.destroy() ;
    }
    
    /**
     * オブジェクト破棄.
     */
    public void destroy() {
        super.stopThread() ;
        executionFlag = false ;
    }
    
    /**
     * オブジェクトクリア.
     */
    protected void clear() {
        queue = null ;
        call = null ;
        executionFlag = false ;
    }
    
    /**
     * 実行処理.
     * @exception Exception 例外.
     */
    protected boolean execution() throws Exception {
        NioElement element = null ;
        element = queue.getQueue() ;
        if( element != null && element.isUse() ) {
            try {
                synchronized( execSync ) {
                    executionFlag = true ;
                }
                executionServer( element ) ;
                return true ;
            } catch( Exception e ) {
                if( element != null ) {
                    element.destroy() ;
                }
                element = null ;
                throw e ;
            } catch( Error err ) {
                if( element != null ) {
                    element.destroy() ;
                }
                element = null ;
                throw err ;
            } finally {
                synchronized( execSync ) {
                    executionFlag = false ;
                }
            }
        }
        return false ;
    }
    
    /**
     * 例外ハンドラ.
     */
    protected void toException( Throwable e ) {
        //LOG.warn( "error",e ) ;
    }
    
    /**
     * 実行中フラグを取得.
     * @return boolean [true]の場合、実行中です.
     */
    public boolean isExecution() {
        boolean ret ;
        synchronized( execSync ) {
            ret = executionFlag ;
        }
        return ret ;
    }
    
    /**
     * １つのコネクションを処理.
     */
    private void executionServer( NioElement element ) throws Exception {
        long time = -1L ;
        ConnectionInfo info = null ;
        int id = -1 ;
        try {
            // pipeline対応.
            // リクエスト情報が存在するまで実行する.
            while(true) {
                id = sequence.getId() ;
                info = connectionInfo.create( element ) ;
                if( LOG.isDebugEnabled() ) {
                    LOG.debug( new StringBuilder().append( " ## ***** start " ).
                        append( "receive(" ).append( id ).append( ") - [" ).
                        append( info.getInetAddress().getHostAddress() ).
                        append( ":" ).append( info.getPort() ).
                        append( "] - threadId:" ).append( threadId ).toString() ) ;
                }
                if( LOG.isDebugEnabled() ) {
                    time = System.currentTimeMillis() ;
                }
                // HTTPヘッダ＋Body読み込み.
                readHttpdRequest( request,info,requestBuffer,id ) ;
                request.setConnectionInfo( info ) ;
                // 実行処理.
                this.call.execution( info,id,request ) ;
                request.clear() ;
                // コメットフラグがOFFの場合の処理.
                // コメットフラグがONの場合は、何もしない.
                if( info.isComet() == false ) {
                    if( info.isUse() == true ) {
                        if( info.isCloseFlag() == true ) {
                            element.setCloseFlag( true ) ;
                        }
                        else {
                            if( info.recyclingConnection() == false ) {
                                element.setCloseFlag( true ) ;
                            }
                            else {
                                element.setCloseFlag( false ) ;
                            }
                        }
                    }
                    else {
                        element.destroy() ;
                    }
                }
                if( LOG.isDebugEnabled() ) {
                    LOG.debug( new StringBuilder().append( " ## @@@@@ end " ).
                        append( "## id[" ).append( id ).
                        append( "]-threadId:" ).append( threadId ).append( " " ).
                        append( ( System.currentTimeMillis() - time ) ).append( "msec" ).toString() ) ;
                }
                if( element.isBuffer() == false ) {
                    break ;
                }
                if( LOG.isDebugEnabled() ) {
                    LOG.debug( "#### pipline - next" ) ;
                }
            }
        } catch( Exception e ) {
            if( element != null ) {
                element.destroy() ;
            }
            if( LOG.isDebugEnabled() ) {
                LOG.debug( new StringBuilder().append( " ## $$$$$ error " ).
                    append( "## id[" ).append( id ).
                    append( "]-threadId:" ).append( threadId ).append( " " ).
                    append( ( System.currentTimeMillis() - time ) ).append( "msec:" ).
                    append( e.getMessage() ).toString() ) ;
            }
            throw e ;
        } finally {
            if( info != null ) {
                info.clear() ;
            }
        }
    }
    
    /**
     * 受信データをHttpdRequestに変換して取得.
     */
    private static final boolean readHttpdRequest( HttpdRequestImpl request,ConnectionInfo info,RequestBuffer requestBuffer,int seqId )
        throws Exception {
        if( info == null || info.isUse() == false ) {
            throw new IOException( "コネクション情報は利用できません" ) ;
        }
        InputStream in = null ;
        try {
            in = info.getInputStream( TIMEOUT ) ;
            ReadHttpdRequest.receiveHttpRequest( request,in,requestBuffer,seqId ) ;
            return true ;
        } catch( Exception e ) {
            throw e ;
        } finally {
            if( in != null ) {
                in.close() ;
            }
        }
    }
}
