package org.maachang.bdb ;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel.MapMode;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.locks.Lock;

/**
 * Berkeley DB Queue実装.
 *
 * @version 2010/01/20
 * @author  masahito suzuki
 * @since   bdb 1.0.4
 */
class BQueueImpl implements BQueue {
    
    /**
     * キュー生成ロック.
     */
    public static final Object QUEUE_LOCK = new Object() ;
    
    /**
     * BDB.
     */
    private BdbImpl bdb = null ;
    
    /**
     * MQヘッダ.
     */
    private BMqHeader header = null ;
    
    /**
     * 最大格納数.
     */
    private long max = -1L ;
    
    /**
     * 警告値.
     */
    private double warning = 0.0f ;
    private long warnLen ;
    
    /**
     * キュー名.
     */
    private String name = null ;
    
    /**
     * ロックオブジェクト.
     */
    private Lock rsync = null ;
    private Lock wsync = null ;
    
    /**
     * クローズフラグ.
     */
    private final AtomicBL closeFlag = new AtomicBL( true ) ;
    
    /**
     * コンストラクタ.
     */
    private BQueueImpl() {
        
    }
    
    /**
     * コンストラクタ.
     * @param manager 対象のBdbマネージャを設定します.
     * @param name BQueue用にオープンするBdb名を設定します.
     * @exception Exception 例外.
     */
    protected BQueueImpl( BdbManager manager,String name ) throws Exception {
        this( manager,name,(long)Integer.MAX_VALUE,0.75d ) ;
    }
    
    /**
     * コンストラクタ.
     * @param manager 対象のBdbマネージャを設定します.
     * @param name BQueue用にオープンするBdb名を設定します.
     * @param max キュー格納最大数を設定します.
     * @param warning 警告値を設定します.<br>たとえば、75％の場合は、0.75dで設定します.
     * @exception Exception 例外.
     */
    protected BQueueImpl( BdbManager manager,String name,long max,double warning ) throws Exception {
        if( manager == null || manager.isDestroy() ||
            name == null || ( name = name.trim() ).length() <= 0 ||
            max < 128 || warning < 0.25d ) {
            if( manager == null || manager.isDestroy() ) {
                throw new IllegalArgumentException( "不正なbdbマネージャが設定されています" ) ;
            }
            if( name == null || ( name = name.trim() ).length() <= 0 ) {
                throw new IllegalArgumentException( "BQueue用にオープンするBdb名が設定されていません" ) ;
            }
            if( max < 128L ) {
                throw new IllegalArgumentException( "最大格納数は、128以上を設定してください" ) ;
            }
            if( warning < 0.25d ) {
                throw new IllegalArgumentException( "警告値は、25%以上を設定してください" ) ;
            }
        }
        BMqHeader header = null ;
        synchronized( QUEUE_LOCK ) {
            header = new BMqHeader( manager.getDirectory(),name ) ;
        }
        BdbImpl bdb = new BdbImpl( true,name,manager ) ;
        this.bdb = bdb ;
        this.header = header ;
        this.max = max ;
        this.warning = warning ;
        this.warnLen = (long)( (double)max * warning ) ;
        this.name = name ;
        this.rsync = bdb.getRSync() ;
        this.wsync = bdb.getWSync() ;
        this.closeFlag.set( false ) ;
        manager.managerq.put( name,this ) ;
    }
    
    /**
     * デストラクタ.
     */
    protected void finalize() throws Exception {
        close() ;
    }
    
    /**
     * オブジェクトクローズ.
     */
    public void close() {
        wsync.lock() ;
        try {
            if( header != null ) {
                header.close() ;
            }
            if( bdb!= null ) {
                try {
                    bdb.close() ;
                } catch( Exception e ) {
                }
            }
            header = null ;
            bdb = null ;
            closeFlag.set( true ) ;
        } finally {
            wsync.unlock() ;
        }
    }
    
    /**
     * 情報追加.
     * @param keyValue Bdbオブジェクトを制御するKeyValueオブジェクトを設定します.
     * @param value 対象の要素を設定します.
     * @exception Exception 例外.
     */
    public void offer( BdbKeyValue keyValue,byte[] value )
        throws Exception {
        if( keyValue == null ) {
            throw new IllegalArgumentException( "KeyValueオブジェクトは設定されていません" ) ;
        }
        if( value == null || value.length <= 0 ) {
            throw new IllegalArgumentException( "追加要素が設定されていません" ) ;
        }
        wsync.lock() ;
        try {
            BdbImpl b = bdb ;
            if( b == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            long s = header.getSequence() ;
            b.putNoLock( keyValue.create( ConvertParam.convertLong( s ),value ) ) ;
        } finally {
            wsync.unlock() ;
            keyValue.clear() ;
        }
    }
    
    /**
     * 情報取得.
     * <p>この処理を実行した場合、取得された内容は削除されます.</p>
     * @param keyValue Bdbオブジェクトを制御するKeyValueオブジェクトを設定します.
     * @return byte[] 対象の要素が返されます.
     * @exception Exception 例外.
     */
    public byte[] poll( BdbKeyValue keyValue ) throws Exception {
        if( keyValue == null ) {
            throw new IllegalArgumentException( "KeyValueオブジェクトは設定されていません" ) ;
        }
        byte[] ret = null ;
        wsync.lock() ;
        try {
            BdbImpl b = bdb ;
            if( b == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            long first = header.getFirst() ;
            long last = header.getLast() ;
            // 先頭のデータを取得.
            while( !b.getToRemoveNoLock( keyValue.create( ConvertParam.convertLong( first ),null ) ) ) {
                first ++ ;
                if( last <= first ) {
                    return null ;
                }
            }
            header.setFirst( first + 1 ) ;
            ret = keyValue.getValue() ;
            keyValue.setValue( null ) ;
        } finally {
            wsync.unlock() ;
            keyValue.clear() ;
        }
        return ret ;
    }
    
    /**
     * 情報取得.
     * <p>この処理を実行した場合、取得された内容は削除されません.</p>
     * @param keyValue Bdbオブジェクトを制御するKeyValueオブジェクトを設定します.
     * @return byte[] 対象の要素が返されます.
     * @exception Exception 例外.
     */
    public byte[] peek( BdbKeyValue keyValue ) throws Exception {
        if( keyValue == null ) {
            throw new IllegalArgumentException( "KeyValueオブジェクトは設定されていません" ) ;
        }
        byte[] ret = null ;
        wsync.lock() ;
        try {
            BdbImpl b = bdb ;
            if( b == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            long last = header.getLast() ;
            long first = header.getFirst() ;
            // 先頭のデータを取得.
            while( !b.getNoLock( keyValue.create( ConvertParam.convertLong( first ),null ) ) ) {
                first ++ ;
                if( last <= first ) {
                    return null ;
                }
            }
            ret = keyValue.getValue() ;
        } finally {
            wsync.unlock() ;
            keyValue.clear() ;
        }
        return ret ;
    }
    
    /**
     * 情報一覧取得.
     * <p>この処理を実行した場合、取得された内容は削除されません.</p>
     * @return Iterator<byte[]> 格納一覧内容が格納されたIteratorが返されます.
     * @exception Exception 例外.
     */
    public Iterator<byte[]> iterator() throws Exception {
        rsync.lock() ;
        try {
            if( bdb == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            return new BQueueIterator( new BdbKeyValue(),bdb,header,rsync,closeFlag ) ;
        } finally {
            rsync.unlock() ;
        }
    }
    
    /**
     * 格納情報数を取得.
     * @return long 格納情報数が返されます.
     * @exception Exception 例外.
     */
    public long length() throws Exception {
        long ret ;
        rsync.lock() ;
        try {
            if( bdb == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            ret = bdb.sizeNoLock() ;
        } finally {
            rsync.unlock() ;
        }
        return ret ;
    }
    
    /**
     * 最大格納数を取得.
     * @return long 最大格納数が返されます.
     */
    public long getMaxLength() {
        return max ;
    }
    
    /**
     * 警告値を取得.
     * @return double 警告値が返されます.
     */
    public double getWarning() {
        return warning ;
    }
    
    /**
     * キュー名を取得.
     */
    public String getName() {
        return name ;
    }
    
    /**
     * BdbKeyValueオブジェクトを取得.
     * <p>※このオブジェクトは、ThreadLocalから取得されているので、Threadセーフです.</p>
     * @return BdbKeyValue BdbKeyValueオブジェクトが返されます.
     */
    public BdbKeyValue getBdbKeyValue() {
        return BdbManager.getBdbKeyValue() ;
    }
    
    /**
     * 現在の格納数が警告値を越しているかチェック.
     * @return boolean [true]の場合、警告値を越しています.
     * @exception Exception 例外.
     */
    public boolean isWarning() throws Exception {
        boolean ret ;
        rsync.lock() ;
        try {
            if( bdb == null ) {
                throw new IOException( "オブジェクトは既にクローズしています" ) ;
            }
            ret = ( warnLen <= ( bdb.sizeNoLock() ) ) ;
        } finally {
            rsync.unlock() ;
        }
        return ret ;
    }
    
    /**
     * このオブジェクトが既にクローズされているかチェック.
     * @return boolean [true]の場合、既にクローズされています.
     */
    public boolean isClose() {
        return closeFlag.get() ;
    }
    
    /** BQueueヘッダオブジェクト **/
    private static class BMqHeader {
        private static final String ADD_DIR = "/queue/" ;
        private static final String HEADER_PLUS = ".header" ;
        private static final long LENGTH = 8L*2L ;
        private RandomAccessFile fp = null ;
        private MappedByteBuffer buf = null ;
        
        BMqHeader( String dir,String name ) throws Exception {
            File fl = new File( dir+ADD_DIR ) ;
            if( !fl.isDirectory() ) {
                if( !fl.mkdirs() ) {
                    throw new IOException( "ディレクトリの作成に失敗しました:" + dir ) ;
                }
            }
            name = dir + ADD_DIR + name + HEADER_PLUS ;
            close() ;
            RandomAccessFile f = null ;
            MappedByteBuffer m = null ;
            try {
                f = new RandomAccessFile( name,"rw" ) ;
                f.setLength( LENGTH ) ;
                m = f.getChannel().map( MapMode.READ_WRITE,0,LENGTH ) ;
            } catch( Exception e ) {
                if( f != null ) {
                    try {
                        f.close() ;
                    } catch( Exception ee ) {
                    }
                    f = null ;
                    m = null ;
                }
                throw e ;
            }
            fp = f ;
            buf = m ;
        }
        
        void close() {
            if( this.fp != null ) {
                try {
                    this.fp.close() ;
                } catch( Exception e ) {
                }
            }
            this.fp = null ;
            this.buf = null ;
        }
        
        long getSequence() throws Exception {
            long ret = buf.getLong( 0 ) ;
            if( ret >= Long.MAX_VALUE ) {
                throw new IOException( "Long(64bit)のシーケンスIDが枯渇しました" ) ;
            }
            ret ++ ;
            buf.putLong( 0,ret ) ;
            return ret ;
        }
        
        long getLast() throws Exception {
            return buf.getLong( 0 ) ;
        }
        
        void setFirst( long first ) {
            buf.putLong( 8,first ) ;
        }
        
        long getFirst() {
            return buf.getLong( 8 ) ;
        }
    }
    
    /** BQueue用Iterator. **/
    private static class BQueueIterator implements Iterator<byte[]> {
        private BdbImpl bdb = null ;
        private BMqHeader header = null ;
        private BdbKeyValue keyValue = null ;
        private Lock rsync = null ;
        private AtomicBL closeFlag = null ;
        
        private long target = -1L ;
        private boolean flg = false ;
        
        private BQueueIterator() {}
        
        BQueueIterator( BdbKeyValue keyValue,BdbImpl bdb,BMqHeader header,Lock rsync,AtomicBL closeFlag )
            throws Exception {
            this.keyValue = keyValue ;
            this.bdb = bdb ;
            this.header = header ;
            this.rsync = rsync ;
            this.closeFlag = closeFlag ;
        }
        
        protected void finalize() throws Exception {
            close() ;
        }
        
        public void close() {
            bdb = null ;
            header = null ;
            keyValue = null ;
            rsync = null ;
            target = -1L ;
            flg = false ;
        }
        
        public boolean hasNext() {
            if( flg ) {
                return true ;
            }
            long last ;
            try {
                rsync.lock() ;
                try {
                    last = header.getLast() ;
                } finally {
                    rsync.unlock() ;
                }
                if( target > last || closeFlag.get() ) {
                    return false ;
                }
                if( target <= -1L ) {
                    rsync.lock() ;
                    try {
                        target = header.getFirst() ;
                    } finally {
                        rsync.unlock() ;
                    }
                }
                long s ;
                boolean getFlag = false ;
                while( true ) {
                    rsync.lock() ;
                    try {
                        getFlag = bdb.get( keyValue.create( ConvertParam.convertLong( target ),null ) );
                    } finally {
                        rsync.unlock() ;
                    }
                    if( getFlag ) {
                        flg = true ;
                        target ++ ;
                        return true ;
                    }
                    else {
                        rsync.lock() ;
                        try {
                            s = header.getFirst() ;
                        } finally {
                            rsync.unlock() ;
                        }
                        if( s > target ) {
                            target = s ;
                        }
                        else {
                            target ++ ;
                        }
                    }
                    rsync.lock() ;
                    try {
                        last = header.getLast() ;
                    } finally {
                        rsync.unlock() ;
                    }
                    if( target > last || closeFlag.get() ) {
                        return false ;
                    }
                }
            } catch( Exception e ) {
                return false ;
            }
        }
        
        public byte[] next() {
            if( closeFlag.get() ) {
                throw new NoSuchElementException() ;
            }
            else if( flg ) {
                flg = false ;
                return keyValue.getValue() ;
            }
            else if( hasNext() ) {
                flg = false ;
                return keyValue.getValue() ;
            }
            throw new NoSuchElementException() ;
        }
        
        public void remove() {
            throw new UnsupportedOperationException() ;
        }
    }
}
