package org.maachang.connector ;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.net.InetAddress;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.maachang.util.ConvertParam;

/**
 * ConnectorPoolThread.
 * 
 * @version 2008/05/25
 * @author masahito suzuki
 * @since MaachangBase 1.00
 */
class ConnectorPoolThread extends Thread {
    
    /**
     * 受信可能最大データ長.
     */
    private static final int MAX_BYTE = 32 * 0x00100000 ;
    private static final Log LOG = LogFactory.getLog( ConnectorPoolThread.class ) ;
    
    private volatile boolean stopFlag = true ;
    private byte[] headerBinary = null ;
    private ConnectorManager manager = null ;
    private ConnectorQueue queue = null ;
    private ConnectorCallback callback = null ;
    
    private ConnectorPoolThread() {
        
    }
    
    public ConnectorPoolThread( byte[] headerBinary,ConnectorManager manager,
        ConnectorCallback callback,ConnectorQueue queue )
        throws Exception {
        if( headerBinary == null || headerBinary.length <= 0 ||
            manager == null || callback == null || queue == null ) {
            throw new IllegalArgumentException( "引数は不正です" ) ;
        }
        this.headerBinary = headerBinary ;
        this.manager = manager ;
        this.callback = callback ;
        this.queue = queue ;
        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 = 30L ;
    
    public void run() {
        boolean endFlag = false ;
        ThreadDeath threadDeach = null ;
        for( ;; ) {
            if( endFlag == true || isStop() == true ) {
                break ;
            }
            ConnectorSession s = null ;
            InetAddress addr = null ;
            int port = -1 ;
            try {
                s = this.queue.getQueue() ;
                if( s == null ) {
                    Thread.sleep( WAIT ) ;
                    continue ;
                }
                else if( s.isClosed() == true ) {
                    s = null ;
                    continue ;
                }
                else {
                    addr = s.getInetAddress() ;
                    port = s.getPort() ;
                    byte[] b = receive( s ) ;
                    if( b != null ) {
                        s.executionUpdateTime() ;
                        try {
                            // echo処理チェック.
                            if( echo( s,b ) == false ) {
                                this.callback.execution( s,b ) ;
                            }
                        } finally {
                            if( s != null ) {
                                s.updateTime() ;
                            }
                        }
                    }
                    b = null ;
                    //if( s.isClosed() == false && s.isLocalHost() == false ) {
                    if( s.isClosed() == false ) {
                        s.updateTime() ;
                        manager.append( s ) ;
                        s = null ;
                    }
                }
            } catch( ConnectorNotHeaderException cn ) {
                if( addr == null ) {
                    LOG.warn( "## 不正な通信ヘッダを検知しました" ) ;
                }
                else {
                    LOG.warn( "## 不正な通信ヘッダを検知しました["+addr.getHostAddress()+"/"+port+"]" ) ;
                }
                s.destroy() ;
                s = null ;
            } catch( InterruptedException ie ) {
                endFlag = true ;
            } catch( OutOfMemoryError mem ) {
            } catch( Exception e ) {
                if( s != null ) {
                    s.destroy() ;
                }
                s = null ;
            } catch( ThreadDeath td ) {
                endFlag = true ;
                threadDeach = td ;
            } finally {
                if( s != null ) {
                    s.destroy() ;
                }
                s = null ;
            }
        }
        this.queue = null ;
        setStop( true ) ;
        if( threadDeach != null ) {
            throw threadDeach ;
        }
    }
    
    private byte[] receive( ConnectorSession s ) throws Exception {
        int headerLen = headerBinary.length ;
        int len = -1 ;
        ByteArrayOutputStream bs = new ByteArrayOutputStream() ;
        InputStream in = s.inputStream() ;
        try {
            if( s.isClosed() == true ) {
                bs.close() ;
                bs = null ;
                return null ;
            }
            byte[] ret = null ;
            int cnt = 0 ;
            for( ;; ) {
                int n = in.read() ;
                if( n <= -1 ) {
                    // 切断された場合は、オブジェクトを破棄.
                    s.destroy() ;
                    break ;
                }
                bs.write( n ) ;
                cnt ++ ;
                if( len <= -1 ) {
                    if( cnt >= headerLen + 4 ) {
                        byte[] bin = bs.toByteArray() ;
                        for( int i = 0 ; i < headerLen ; i ++ ) {
                            if( headerBinary[ i ] != bin[ i ] ) {
                                throw new ConnectorNotHeaderException( "通信ヘッダは不正です" ) ;
                            }
                        }
                        len = ConvertParam.convertInt( headerLen,bin ) ;
                        // 不正な受信サイズ.
                        if( len <= 0 || len >= MAX_BYTE ) {
                            s.destroy() ;
                            s = null ;
                            bs.close() ;
                            bs = null ;
                            return null ;
                        }
                    }
                }
                else if( cnt >= len ) {
                    ret = bs.toByteArray() ;
                    break ;
                }
            }
            bs.close() ;
            bs = null ;
            return ret ;
        } catch( ConnectorNotHeaderException ce ) {
            if( s != null ) {
                s.destroy() ;
            }
            throw ce ;
        } catch( Exception e ) {
            if( s != null ) {
                s.destroy() ;
            }
            return null ;
        } finally {
            if( bs != null ) {
                try {
                    bs.close() ;
                } catch( Exception e ) {
                }
            }
            bs = null ;
        }
    }
    
    /**
     * echo処理.
     */
    private final boolean echo( ConnectorSession c,byte[] b ) {
        try {
            if( b.length == headerBinary.length + 8 ) {
                int n = ConvertParam.convertInt( headerBinary.length + 4,b ) ;
                if( n == EchoClientConnector.ECHO_CODE ) {
                    try {
                        c.send( ConvertParam.convertInt( EchoClientConnector.ECHO_CODE ) ) ;
                    } catch( Exception ee ) {
                    }
                    return true ;
                }
            }
        } catch( Exception e ) {
        }
        return false ;
    }
}

