package org.maachang.proxy.engine.net.http;

import java.io.BufferedInputStream;
import java.nio.channels.CancelledKeyException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.maachang.proxy.engine.net.RequestInfo;
import org.maachang.util.SequenceSync;

/**
 * 処理用プーリングスレッド.
 * 
 * @version 2007/11/17
 * @author  masahito suzuki
 * @since MaachangProxy 1.00
 */
class NetPoolThread extends Thread {
    private static final Log LOG = LogFactory.getLog( NetPoolThread.class ) ;
    private NetUseQueue queue = null ;
    private HttpCallbackReceive receive = null ;
    private SequenceSync sequence = null ;
    private volatile boolean stopFlag = true ;
    
    private NetPoolThread() {
        
    }
    
    public NetPoolThread( NetUseQueue queue,HttpCallbackReceive receive,
        SequenceSync sequence )
        throws Exception {
        this.queue = queue ;
        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 ;
            try {
                if( endFlag == true || isStop() == true ) {
                    break ;
                }
                conn = this.queue.getQueue() ;
                if( conn != null ) {
                    ConnectionInfo info = conn.getConnectionInfo() ;
                    if( info == null || info.isUse() == false ) {
                        continue ;
                    }
                    int seqId = sequence.getId() ;
                    if( LOG.isInfoEnabled() ) {
                        LOG.info( new StringBuilder().append( "receive(" ).append( seqId ).append( ") - [" ).
                            append( info.getSocket().socket().getInetAddress().getHostAddress() ).
                            append( ":" ).append( info.getSocket().socket().getPort() ).
                            append( "]" ).toString() ) ;
                    }
                    if( LOG.isDebugEnabled() ) {
                        long tm = System.currentTimeMillis() ;
                        this.receive.execution( info,seqId,readHttpdRequest( conn,seqId ) ) ;
                        LOG.debug( "## receiveTime("+seqId+") - "+( System.currentTimeMillis()-tm ) + "ms." ) ;
                    }
                    else {
                        this.receive.execution( info,seqId,readHttpdRequest( conn,seqId ) ) ;
                    }
                    if( conn.isCloseFlag() == true ) {
                        conn.destroy() ;
                    }
                    else {
                        boolean isCloseFlag = false ;
                        try {
                            if( info.isUse() == false ) {
                                isCloseFlag = true ;
                            }
                        } catch( Exception ex ) {
                            isCloseFlag = true ;
                        }
                        if( isCloseFlag == false ) {
                            info.update() ;
                            if( info.getCount() > 0 ) {
                                if( info.getTimeout() > 0L ) {
                                    info.recyclingConnection() ;
                                }
                            }
                            else {
                                isCloseFlag = true ;
                            }
                        }
                        if( isCloseFlag == true ) {
                            conn.destroy() ;
                        }
                    }
                }
                else {
                    sleep( WAIT_TIME ) ;
                }
            } catch( CloseClientException ce ) {
                errorFlag = true ;
            } 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 ) {
                    conn.destroy() ;
                }
            }
        }
        this.queue = null ;
        this.receive = null ;
        //LOG.info( "## Thread:" + this.getName() + " 終了" ) ;
        setStop( true ) ;
        if( threadDeach != null ) {
            throw threadDeach ;
        }
    }
    
    private static final RequestInfo readHttpdRequest( HttpSocket socket,int seqId )
        throws Exception {
        if( socket == null || socket.isClosed() == true ) {
            return null ;
        }
        BufferedInputStream buf = null ;
        try {
            buf = new BufferedInputStream( new NetInputStream( socket.inputStream() ) ) ;
            RequestInfo ret = ReadHttpdRequest.receiveHttpRequest( buf,seqId ) ;
            buf.close() ;
            buf = null ;
            socket.update() ;
            return ret ;
        } finally {
            if( buf != null ) {
                try {
                    buf.close() ;
                } catch( Exception e ) {
                }
                buf = null ;
            }
        }
    }
}
