package org.maachang.comet.net.nio ;

import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;

/**
 * Nio用InputStream.
 * 
 * @version 2008/05/28
 * @author  masahito suzuki
 * @since   MaachangComet 1.1B
 */
class NioInputStream extends InputStream {
    
    private static final long TIMEOUT = 5000L ;
    private static final long MIN_TIMEOUT = 500L ;
    
    private ReceiveBuffer recvBuffer = null ;
    private byte[] buffer = null ;
    private int bufferToPos = 0 ;
    private long timeout = -1L ;
    private final Object sync = new Object() ;
    
    public NioInputStream() {
        
    }
    
    public InputStream create( NioElement element )
        throws Exception {
        return create( element,TIMEOUT ) ;
    }
    
    public InputStream create( NioElement element,long timeout )
        throws Exception {
        if( element == null || element.isUse() == false ) {
            throw new IOException( "コネクションは既に破棄されています" ) ;
        }
        synchronized( sync ) {
            this.resetObject() ;
            this.recvBuffer = element.getBuffer() ;
            this.timeout = ( timeout <= MIN_TIMEOUT ) ? MIN_TIMEOUT : timeout ;
            this.buffer = null ;
            this.bufferToPos = 0 ;
        }
        return this ;
    }
    
    protected void finalize() throws Exception {
        try {
            close() ;
        } catch( Exception e ) {
        }
    }
    
    protected void resetObject() {
        synchronized( sync ) {
            // pipeline対応.
            // データが存在しない場合は、リセット.
            if( recvBuffer != null &&
                recvBuffer.length() <= 0 ) {
                recvBuffer.reset() ;
            }
            recvBuffer = null ;
            buffer = null ;
            bufferToPos = 0 ;
        }
    }
    
    public void close() {
        resetObject() ;
    }
    
    public int available() throws IOException {
        if( isUse() == false ) {
            throw new IOException( "オブジェクトは既にクローズしています" ) ;
        }
        int ret ;
        synchronized( sync ) {
            ret = recvBuffer.length() ;
        }
        return ret ;
    }
    
    public int read() throws IOException {
        if( isUse() == false ) {
            throw new IOException( "オブジェクトは既にクローズしています" ) ;
        }
        ReceiveBuffer bm ;
        byte[] b = null ;
        int bPos = 0 ;
        synchronized( sync ) {
            bm = recvBuffer ;
            b = buffer ;
            bPos = bufferToPos ;
        }
        if( b == null ) {
            long time = System.currentTimeMillis() ;
            for( ;; ) {
                if( isUse() == false ) {
                    throw new IOException( "オブジェクトは既にクローズしています" ) ;
                }
                if( ( b = bm.get() ) != null ) {
                    synchronized( sync ) {
                        buffer = b ;
                        bufferToPos = 0 ;
                        bPos = 0 ;
                    }
                    break ;
                }
                if( time + timeout <= System.currentTimeMillis() ) {
                    throw new SocketTimeoutException( "SocketTimeout" ) ;
                }
                try { Thread.sleep( 1L ) ; } catch( Exception e ) {}
            }
        }
        int ret = ( ( int )b[ bPos ] & 0x000000ff ) ;
        if( bPos + 1 >= b.length ) {
            synchronized( sync ) {
                buffer = null ;
                bufferToPos = 0 ;
            }
        }
        else {
            synchronized( sync ) {
                bufferToPos ++ ;
            }
        }
        return ret ;
    }
    
    public synchronized boolean markSupported() {
        return false ;
    }
    
    private synchronized boolean isUse() {
        boolean ret ;
        synchronized( sync ) {
            ret = ( recvBuffer != null && recvBuffer.isUse() ) ;
        }
        return ret ;
    }
}
