package org.maachang.mimdb.server ;

import java.io.IOException;
import java.lang.ref.SoftReference;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.maachang.mimdb.core.MimdbQueryPrepared;
import org.maachang.mimdb.core.MimdbResult;
import org.maachang.mimdb.core.MimdbResultRow;
import org.maachang.mimdb.core.MimdbTable;
import org.maachang.mimdb.core.MimdbTableManager;
import org.maachang.mimdb.core.QueryCompileInfo;
import org.maachang.mimdb.core.impl.MimdbUtils;
import org.maachang.mimdb.core.util.AtomicNumber32;
import org.maachang.mimdb.core.util.ByteArrayIO;
import org.maachang.mimdb.core.util.ObjectBinary;
import org.maachang.mimdb.core.util.jsnappy.JSnappy;
import org.maachang.mimdb.core.util.jsnappy.JSnappyBuffer;

/**
 * Mimdb通信制御スレッド.
 * 
 * @version 2014/01/16
 * @author  masahito suzuki
 * @since MasterInMemDB 1.02
 */
public final class MimdbIOThread extends Thread {
    
    /** 定数. **/
    protected static final int LINGER = 5 ;
    protected static final boolean TCP_NO_DELAY = false ;
    protected static final boolean KEEP_ALIVE = false ;
    protected static final boolean OOB_INLINE = false ;
    protected static final int BUFFER_LENGTH = 8192 ;
    
    /** Selectorタイムアウト. **/
    private static final int SELECTOR_TIMEOUT = 1000 ;
    
    /** IOスレッド番号. **/
    private int threadNo ;
    
    /** セレクタ. **/
    private NioSelector selector ;
    
    /** 通信圧縮モード. **/
    private boolean compress ;
    
    /** スレッド停止フラグ. **/
    private final AtomicNumber32 stopFlag = new AtomicNumber32( 0 ) ;
    
    /** Acceptキュー. **/
    private final Queue<MimdbConnectElement> queue = new ConcurrentLinkedQueue<MimdbConnectElement>() ;
    
    /**
     * コンストラクタ.
     * @param no 対象のスレッド番号を設定します.
     * @param flg 圧縮モードが設定されます.
     * @param s FsHttpServerThreadを設定します.
     * @exception Exception 例外.
     */
    public MimdbIOThread( final int no,final boolean flg ) throws Exception {
        threadNo = no ;
        compress = flg ;
        selector = new NioSelector() ;
    }
    
    /** 情報出力. **/
    protected static final void print( final String s ) {
        MimdbServer.print( s ) ;
    }
    /** 情報出力. **/
    protected static final void println( final String s ) {
        MimdbServer.println( s ) ;
    }
    
    /**
     * スレッド開始.
     * @exception Exception 例外.
     */
    public void startThread() throws Exception {
        this.stopFlag.set( 1 ) ;
        this.setDaemon( true ) ;
        this.start() ;
    }
    
    /**
     * スレッド停止.
     */
    public void stopThread() {
        this.stopFlag.set( 0 ) ;
    }
    
    /**
     * スレッドが停止しているかチェック.
     * @return boolean [true]の場合、スレッド停止しています.
     */
    public boolean isStop() {
        return ( stopFlag.get() == 0 ) ;
    }
    
    /**
     * SocketChannelを登録.
     * @param channel SocketChannelを設定します.
     * @exception Exception 例外.
     */
    public void register( final SocketChannel channel )
        throws Exception {
        register( channel,BUFFER_LENGTH,BUFFER_LENGTH >> 1 ) ;
    }
    
    /**
     * SocketChannelを登録.
     * @param channel SocketChannelを設定します.
     * @exception Exception 例外.
     */
    public void register( final SocketChannel channel,final int sendBuffer,final int recvBuffer )
        throws Exception {
        
        try {
            
            // 対象のAccpet済みSocketChannelを受信処理Selectorに登録.
            channel.configureBlocking( false ) ;
            channel.setOption( StandardSocketOptions.SO_REUSEADDR,true ) ;
            channel.setOption( StandardSocketOptions.SO_KEEPALIVE,KEEP_ALIVE ) ;
            channel.setOption( StandardSocketOptions.TCP_NODELAY,TCP_NO_DELAY ) ;
            channel.setOption( StandardSocketOptions.SO_SNDBUF,sendBuffer ) ;
            channel.setOption( StandardSocketOptions.SO_RCVBUF,recvBuffer ) ;
            
            // 要素の初期化.
            MimdbConnectElement em = new MimdbConnectElement() ;
            em.channel = channel ;
            queue.offer( em ) ;
            selector.wakeup() ;
            
        } catch( Throwable t ) {
            try {
                channel.close() ;
            } catch( Throwable tt ) {
            }
        }
        
    }
    
    /**
     * 受信処理.
     */
    public void run() {
        ThreadDeath tdObject = null ;
        boolean endFlag = false ;
        
        int len ;
        Iterator<SelectionKey> it ;
        
        SelectionKey key = null ;
        SocketChannel channel = null ;
        MimdbConnectElement em = null ;
        
        final ByteBuffer buf = ByteBuffer.allocateDirect( BUFFER_LENGTH >> 1 ) ;
        final ByteArrayIO sendWork = new ByteArrayIO( BUFFER_LENGTH >> 1 ) ;
        final int[] off = new int[ 1 ] ;
        
        // 受信チェック用のバイナリ長. **/
        // mim   (3).
        // type  (1).
        // length(4).
        final byte[] tmp = new byte[ ConnectionDefine.HEAD_PLUS ] ;
        
        // スレッド開始.
        final NioSelector s = selector ;
        
        println( "****** ioThread [開始]." + threadNo + " *****" ) ;
        
        while( stopFlag.get() == 1 && !endFlag ) {
            
            try {
                
                // 新規コネクションが存在する場合.
                if( !queue.isEmpty() ) {
                    
                    // AcceptQueueにセットされた情報をSelectorに登録.
                    while( !queue.isEmpty() && stopFlag.get() == 1 ) {
                        if( ( em = queue.poll() ) == null ) {
                            continue ;
                        }
                        channel = (SocketChannel)em.channel ;
                        s.register( channel,SelectionKey.OP_READ,em ) ;
                        em = null ;
                    }
                    
                }
                
                // 処理条件が存在するかチェック.
                if( s.select( SELECTOR_TIMEOUT ) > 0 && ( it = s.iterator() ) != null ) {
                    
                    // 処理分ループ.
                    while( it.hasNext() && stopFlag.get() == 1 ) {
                        
                        // Socket処理実行.
                        key = it.next() ;
                        it.remove() ;
                        
                        // 取得情報が無効な場合は、オブジェクトクローズ.
                        if( key == null || !key.isValid() ) {
                            
                            // 対象キーが存在しない場合は処理しない.
                            if( key != null ) {
                                
                                // コネクション破棄.
                                NioSelector.destroyKey( key ) ;
                                
                            }
                            
                        }
                        ////////////////////////////////////////////////////////
                        // 読み込み処理.
                        ////////////////////////////////////////////////////////
                        else if( key.isReadable() ) {
                            
                            // ソケット情報の取得.
                            em = ( MimdbConnectElement )key.attachment() ;
                            
                            // データ受信.
                            buf.clear() ;
                            if( ( len = em.channel.read( buf ) ) <= 0 ) {
                                // 処理クローズの場合.
                                if( len == -1 ) {
                                    
                                    // コネクション破棄.
                                    NioSelector.destroyKey( key ) ;
                                }
                            }
                            // 情報が取得できた場合.
                            else {
                                
                                // 受信データを受信バッファにセット.
                                buf.flip() ;
                                em.recvBuffer.write( buf ) ;
                                
                                // 初めての受信の場合.
                                if( em.recvType == 0 ) {
                                    
                                    // 受信条件を解析.
                                    recvAnalysis( tmp,em ) ;
                                }
                                
                                // 受信完了チェック.
                                if( endRecv( off,sendWork,em,compress ) ) {
                                    
                                    // 受信完了の場合は、送信条件を許可する.
                                    key.interestOps( SelectionKey.OP_READ | SelectionKey.OP_WRITE ) ;
                                    
                                }
                            }
                            
                        }
                        ////////////////////////////////////////////////////////
                        // 書き込み処理.
                        ////////////////////////////////////////////////////////
                        else if( key.isWritable() ) {
                            
                            // ソケット情報の取得.
                            em = ( MimdbConnectElement )key.attachment() ;
                            
                            // 送信処理.
                            buf.clear() ;
                            if( send( buf,em ) ) {
                                // 送信が完了した場合は、送信条件をOFF.
                                key.interestOps( SelectionKey.OP_READ ) ;
                            }
                            
                        }
                        
                    }
                    
                }
                
            } catch( ThreadDeath td ) {
                NioSelector.destroyKey( key ) ;
                tdObject = td ;
                endFlag = true ;
            } catch( Throwable t ) {
                NioSelector.destroyKey( key ) ;
                // Exception 以上の例外を表示.
                if( t instanceof Error ) {
                    println( MimdbUtils.getStackTrace( t ) ) ;
                }
                // InterruptedException.
                // ThreadDeath
                // これ以外の例外は無視.
            }
        }
        
        // 後処理.
        stopFlag.set( 0 ) ;
        println( "****** ioThread [停止]." + threadNo + " *****" ) ;
        
        // セレクタ情報のクリア.
        s.close() ;
        selector = null ;
        
        // スレッドデスを返却.
        if( tdObject != null ) {
            throw tdObject ;
        }
    }
    
    /**
     * 受信情報を解析.
     */
    private static final void recvAnalysis( final byte[] tmp,final MimdbConnectElement em )
        throws Exception {
        
        // 受信データ長が、チェック分より少ない場合は、処理しない.
        if( em.recvBuffer.writeLength() < tmp.length ) {
            return ;
        }
        
        // 参照取得.
        em.recvBuffer.peek( tmp ) ;
        
        byte[] h = ConnectionDefine.HEAD ;
        int off = ConnectionDefine.HEAD_LENGTH ;
        
        // 先頭３バイトがmimかチェック.
        for( int i = 0 ; i < off ; i ++ ) {
            if( tmp[ i ] != h[ i ] ) {
                // 一致しない場合は、エラー.
                throw new IOException( "通信ヘッダが一致しません" ) ;
            }
        }
        
        int type = tmp[ off ] & 0xff ;
        int len = ( ( tmp[ off + 1 ] & 0xff ) |
            ( ( tmp[ off + 2 ] & 0xff ) << 8 ) |
            ( ( tmp[ off + 3 ] & 0xff ) << 16 ) |
            ( ( tmp[ off + 4 ] & 0xff ) << 24 ) ) ;
        
        // 長さが64kを越える場合.
        if( len <= 0 || len > 65535 ) {
            // エラー出力.
            throw new IOException( "データ長が長すぎです:" + len ) ;
        }
        
        // 正常な電文.
        em.recvType = type ;
        em.nowRecvLength = len ;
    }
    
    /** 受信完了時の処理. **/
    protected static final boolean endRecv( final int[] off,final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress )
        throws Exception {
        if( em.recvBuffer.writeLength() < em.nowRecvLength ) {
            return false ;
        }
        
        byte[] recv ;
        
        // タイプ取得.
        int type = em.recvType ;
        
        // 圧縮されている場合.
        if( ( em.recvType & 0x80 ) == 0x80 ) {
            
            // タイプを復元.
            type = em.recvType & 0x7f ;
            
            // 圧縮を解凍.
            recv = new byte[ em.nowRecvLength - ConnectionDefine.HEAD_PLUS ] ;
            em.recvBuffer.skip( ConnectionDefine.HEAD_PLUS ) ;
            em.recvBuffer.read( recv ) ;
            JSnappyBuffer buf = JSnappy.decompress( recv,0,recv.length ) ;
            recv = null ;
            recv = buf.toByteArray() ;
        }
        // 圧縮されていない場合.
        else {
            // 普通に取得.
            recv = new byte[ em.nowRecvLength - ConnectionDefine.HEAD_PLUS ] ;
            em.recvBuffer.skip( ConnectionDefine.HEAD_PLUS ) ;
            em.recvBuffer.read( recv ) ;
        }
        
        // 受信完了.
        em.recvType = 0 ;
        em.nowRecvLength = 0 ;
        
        // 処理タイプを元に情報を返信.
        switch( type ) {
            // テーブル情報要求.
            case ConnectionDefine.TABLE_INFO : 
                sendTable( off,sendWork,em,compress,recv ) ;
                return true ;
            
            // SQL文受信.
            case ConnectionDefine.EXECUTION_SQL :
                sendFirstResult( off,sendWork,em,compress,recv ) ;
                return true ;
            
            // 結果情報を要求.
            case ConnectionDefine.RESULT_GET :
                sendNextResult( off,sendWork,em,compress,recv ) ;
                return true ;
            
            // クローズ条件を返却.
            case ConnectionDefine.CLOSE :
                executionClose( off,sendWork,em,compress,recv ) ;
                return true ;
        }
        
        // それ以外の条件が受信された場合はエラー.
        throw new IOException( "不明な受信タイプ:" + type ) ;
    }
    
    /** 送信条件をパック化. **/
    private static final void packSendData( final ByteArrayIO sendWork,
        int type,final MimdbConnectElement em,boolean compress )
        throws Exception {
        
        byte[] b = sendWork.toByteArray() ;
        sendWork.clear() ;
        
        // 圧縮条件の場合.
        if( compress && b.length >= ConnectionDefine.COMPRESS_LENGTH ) {
            
            // データを圧縮.
            JSnappyBuffer buf = JSnappy.compress( b,0,b.length ) ;
            b = buf.toByteArray() ;
            type |= 0x80 ;
            
        }
        
        // ヘッダ条件を作成.
        byte[] tmp = new byte[ ConnectionDefine.HEAD_PLUS ] ;
        int off = ConnectionDefine.HEAD_LENGTH ;
        System.arraycopy( ConnectionDefine.HEAD,0,tmp,0,off ) ;
        tmp[ off ] = (byte)type ;
        
        int len = b.length + ConnectionDefine.HEAD_PLUS ;
        tmp[ off+1 ] = (byte)( len & 0xff ) ;
        tmp[ off+2 ] = (byte)( ( len & 0xff00 ) >> 8 ) ;
        tmp[ off+3 ] = (byte)( ( len & 0xff0000 ) >> 16 ) ;
        tmp[ off+4 ] = (byte)( ( len & 0xff000000 ) >> 24 ) ;
        
        // 送信対象条件をセット.
        em.sendHeader = tmp ;
        em.nowSendHeaderOffset = 0 ;
        em.sendBuffer = b ;
        em.nowSendOffset = 0 ;
        em.nowSendLength = b.length ;
        
        // 最終アクセス時間をセット.
        em.lastAccess = System.currentTimeMillis() ;
    }
    
    /** エラーメッセージ返却. **/
    private static final void sendError( final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress,final String message )
        throws Exception {
        
        // バッファクリア.
        sendWork.clear() ;
        
        // エラーメッセージをセット.
        ObjectBinary.encode( sendWork,message ) ;
        
        // 送信条件をセット.
        packSendData( sendWork,ConnectionDefine.ERROR,em,compress ) ;
    }
    
    /** テーブル情報要求. **/
    private static final void sendTable( final int[] off,final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress,final byte[] recv )
        throws Exception {
        
        off[ 0 ] = 0 ;
        int len = recv.length ;
        
        // 更新ID.
        long dbId = (Long)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // テーブル名.
        String tableName = (String)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // 指定テーブル名を取得.
        MimdbTable table = (MimdbTable)MimdbTableManager.getInstance().get( tableName ) ;
        
        // テーブルが存在しない場合はエラー返却.
        if( table == null ) {
            sendError( sendWork,em,compress,"対象のテーブル名[" + tableName + "]は存在しません" ) ;
        }
        // テーブル更新IDが一致しない場合.
        else if( dbId <= 0 || table.getDbId() != dbId ) {
            
            sendWork.clear() ;
            
            // フラグ[false]をセット.
            ObjectBinary.encode( sendWork,false ) ;
            
            // テーブル構成をバイナリ変換.
            table.getBinary( sendWork ) ;
            
            // データ送信.
            packSendData( sendWork,ConnectionDefine.TABLE_INFO,em,compress ) ;
            
        }
        // テーブル更新IDが一致する場合.
        else {
            
            sendWork.clear() ;
            
            // フラグ[true]をセット.
            ObjectBinary.encode( sendWork,true ) ;
            
            // データ送信.
            packSendData( sendWork,ConnectionDefine.TABLE_INFO,em,compress ) ;
        }
    }
    
    /** SQL文受信. **/
    private static final void sendFirstResult( final int[] off,final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress,final byte[] recv )
        throws Exception {
        
        off[ 0 ] = 0 ;
        int len = recv.length ;
        int newId = -1 ;
        int offset = -1 ;
        int limit = -1 ;
        
        QueryCompileInfo prepared = null ;
        SoftReference<QueryCompileInfo> o ;
        
        // コンパイル済みの条件.
        boolean preparedFlag = (Boolean)ObjectBinary.decodeBinary( off,recv,len ) ;
        boolean nowSendFlag = true ;
        int preparedId = -1 ;
        
        // コンパイル済みの条件の場合.
        if( preparedFlag ) {
            
            // コンパイル済みの条件を新規送信チェック.
            nowSendFlag = (Boolean)ObjectBinary.decodeBinary( off,recv,len ) ;
            
            // オフセット、リミット値を取得.
            offset = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
            limit = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
            
            // コンパイル済みの新規送信でない場合.
            if( !nowSendFlag ) {
                
                // コンパイル済み条件ID.
                preparedId = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
                
                // 指定条件のIDが存在するかチェック.
                if( preparedId <= -1 || ( o = em.preparedList.get( preparedId ) ) == null ||
                    ( prepared = o.get() ) == null ) {
                    
                    // 条件を削除.
                    em.preparedList.remove( preparedId ) ;
                    
                    //
                    // コンパイル済みの条件が存在しないことを通知.
                    //
                    
                    sendWork.clear() ;
                    
                    // フラグ[false]をセット.
                    ObjectBinary.encode( sendWork,false ) ;
                    
                    // データ送信.
                    packSendData( sendWork,ConnectionDefine.EXECUTION_SQL,em,compress ) ;
                    return ;
                }
                
                // 現在のプリコンパイル済みIDをセット.
                newId = preparedId ;
                
            }
            // コンパイル済みの新規送信.
            else {
                
                // コンパイル済みのオブジェクトを復元.
                prepared = new QueryCompileInfo() ;
                off[0] = prepared.toObject( recv,off[0],len ) ;
                
            }
            
        }
        // コンパイル済みの条件でない場合[Statement実行].
        else {
            
            // オフセット、リミット値を取得.
            offset = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
            limit = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
            
            // コンパイル済みのオブジェクトを復元.
            prepared = new QueryCompileInfo() ;
            off[0] = prepared.toObject( recv,off[0],len ) ;
            
        }
        
        // 処理を実行する.
        MimdbQueryPrepared ps = new MimdbQueryPrepared( prepared ) ;
        
        // パラメータ数分、条件をセット.
        if( ps.paramsLength() > 0 ) {
            int plen = ps.paramsLength() ;
            for( int i = 0 ; i < plen ; i ++ ) {
                ps.setParams( i,ObjectBinary.decodeBinary( off,recv,len ) ) ;
            }
        }
        MimdbResult res = null ;
        
        // オフセット、リミット値をセット.
        if( offset >= 0 ) {
            ps.setOffset( offset ) ;
        }
        if( limit >= 0 ) {
            ps.setLimit( limit ) ;
        }
        
        // 処理実行.
        try {
            res = ps.executeQuery() ;
        } catch( Exception e ) {
            //sendError( sendWork,em,compress,MimdbUtils.getStackTrace( e ) ) ;
            sendError( sendWork,em,compress,e.getMessage() ) ;
            return ;
        }
        
        // 処理成功.
        if( preparedFlag && nowSendFlag ) {
            
            // 新規のプリコンパイル済み条件の場合は、
            // コネクション条件に、プリコンパイル済み情報をセット.
            newId = em.getPreparedSequenceId() ;
            em.preparedList.put( newId,new SoftReference<QueryCompileInfo>( prepared ) ) ;
        }
        
        // 結果情報を格納.
        int resId = em.getResultSetSequenceId() ;
        em.resultList.put( resId,res ) ;
        
        // 結果情報を返却.
        sendWork.clear() ;
        
        // フラグ[true]をセット.
        ObjectBinary.encode( sendWork,true ) ;
        
        // プリコンパイル済みIDをセット.
        if( newId != -1 ) {
            // -1以外は送信する(-1はプリコンパイル利用をしない場合のみ).
            ObjectBinary.encode( sendWork,newId ) ;
        }
        
        // 結果情報IDをセット.
        ObjectBinary.encode( sendWork,resId ) ;
        
        // データ件数を返却.
        ObjectBinary.encode( sendWork,res.length() ) ;
        
        // 最大データ件数を返却.
        ObjectBinary.encode( sendWork,res.maxLength() ) ;
        
        // データ送信.
        packSendData( sendWork,ConnectionDefine.EXECUTION_SQL,em,compress ) ;
    }
    
    /** ResultSet受信要求. **/
    private static final void sendNextResult( final int[] off,final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress,final byte[] recv )
        throws Exception {
        
        off[ 0 ] = 0 ;
        int len = recv.length ;
        
        // 結果管理ID.
        int id = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // 取得開始位置.
        int offset = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // フェッチサイズ.
        int fetch = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // 指定更新IDの条件が存在するかチェック.
        MimdbResult res = em.resultList.get( id ) ;
        
        // 指定IDが存在しない場合.
        if( res == null ) {
            
            //
            // 情報が存在しないことを通知.
            // ResultSetがクローズされている場合.
            //
            
            sendWork.clear() ;
            
            // フラグ[false]をセット.
            ObjectBinary.encode( sendWork,false ) ;
            
            // データ送信.
            packSendData( sendWork,ConnectionDefine.RESULT_GET,em,compress ) ;
            return ;
            
        }
        
        // 情報をセットして返却.
        int length = res.length() - offset ;
        length = ( fetch > length ) ? length : fetch ;
        
        sendWork.clear() ;
        
        // フラグ[true]をセット.
        ObjectBinary.encode( sendWork,true ) ;
        
        // データ返却数をセット.
        ObjectBinary.encode( sendWork,length ) ;
        
        // データ移動.
        if( res.absolute( offset ) ) {
            
            int j,lenJ ;
            
            MimdbResultRow row = res.get() ;
            lenJ = row.size() ;
            
            // パラメータ長をセット.
            ObjectBinary.encode( sendWork,lenJ ) ;
            
            // データセット.
            for( int i = 0 ; i < length ; i ++ ) {
                for( j = 0 ; j < lenJ ; j ++ ) {
                    ObjectBinary.encode( sendWork,row.getValue( j ) ) ;
                }
                if( res.next() ) {
                    row = res.get() ;
                }
                else {
                    break ;
                }
            }
            
        }
        
        // 情報送信.
        packSendData( sendWork,ConnectionDefine.RESULT_GET,em,compress ) ;
        
    }
    
    /** クローズ要求. **/
    private static final void executionClose( final int[] off,final ByteArrayIO sendWork,
        final MimdbConnectElement em,final boolean compress,final byte[] recv )
        throws Exception {
        
        off[ 0 ] = 0 ;
        int len = recv.length ;
        
        // クローズタイプ(1:コネクション,2:ResultSet).
        int type = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // ID.
        int id = (Integer)ObjectBinary.decodeBinary( off,recv,len ) ;
        
        // 送信バッファクリア.
        sendWork.clear() ;
        
        // コネクションクローズ.
        if( type == ConnectionDefine.CLOSE_CONNECTION ) {
            
            em.clear() ;
            
            // フラグ[true]をセット.
            ObjectBinary.encode( sendWork,true ) ;
            
            // データ送信.
            packSendData( sendWork,ConnectionDefine.CLOSE,em,compress ) ;
        }
        // ResultSetクローズ.
        else if( type == ConnectionDefine.CLOSE_RESULT_SET ) {
            
            MimdbResult res = em.resultList.get( id ) ;
            // 情報が存在する場合.
            if( res != null ) {
                
                // 管理情報から削除.
                em.resultList.remove( id ) ;
                res.clear() ;
                
                // フラグ[true]をセット.
                ObjectBinary.encode( sendWork,true ) ;
                
            }
            // 情報が存在しない場合.
            else {
                
                // フラグ[false]をセット.
                ObjectBinary.encode( sendWork,false ) ;
            }
            
            // データ送信.
            packSendData( sendWork,ConnectionDefine.CLOSE,em,compress ) ;
            
        }
        // タイプ不明.
        else {
            sendError( sendWork,em,compress,"クローズ対象の処理タイプ[" + type + "]は不明です" ) ;
        }
    }
    
    /** データ送信. **/
    private static final boolean send( final ByteBuffer buf,final MimdbConnectElement em )
        throws Exception {
        
        // 送信終了.
        if( em.sendBuffer == null ) {
            return true ;
        }
        
        // ヘッダが未送信状態の場合.
        if( em.sendHeader != null ) {
            
            // ヘッダセット.
            final int headerLen = em.sendHeader.length - em.nowSendHeaderOffset ;
            buf.put( em.sendHeader,em.nowSendHeaderOffset,headerLen ) ;
            
            // bodyセット.
            int len = buf.remaining() ;
            if( len > em.nowSendLength ) {
                len = em.nowSendLength ;
            }
            
            // 送信処理.
            buf.put( em.sendBuffer,0,len ) ;
            buf.flip() ;
            int sendLen = em.channel.write( buf ) ;
            
            // ヘッダ送信長よりも少ない場合.
            if( sendLen < headerLen ) {
                
                // ヘッダ送信された内容をセット.
                em.nowSendHeaderOffset += sendLen ;
                return false ;
            }
            
            // ヘッダ情報が送信された場合.
            em.sendHeader = null ;
            em.nowSendHeaderOffset = 0 ;
            
            // 確実に送信された長さをBodyのオフセット値にセット.
            em.nowSendOffset += ( sendLen - headerLen ) ;
            
        }
        else {
            
            // 送信長をセット.
            int len = buf.limit() ;
            if( len > em.nowSendLength - em.nowSendOffset ) {
                len = em.nowSendLength - em.nowSendOffset ;
            }
            
            // 送信処理.
            buf.put( em.sendBuffer,em.nowSendOffset,len ) ;
            buf.flip() ;
            int sendLen = em.channel.write( buf ) ;
            
            // 確実に送信された長さをBodyのオフセット値にセット.
            em.nowSendOffset += sendLen ;
            
        }
        
        // 送信長が一致する場合は、送信完了.
        if( em.nowSendLength <= em.nowSendOffset ) {
            em.sendHeader = null ;
            em.nowSendHeaderOffset = 0 ;
            em.sendBuffer = null ;
            em.nowSendLength = 0 ;
            em.nowSendOffset = 0 ;
            return true ;
        }
        return false ;
        
    }
}

