package org.maachang.mimdb.server ;

import java.io.BufferedOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.channels.ServerSocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.maachang.mimdb.MimdbDatabase;
import org.maachang.mimdb.core.MimdbMetaData;
import org.maachang.mimdb.core.MimdbQuery;
import org.maachang.mimdb.core.MimdbResult;
import org.maachang.mimdb.core.MimdbResultRow;
import org.maachang.mimdb.core.MimdbTableManager;
import org.maachang.mimdb.core.util.AtomicNumber32;
import org.maachang.mimdb.core.util.AtomicObject;

/**
 * Mimdbメンテナンスサーバー処理.
 * 
 * @version 2013/11/08
 * @author masahito suzuki
 * @since MasterInMemDB 1.00
 */
public class MimdbMaintenanceServer {
    private String addr = null ;
    private int port = -1 ;
    private int backlog = -1 ;
    
    private AcceptMaintenanceServer server = null ;
    
    private MimdbMaintenanceServer() {}
    
    /**
     * コンストラクタ.
     * @param map データベース管理テーブルを設定します.
     * @parma addr 対象のバインドIPアドレスを設定します.
     *             [null]を設定した場合、バインドアドレスは[0.0.0.0]となります.
     * @param port 対象の受信ポート番号を設定します.
     * @param backlog 接続最大数を設定します.
     *                [-1]の場合は、規定数(1)が設定されます.
     * @exception Exception 例外.
     */
    public MimdbMaintenanceServer( Map<String,MimdbDatabase> map,String addr,int port,int backlog )
        throws Exception {
        if( backlog <= 0 ) {
            backlog = 1 ;
        }
        AcceptMaintenanceServer s = new AcceptMaintenanceServer( map,addr,port,backlog ) ;
        this.server = s ;
        this.addr = addr ;
        this.port = port ;
        this.backlog = backlog ;
    }
    
    /**
     * オブジェクト破棄.
     */
    public void destroy() {
        this.server.stopThread() ;
        this.server = null ;
        this.addr = null ;
        this.port = -1 ;
        this.backlog = -1 ;
    }
    
    /**
     * メンテナンス用サーバーが動作中かチェック.
     * @return boolean [true]の場合、動作中です.
     */
    public boolean isStart() {
        return ( server != null ) ? server.isThread() : false ;
    }
    
    /**
     * バインドIPアドレスを取得.
     * @return String バインドIPアドレスが返却されます.
     */
    public String getBindAddress() {
        return addr ;
    }
    
    /**
     * バインドポート番号を取得.
     * @return int バインドポート番号が返却されます.
     */
    public int getBindPort() {
        return port ;
    }
    
    /**
     * 接続最大数を取得.
     * @return int 接続最大数が返却されます.
     */
    public int getBacklog() {
        return backlog ;
    }
}

/** Accept用スレッド. **/
class AcceptMaintenanceServer extends Thread {
    private AcceptMaintenanceServer() {}
    
    /** サーバーソケット. **/
    private ServerSocket server ;
    
    /** Socket受信処理用. **/
    private ReceiveMaintenanceServer[] recvList ;
    
    /** スレッド停止フラグ. **/
    private final AtomicNumber32 stopFlag = new AtomicNumber32( 0 ) ;
    
    /** Socket受信処理用割り当て項番. **/
    private int recvSeq = 0 ;
    
    /** Acceptサーバーソケットの生成. **/
    public AcceptMaintenanceServer( Map<String,MimdbDatabase> map,String bind,int port,int backlog )
        throws Exception {
        
        // 受信ソケット処理スレッド数の作成.
        int len = ( int )( backlog * 1.5 ) ;
        recvList = new ReceiveMaintenanceServer[ len ] ;
        for( int i = 0 ; i < len ; i ++ ) {
            recvList[ i ] = new ReceiveMaintenanceServer( map ) ;
        }
        
        // 通信はIPV4のみ.
        //System.setProperty( "java.net.preferIPv4Stack","true" ) ;
        // DNSキャッシュは300秒.
        System.setProperty( "networkaddress.cache.ttl","300" ) ;
        // DNS解決失敗した場合のキャッシュ保持しない.
        System.setProperty( "networkaddress.cache.negative.ttl","0" ) ;
        
        ServerSocketChannel channel = ServerSocketChannel.open() ;
        channel.configureBlocking( true ) ;
        ServerSocket soc = channel.socket() ;
        soc.setReuseAddress( true ) ;
        soc.setSoTimeout( 1000 ) ;// acceptタイムアウトは1秒.
        //soc.setPerformancePreferences( 1,2,0 ) ;
        //soc.setReceiveBufferSize( 0x00020000 ) ;
        
        // サーバーソケットBind.
        if( bind == null ) {
            soc.bind( new InetSocketAddress( port ),backlog ) ;
        }
        else {
            soc.bind( new InetSocketAddress( InetAddress.getByName( bind ),port ),backlog ) ;
        }
        server = soc ;
        stopFlag.set( 1 ) ;
        this.setDaemon( true ) ;
        this.start() ;
    }
    
    /** スレッド停止. **/
    public void stopThread() {
        stopFlag.set( 0 ) ; // Acceptスレッド停止.
        // 受信スレッドの停止.
        if( recvList != null ) {
            int len = recvList.length ;
            for( int i = 0 ; i < len ; i ++ ) {
                recvList[ i ].stopThread() ;
            }
            recvList = null ;
        }
    }
    
    /** スレッドが実行中かチェック. **/
    public boolean isThread() {
        return stopFlag.get() != 0 ;
    }
    
    /** スレッド処理. **/
    public void run() {
        Socket s = null ;
        int cnt = 0 ;
        int len = recvList.length ;
        try {
            while( true ) {
                try {
                    // 停止フラグが設定されている場合はスレッド処理を停止.
                    if( stopFlag.get() == 0 ) {
                        break ;
                    }
                    // ソケット受信.
                    if( ( s = server.accept() ) == null ) {
                        continue ;
                    }
                    // 停止フラグが設定されている場合はスレッド処理を停止.
                    if( stopFlag.get() == 0 ) {
                        break ;
                    }
                    // 空いているスレッドに割り当てる.
                    cnt = 0 ;
                    while( cnt ++ < len ) {
                        if( recvSeq >= len ) {
                            recvSeq = 0 ;
                        }
                        if( recvList[ recvSeq ++ ].setSocket( s ) ) {
                            s = null ;
                            break ;
                        }
                    }
                    // 割り当て先が存在しない場合は、s != nullなので、強制切断となる.
                } catch( ThreadDeath td ) {
                    throw td ;
                } catch( Throwable t ) {
                    // エラーは無視.
                } finally {
                    if( s != null ) {
                        try {
                            s.close() ;
                        } catch( Exception e ) {
                        }
                    }
                }
            }
            
        } catch( ThreadDeath td ) {
            throw td ;
        } finally {
            stopThread() ; // スレッド停止.
        }
    }
    
}

/** 受信用スレッド. **/
class ReceiveMaintenanceServer extends Thread {
    /** 受信バッファ. **/
    private static final int MAX_READ = 4096 ;
    
    /** 文字コード. **/
    private static final String CHARSET = "UTF8" ;
    
    /** 組み込みオブジェクト. **/
    private Map<String,MimdbDatabase> map = null ;
    
    /** ソケット. **/
    private final AtomicObject<Socket> socket = new AtomicObject<Socket>() ;
    
    /** スレッド停止フラグ. **/
    private final AtomicNumber32 stopFlag = new AtomicNumber32( 0 ) ;
    
    /** 待機オブジェクト. **/
    private final Object waitObject = new Object() ;
    
    /** ソケットスレッド. **/
    public ReceiveMaintenanceServer( Map<String,MimdbDatabase> map ) {
        this.map = map ;
        stopFlag.set( 1 ) ;
        this.setDaemon( true ) ;
        this.start() ;
    }
    
    /** スレッド停止. **/
    public void stopThread() {
        stopFlag.set( 0 ) ; // スレッド停止.
        Socket s = socket.get() ;
        if( s != null ) {
            try {
                s.close() ;
            } catch( Exception e ) {
            }
        }
    }
    
    /** 新しいソケットオブジェクトをセット. **/
    public boolean setSocket( Socket s )
        throws Exception {
        if( stopFlag.get() == 0 ) {
            return false ;
        }
        if( socket.get() == null ) {
            // 受信ソケット初期設定.
            s.setReuseAddress( true ) ;
            s.setSoLinger( true,5 ) ;
            s.setSendBufferSize( 32768 ) ;
            s.setReceiveBufferSize( 32768 ) ;
            s.setKeepAlive( false ) ;
            s.setTcpNoDelay( true ) ;
            s.setOOBInline( true ) ;
            s.setSoTimeout( 1000 ) ;
            //s.setPerformancePreferences( 1,2,0 ) ;
            //s.setTrafficClass( 0x00000004 | 0x00000008 | 0x00000010 ) ;
            
            // ソケットをセットして、スレッド再開.
            socket.set( s ) ;
            synchronized( waitObject ) {
                waitObject.notify() ;
            }
            return true ;
        }
        return false ;
    }
    
    /** スレッド処理. **/
    public void run() {
        Socket s = null ;
        byte[] buf = new byte[ MAX_READ ] ;
        try {
            while( true ) {
                // 受信処理.
                try {
                    // 停止フラグが設定されている場合はスレッド処理を停止.
                    if( stopFlag.get() == 0 ) {
                        break ;
                    }
                    if( ( s = socket.get() ) == null ) {
                        // ソケットオブジェクトが定義されていない場合は待機.
                        synchronized( waitObject ) {
                            waitObject.wait( 100 ) ;
                        }
                        // waitから外れた場合にソケットが定義されていない場合はコンテニュ.
                        if( ( s = socket.get() ) == null ) {
                            continue ;
                        }
                    }
                    // 停止フラグが設定されている場合はスレッド処理を停止.
                    if( stopFlag.get() == 0 ) {
                        break ;
                    }
                    // 受信処理.
                    if( recv( s,buf ) ) {
                        // ソケット正常クリア.
                        s = null ;
                    }
                } catch( ThreadDeath td ) {
                    throw td ;
                } catch( Throwable t ) {
                    // エラーは無視.
                } finally {
                    if( s != null ) {
                        socket.set( null ) ;
                        try {
                            s.getInputStream().close() ;
                        } catch( Exception e ) {
                        }
                        try {
                            s.getOutputStream().close() ;
                        } catch( Exception e ) {
                        }
                        try {
                            s.close() ;
                        } catch( Exception e ) {
                        }
                    }
                }
            }
        } catch( ThreadDeath td ) {
            throw td ;
        } finally {
            stopThread() ; // スレッド停止.
        }
    }
    
    /** 受信処理. **/
    private final boolean recv( Socket s,byte[] buf ) throws Exception {
        int len = -1 ;
        int recvLen = -1 ;
        int off = 0 ;
        InputStream in = s.getInputStream() ;
        // データ受信処理.
        while( true ) {
            if( stopFlag.get() == 0 ) {
                return false ;
            }
            // データ受信.
            try {
                if( ( len = in.read( buf,off,MAX_READ - off ) ) <= 0 ) {
                    if( len <= -1 ) {
                        return false ;
                    }
                    continue ;
                }
            } catch( SocketTimeoutException st ) {
                continue ;
            }
            // 初っ端の情報取得.
            if( off == 0 ) {
                // 先頭２バイトの情報を受信バッファ最大長として設定.
                recvLen = ( buf[ 0 ] & 0x000000ff ) | ( ( buf[ 1 ] & 0x000000ff ) << 8 ) ;
                // ヘッダデータ長がバッファ長を超える場合.
                if( recvLen + 2 >= MAX_READ ) {
                    return false ;
                }
            }
            off += len ;
            // 受信終了.
            if( off >= recvLen + 2 ) {
                break ;
            }
        }
        // 対象条件を文字列化.
        String rv = new String( buf,2,recvLen,CHARSET ) ;
        
        // 文字条件から処理を行う.
        execution( map,rv,new BufferedOutputStream( s.getOutputStream() ),buf ) ;
        
        return true ;
    }
    
    /** 処理実行. **/
    private static final void execution( Map<String,MimdbDatabase> map,String sv,OutputStream out,byte[] buf )
        throws Exception {
        try {
            // コマンド情報を整形.
            sv = sv.trim() ;
            if( sv.endsWith( ";" ) ) {
                sv = sv.substring( 0,sv.length()-1 ).trim() ;
            }
            String lw = sv.trim().toLowerCase() ;
            String[] lcmd = analysisCmd( lw ) ;
            String top = lcmd[ 0 ] ;
            MimdbDatabase emb = null ;
            if( lw.equals( "?" ) ) {
                StringBuilder b = new StringBuilder() ;
                b.append( "[database]" ).append( "  データベース名一覧" ).append( "\n" ) ;
                b.append( "[tables] $(databaseName)" ).append( "  テーブル一覧" ).append( "\n" ) ;
                b.append( "[length] $(databaseName)" ).append( "  テーブル数" ).append( "\n" ) ;
                b.append( "[reload] $(databaseName)" ).append( "  database再読み込み" ).append( "\n" ) ;
                b.append( "[id      $(databaseName) $(tableName)]" ).append( "  テーブル更新ID取得" ).append( "\n" ) ;
                b.append( "[update  $(databaseName) $(tableName) or *]" ).append( "  テーブル更新" ).append( "\n" ) ;
                // 登録テーブルリスト.
                sendString( b.toString(),out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( top.equals( "database" ) ) {
                // 登録データベース一覧を取得.
                int len = map.size() ;
                String[] v = new String[ len ] ;
                int c = 0 ;
                Iterator<String> it = map.keySet().iterator() ;
                while( it.hasNext() ) {
                    if( ( c & 1 ) == 1 ) {
                        v[ c ++ ] = "    " + it.next() ;
                    }
                    else {
                        v[ c ++ ] = it.next() ;
                    }
                }
                sendString( v,out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( top.equals( "tables" ) ) {
                // コマンドが足りない場合.
                if( lcmd.length <= 1 ) {
                    sendString( "tablesコマンドは、データベース名が必要です",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // データベース情報を取得.
                emb = map.get( lcmd[ 1 ] ) ;
                if( emb == null ) {
                    sendString( "指定データベース名(" + lcmd[ 1 ] + ")は存在しません",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // 登録テーブル名と、バージョンを表示.
                String[] list = emb.getTables() ;
                int len = list.length ;
                String[] v = new String[ len ] ;
                for( int i = 0 ; i < len ; i ++ ) {
                    v[ i ] = list[ i ] + " " + MimdbTableManager.getInstance().getId( list[ i ] ) ;
                }
                // 登録テーブルリスト.
                sendString( v,out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( top.equals( "length" ) ) {
                // コマンドが足りない場合.
                if( lcmd.length <= 1 ) {
                    sendString( "lengthコマンドは、データベース名が必要です",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // データベース情報を取得.
                emb = map.get( lcmd[ 1 ] ) ;
                if( emb == null ) {
                    sendString( "指定データベース名(" + lcmd[ 1 ] + ")は存在しません",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // 登録テーブル件数.
                sendString( String.valueOf( emb.length() ),out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( top.equals( "reload" ) ) {
                // コマンドが足りない場合.
                if( lcmd.length <= 1 ) {
                    sendString( "reloadコマンドは、データベース名が必要です",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // データベース情報を取得.
                emb = map.get( lcmd[ 1 ] ) ;
                if( emb == null ) {
                    sendString( "指定データベース名(" + lcmd[ 1 ] + ")は存在しません",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // MimdbEmbedded全体の再ロード.
                emb.loadAll() ;
                sendString( "true",out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( top.equals( "id" ) ) {
                // コマンドが足りない場合.
                if( lcmd.length <= 2 ) {
                    sendString( "idコマンドは、データベース名、テーブル名が必要です",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // データベース情報を取得.
                emb = map.get( lcmd[ 1 ] ) ;
                if( emb == null ) {
                    sendString( "指定データベース名(" + lcmd[ 1 ] + ")は存在しません",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // updateテーブルID.
                if( emb.isTable( lcmd[ 2 ] ) ) {
                    sendString( String.valueOf( MimdbTableManager.getInstance().getId( lcmd[ 2 ] ) ),out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                }
                else {
                    sendString( "-1",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                }
            }
            else if( top.equals( "update" ) ) {
                // コマンドが足りない場合.
                if( lcmd.length <= 2 ) {
                    sendString( "updateコマンドは、データベース名、テーブル名が必要です",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // データベース情報を取得.
                emb = map.get( lcmd[ 1 ] ) ;
                if( emb == null ) {
                    sendString( "指定データベース名(" + lcmd[ 1 ] + ")は存在しません",out,buf ) ;
                    endByte( out ) ;
                    out.flush() ;
                    return ;
                }
                // テーブル更新処理系.
                sendString( updateTable( emb,lcmd[ 2 ].trim() ),out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
            else if( lw.indexOf( " from " ) != -1 ) {
                // sql実行.
                executionSQL( sv,out,buf ) ;
                out.flush() ;
            }
            else {
                sendString( "[" + sv + "]のコマンドは対応していません",out,buf ) ;
                endByte( out ) ;
                out.flush() ;
            }
        } catch( Throwable t ) {
            // エラー.
            sendString( t.toString(),out,buf ) ;
            endByte( out ) ;
            out.flush() ;
        }
    }
    
    /** 終端バイト長. **/
    private static final byte[] NOT_BYTE = new byte[]{ (byte)0,(byte)0 } ;
    
    /** 終端バイト. **/
    private static final void endByte( OutputStream os ) throws Exception {
        os.write( NOT_BYTE,0,2 ) ;
    }
    
    /** 単純文字列送信. **/
    private static final void sendString( Object out,OutputStream os,byte[] buf ) throws Exception {
        String s ;
        if( out instanceof String[] ) {
            String[] n = (String[])out ;
            int len = n.length ;
            StringBuilder b = new StringBuilder() ;
            for( int i = 0 ; i < len ; i ++ ) {
                if( i != 0 ) {
                    b.append( "\n" ) ;
                }
                b.append( n[ i ] ) ;
            }
            s = b.toString() ;
        }
        else {
            s = (String)out ;
        }
        byte[] res = s.getBytes( CHARSET ) ;
        int len = res.length ;
        if( len + 2 >= MAX_READ ) {
            len = MAX_READ-2 ;
        }
        // ヘッダセット.
        buf[ 0 ] = (byte)(len & 0x000000ff) ;
        buf[ 1 ] = (byte)(( len & 0x0000ff00 ) >> 8) ;
        System.arraycopy( res,0,buf,2,len ) ;
        res = null ;
        os.write( buf,0,len+2 ) ;
    }
    
    /** テーブルアップデート処理. **/
    private static final String updateTable( MimdbDatabase emb,String table ) throws Exception {
        if( emb.isTable( table ) ) {
            if( emb.reload( table ) ) {
                return "true" ;
            }
            return "false" ;
        }
        else if( "*".equals( table ) ) {
            emb.reloadAll() ;
            return "true" ;
        }
        return "false" ;
    }
    
    /** SQL実行. **/
    private static final void executionSQL( String sql,OutputStream out,byte[] b ) throws Exception {
        String s ;
        // SQL実行.
        MimdbQuery stmt = MimdbQuery.create( sql ) ;
        MimdbResult res = stmt.executeQuery() ;
        
        // メタデータを出力.
        MimdbMetaData meta = res.getMetaData() ;
        int len = meta.getColumnSize() ;
        StringBuilder buf = new StringBuilder() ;
        for( int i = 0 ; i < len ; i ++ ) {
            if( i != 0 ) {
                buf.append( "| " ) ;
            }
            buf.append( meta.getColumnName( i ) ) ;
        }
        s = buf.append( "\n" ).toString() ;
        buf = null ;
        sendString( s,out,b ) ;
        
        // resultSet条件を文字描画.
        int j ;
        MimdbResultRow r ;
        int resLen = res.length() ;
        for( int i = 0 ; i < resLen ; i ++ ) {
            buf = new StringBuilder() ;
            if( i != 0 ) {
                buf.append( "\n" ) ;
            }
            r = res.get( i ) ;
            for( j = 0 ; j < len ; j ++ ) {
                if( j != 0 ) {
                    buf.append( "| " ) ;
                }
                buf.append( r.get( j ) ) ;
            }
            s = buf.toString() ;
            buf = null ;
            sendString( s,out,b ) ;
        }
        // 合計件数を返却.
        sendString( "\n" + resLen + "件",out,b ) ;
        endByte( out ) ;
    }
    
    /** コマンド分解. **/
    private static final String[] analysisCmd( String s ) {
        char c ;
        int cote = -1 ;
        int len = s.length() ;
        StringBuilder buf = new StringBuilder() ;
        List<String> cmd = new ArrayList<String>() ;
        
        for( int i = 0 ; i < len ; i ++ ) {
            c = s.charAt( i ) ;
            if( cote != -1 ) {
                if( c == cote ) {
                    cmd.add( buf.toString().substring( 1,buf.length() ) ) ;
                    buf = new StringBuilder() ;
                    cote = -1 ;
                }
                else {
                    buf.append( c ) ;
                }
            }
            else if( c == '\'' || c == '\"' ) {
                cote = c ;
                buf.append( c ) ;
            }
            else if( c == ' ' || c == '\t' ) {
                if( buf.length() > 0 ) {
                    cmd.add( buf.toString().trim() ) ;
                    buf = new StringBuilder() ;
                }
            }
            else {
                buf.append( c ) ;
            }
        }
        if( buf.length() > 0 ) {
            cmd.add( buf.toString().trim() ) ;
        }
        String[] ret = new String[ cmd.size() ] ;
        for( int i = 0 ; i < cmd.size() ; i ++ ) {
            ret[ i ] = cmd.get( i ) ;
        }
        return ret ;
    }
}
