package org.maachang.comet.net.nio ;

import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.maachang.util.ArrayBinary;
import org.maachang.util.thread.LoopThread;


/**
 * Nio受信サーバ群管理.
 * 
 * @version 2008/05/28
 * @author  masahito suzuki
 * @since   MaachangComet 1.1C
 */
class NioReceiveServerArray {
    
    /**
     * 最小サイズ.
     */
    private static final int MIN_SIZE = 1 ;
    
    /**
     * NioReceiveServer.
     */
    private NioReceiveServer[] recvs = null ;
    
    /**
     * 受信スレッド管理数.
     */
    private int size = -1 ;
    
    /**
     * シーケンスID.
     */
    private volatile int seq = 0 ;
    
    /**
     * コンストラクタ.
     */
    private NioReceiveServerArray() {
        
    }
    
    /**
     * コンストラクタ.
     * @param queue 対象のキューを設定します.
     * @param size 受信スレッド数を設定します.
     * @exception Exception 例外.
     */
    public NioReceiveServerArray( ReceiveLinkQueue queue,int size )
        throws Exception {
        if( queue == null ) {
            throw new IllegalArgumentException( "引数は不正です" ) ;
        }
        if( size <= MIN_SIZE ) {
            size = MIN_SIZE ;
        }
        NioReceiveServer[] rv = new NioReceiveServer[ size ] ;
        for( int i = 0 ; i < size ; i ++ ) {
            rv[ i ] = new NioReceiveServer( queue ) ;
        }
        this.recvs = rv ;
        this.size = size ;
        this.seq = 0 ;
    }
    
    /**
     * デストラクタ.
     */
    protected void finalize() throws Exception {
        destroy() ;
    }
    
    /**
     * オブジェクト破棄.
     */
    public synchronized void destroy() {
        if( recvs != null ) {
            int len = recvs.length ;
            for( int i = 0 ; i < len ; i ++ ) {
                if( recvs[ i ] != null ) {
                    recvs[ i ].destroy() ;
                }
                recvs[ i ] = null ;
            }
            recvs = null ;
        }
    }
    
    /**
     * Acceptされた要素を受信用にセット.
     * @param element 対象のElementを設定します.
     * @exception Exception 例外.
     */
    public synchronized void setElement( NioElement element )
        throws Exception {
        if( element == null ) {
            return ;
        }
        if( recvs == null ) {
            element.destroy() ;
            return ;
        }
        if( seq >= size ) {
            seq = 0 ;
        }
        recvs[ seq ].setElement( element ) ;
        seq ++ ;
    }
}

/**
 * 1つの受信ソケット.
 */
class NioReceiveServer extends LoopThread {
    
    /**
     * Selectorタイムアウト.
     */
    private static final int SELECTOR_TIMEOUT = 1 ;
    
    /**
     * 受信バッファサイズ.
     */
    private static final int BUFFER_LENGTH = 512 ;
    
    /**
     * Selector.
     */
    private Selector selector = null ;
    
    /**
     * 読み込み可能キュー.
     */
    private IsReadQueue isRecvQueue = null ;
    
    /**
     * 書き込み可能キュー.
     */
    private IsWriteQueue isWriteQueue = null ;
    
    /**
     * 受信処理キュー.
     */
    private ReceiveLinkQueue executionQueue = null ;
    
    /**
     * 受信バッファ.
     */
    private ByteBuffer buffer = null ;
    
    /**
     * コンストラクタ.
     */
    private NioReceiveServer() {
        
    }
    
    /**
     * コンストラクタ.
     * @param queue 受信実行スレッドとリンクするキューを設定します.
     * @exception Exception 例外.
     */
    public NioReceiveServer( ReceiveLinkQueue executionQueue )
        throws Exception {
        if( executionQueue == null ) {
            throw new IllegalArgumentException( "引数は不正です" ) ;
        }
        this.selector = Selector.open() ;
        this.isRecvQueue = new IsReadQueue() ;
        this.isWriteQueue = new IsWriteQueue() ;
        this.executionQueue = executionQueue ;
        this.buffer = ByteBuffer.allocateDirect( BUFFER_LENGTH ) ;
        this.startThread() ;
    }
    
    /**
     * デストラクタ.
     */
    protected void finalize() throws Exception {
        this.destroy() ;
    }
    
    /**
     * オブジェクト破棄.
     */
    public void destroy() {
        super.stopThread() ;
    }
    
    /**
     * AcceptされたNioElementを受信用にセット.
     * @param element 対象のElementを設定します.
     * @exception Exception 例外.
     */
    public synchronized void setElement( NioElement element )
        throws Exception {
        element.setIsWriteQueue( isWriteQueue ) ;
        isRecvQueue.append( element ) ;
    }
    
    /**
     * オブジェクトクリア.
     */
    protected void clear() {
        if( this.selector != null ) {
            try {
                Iterator itr = this.selector.keys().iterator();
                while(itr.hasNext()){
                    try {
                        SelectionKey key = (SelectionKey)itr.next();
                        Channel c = key.channel();
                        c.close();
                    } catch( Exception ee ) {
                    }
                }
            } catch(Exception e) {}
        }
        this.selector = null ;
        this.isRecvQueue = null ;
        this.isWriteQueue = null ;
        this.executionQueue = null ;
        this.buffer = null ;
    }
    
    /**
     * 実行処理.
     * @exception Exception 例外.
     */
    protected boolean execution() throws Exception {
        // selector処理.
        while ( selector.select( SELECTOR_TIMEOUT ) > 0 ) {
            //setSendRecv() ;
            Iterator itr = selector.selectedKeys().iterator() ;
            while(itr.hasNext()) {
                setSendRecv() ;
                SelectionKey key = ( SelectionKey )itr.next() ;
                itr.remove() ;
                
                // use-socket.
                if( key.isValid() ) {
                    boolean isReadable = key.isReadable() ;
                    boolean isWritable = key.isWritable() ;
                    // read.
                    if( isReadable ) {
                        NioElement em = null ;
                        try {
                            em = ( NioElement )key.attachment() ;
                            // 差分データ読み込み処理.
                            if( recv( em ) == false ) {
                                // データが0バイトの場合は、クライアント
                                // コネクション切断なので、要素を破棄する.
                                em.destroy() ;
                                continue ;
                            }
                            // データ送信がはじめての場合.
                            if( em.isExecutionFlag() == false ) {
                                // 接続条件をPoolキューに通知.
                                em.setExecutionFlag( true ) ;
                                executionQueue.append( em ) ;
                            }
                            key.interestOps( SelectionKey.OP_READ ) ;
                        } catch( Exception e ) {
                            if( em != null ) {
                                em.destroy() ;
                            }
                            continue ;
                        }
                    }
                    
                    // write.
                    if( isWritable ) {
                        NioElement em = null ;
                        try {
                            em = ( NioElement )key.attachment() ;
                            ByteBuffer buf = em.getWriteBuffer().getQueue() ;
                            // 送信データが存在する場合.
                            if( buf != null ) {
                                // 書き込みの終端を検知.
                                if( buf.limit() == 0 ) {
                                    // コネクションクローズの場合.
                                    if( em.isCloseFlag() == true ) {
                                        em.destroy() ;
                                    }
                                    // 書き込み終了なので、OP_READのみとする.
                                    else {
                                        key.interestOps( SelectionKey.OP_READ ) ;
                                    }
                                }
                                // 通常の送信データが存在する場合は、
                                // SocketChannelに書き込む.
                                else {
                                    SocketChannel ch = em.getChannel() ;
                                    ch.write( buf ) ;
                                }
                            }
                        } catch( Exception e ) {
                            if( em != null ) {
                                em.destroy() ;
                            }
                            continue ;
                        }
                    }
                    
                }
                // not-use-socket.
                else {
                    try { (( NioElement )key.attachment()).destroy() ; } catch( Exception e ){}
                }
            }
        }
        setSendRecv() ;
        return true ;
    }
    
    /**
     * 送受信用条件を付加.
     */
    private void setSendRecv() throws Exception {
        // 受信データ存在確認.
        while( isRecvQueue.size() > 0 ) {
            NioElement element = isRecvQueue.getQueue() ;
            if( element != null && element.isUse() ) {
                SelectionKey key = element.getChannel().register(
                    selector,SelectionKey.OP_READ,element ) ;
                if( key != null ) {
                    element.setKey( key ) ;
                }
                else {
                    element.destroy() ;
                }
            }
        }
        // 送信データ存在確認.
        while( isWriteQueue.size() > 0 ) {
            NioElement element = isWriteQueue.getQueue() ;
            if( element.isUse() == true && element.getKey() != null ) {
                element.getKey().interestOps( SelectionKey.OP_READ | SelectionKey.OP_WRITE ) ;
            }
        }
    }
    
    /**
     * データ受信処理.
     */
    private boolean recv( NioElement emt ) throws Exception {
        if( emt.isUse() == false ) {
            return false ;
        }
        ArrayBinary emtBuf = emt.getBuffer() ;
        SocketChannel channel = emt.getChannel() ;
        boolean flg = false ;
        for( ;; ) {
            buffer.clear() ;
            int len = channel.read( buffer ) ;
            if( len <= 0 ) {
                if( len <= -1 ) {
                    return false ;
                }
                break ;
            }
            buffer.flip() ;
            emtBuf.write( buffer ) ;
            flg = true ;
        }
        if( flg == true ) {
            emt.update() ;
        }
        return flg ;
    }
}

