/*
 * The contents of this file are subject to the terms 
 * of the Common Development and Distribution License 
 * (the License).  You may not use this file except in
 * compliance with the License.
 * 
 * You can obtain a copy of the license at 
 * https://glassfish.dev.java.net/public/CDDLv1.0.html or
 * glassfish/bootstrap/legal/CDDLv1.0.txt.
 * See the License for the specific language governing 
 * permissions and limitations under the License.
 * 
 * When distributing Covered Code, include this CDDL 
 * Header Notice in each file and include the License file 
 * at glassfish/bootstrap/legal/CDDLv1.0.txt.  
 * If applicable, add the following below the CDDL Header, 
 * with the fields enclosed by brackets [] replaced by
 * you own identifying information: 
 * "Portions Copyrighted [year] [name of copyright owner]"
 * 
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 */
package com.sun.enterprise.web.connector.grizzly;

import java.io.IOException;

import java.net.Socket;

import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

import java.util.logging.Level;

/**
 * Read available data on a non blocking <code>SocketChannel</code>. 
 * <code>StreamAlgorithm</code> stategy will decide if more bytes are required
 * or not. Once the <code>StreamAlgorithm</code> is ready, the 
 * <code>ProcessorTask</code> attached to this class will be executed.
 *
 * @author Scott Oaks
 * @author Jean-Francois Arcand
 */
public class ReadTask extends TaskBase {
    

    /**
     * The <code>TaskContext</code> instance associated with this object.
     * The <code>TaskContext</code> is initialized at startup and then recycled.
     */
    protected TaskContext taskContext;
    
    
    /**
     * The <code>TaskEvent</code> instance used by this object to notify its 
     * listeners
     */
    protected TaskEvent taskEvent;

    
    /**
     * The <code>ByteBuffer</code> used by this task to buffer the request
     * stream.
     */
    protected ByteBuffer byteBuffer;
    
    
    /**
     * The <code>ProcessorTask</code> used by this class.
     */
    protected ProcessorTask processorTask;
  
    
    /**
     * Max post size.
     */
    protected int maxPostSize = 25 * 1024 * 1024;
     
    
    /**
     * The recycled <code>OutputStream</code> used by this buffer.
     */
    protected ByteBufferInputStream inputStream;


    /**
     * The <code>Algorithm</code> used to parse the request and determine
     * of the bytes has been all read from the <code>SocketChannel</code>
     */
    protected StreamAlgorithm algorithm;
    
    
    /**
     * <code>true</code> only when another object has already read bytes
     * from the channel.
     */
    protected boolean bytesAvailable = false;

    
    /**
     * Has the task been returned to the pool
     */
    protected volatile boolean isReturned = false;


    // ----------------------------------------------------- Constructor ----/

    
    public ReadTask(){
        ;//
    }
    
    
    public ReadTask(StreamAlgorithm algorithm,
                    boolean useDirectByteBuffer, boolean useByteBufferView){
        type = READ_TASK;    
        this.algorithm = algorithm;       
        byteBuffer = algorithm.allocate(useDirectByteBuffer,useByteBufferView);
        inputStream = new ByteBufferInputStream();
    }
    
    
    /**
     * Force this task to always use the same <code>ProcessorTask</code> instance.
     */
    public void attachProcessor(ProcessorTask processorTask){
        this.processorTask = processorTask;        
        configureProcessorTask();
    }  
    
    
    /**
     * Set appropriate attribute on the <code>ProcessorTask</code>.
     */
    public void configureProcessorTask(){
        // Disable blocking keep-alive mechanism. Keep-Alive mechanism
        // will be managed by this class instead.
        processorTask.useAlternateKeepAlive(true);
        processorTask.setSelectionKey(key);
        processorTask.setSocket(((SocketChannel)key.channel()).socket());
        processorTask.setHandler(algorithm.getHandler());

    }  
    
      
    /**
     * Return the <code>ProcessorTask</code> to the pool.
     */
    public void detachProcessor(){
        if (processorTask != null){
            processorTask.recycle();           
        }
        
        // Notify listeners
        if ( listeners != null ) {
            for (int i=listeners.size()-1; i > -1; i--){
                if ( taskEvent == null ) {
                    taskEvent = new TaskEvent<ReadTask>();
                }
                taskEvent.attach(this);            
                taskEvent.setStatus(TaskEvent.COMPLETED);           
                listeners.get(i).taskEvent(taskEvent);
            }                
            clearTaskListeners();
        }
        
        if (recycle && processorTask != null){
            selectorThread.returnTask(processorTask);
            processorTask = null;
        }
    }
    
    
    /**
     * Read data from the socket and process it using this thread, and only if 
     * the <code>StreamAlgorith</code> stategy determine no more bytes are 
     * are needed.
     */
    public void doTask() throws IOException {
        int count = 0;
        Socket socket = null;
        SocketChannel socketChannel = null;
        boolean keepAlive = false;
        Exception exception = null;
        isReturned = false;
        
        try {
            socketChannel = (SocketChannel)key.channel();
            socket = socketChannel.socket();
            algorithm.setSocketChannel(socketChannel);
           
            int loop = 0;
            int bufferSize = 0;
            while ( socketChannel.isOpen() && (bytesAvailable || 
                    ((count = socketChannel.read(byteBuffer))> -1))){

                // Avoid calling the Selector.
                if ( count == 0 && !bytesAvailable){
                    loop++;
                    if (loop > 2){
                        break;
                    }
                    continue;
                }
                bytesAvailable = false;
                
                byteBuffer = algorithm.preParse(byteBuffer);
                inputStream.setByteBuffer(byteBuffer);
                inputStream.setSelectionKey(key);
                
                // try to predict which HTTP method we are processing
                if ( algorithm.parse(byteBuffer) ){ 
                    keepAlive = executeProcessorTask();
                    if (!keepAlive) {
                        break;
                    }
                } else {
                    // We must call the Selector since we don't have all the 
                    // bytes
                    keepAlive = true;
                }
            } 
        // Catch IO AND NIO exception
        } catch (IOException ex) {
            exception = ex;
        } catch (RuntimeException ex) {
            exception = ex;      
        } finally {                   
            manageKeepAlive(keepAlive,count,exception);
        }
    }


    /**
     * Evaluate if the <code>SelectionKey</code> needs to be registered to 
     * the main <code>Selector</code>
     */
    protected void manageKeepAlive(boolean keepAlive,int count, 
            Exception exception){         

        // The key is invalid when the Task has been cancelled.
        if ( count == -1 || !key.isValid() || exception != null ){
            keepAlive = false;
            
            if ( exception != null){
                // Make sure we have detached the processorTask
                detachProcessor();
                SelectorThread.logger().
                  log(Level.FINEST, "SocketChannel Read Exception: ",exception);
            }
        }

        if (keepAlive) {    
            registerKey(); 
        } 
            
        terminate(keepAlive);
    }
 
    
    /**
     * Execute the <code>ProcessorTask</code> only if the request has
     * been fully read. Guest the size of the request by using the 
     * content-type HTTP headers.
     * @return false if the request wasn't fully read by the channel.
     *         so we need to respin the key on the Selector.
     */
    public boolean executeProcessorTask() throws IOException{                  
        boolean registerKey = false;
        
        if (SelectorThread.logger().isLoggable(Level.FINEST))
            SelectorThread.logger().log(Level.FINEST,"executeProcessorTask");
        
        if ( algorithm.getHandler()
                .handle(null, Handler.REQUEST_BUFFERED) == Handler.BREAK ){
            return true;
        }
        
        // Get a processor task. If the processorTask != null, that means we
        // failed to load all the bytes in a single channel.read().
        if (processorTask == null){
            attachProcessor(selectorThread.getProcessorTask());
        }
        
        try {           
            // The socket might not have been read entirely and the parsing
            // will fail, so we need to respin another event.
            registerKey = processorTask.process(inputStream,null);           
        } catch (Exception e) {
            SelectorThread.logger()
                .log(Level.SEVERE,"readTask.processException", e);
        } 
        detachProcessor();
        return registerKey;
    }

    
    /**
     * Return this object to the pool
     */
    protected void returnTask(){
        if ( recycle && !isReturned ) {
            isReturned = true;
            selectorThread.returnTask(this);
        } 
    }
    

    public void taskEvent(TaskEvent event){
        if (event.getStatus() == TaskEvent.COMPLETED){
            terminate(false);
        } 
    }
    

    /**
     * Complete the processing.
     */
    public void terminate(boolean keepAlive){          
        // Safeguard to avoid returning the instance more than once.
        if ( isReturned ){
            return;
        }
        
        if ( !keepAlive ){
            finishConnection();
        }
        recycle(); 
        returnTask();
    }
    
    
    /**
     * Clear the current state and make this object ready for another request.
     */
    public void recycle(){
        byteBuffer = algorithm.postParse(byteBuffer);   

        byteBuffer.clear(); 
        inputStream.recycle();
        algorithm.recycle();
        key = null;
        inputStream.setSelectionKey(null);                             
    }
    
    // -------------------------------------------------------- TaskEvent ---// 

       
    /**
     * Cancel the <code>SelectionKey</code> and close its underlying 
     * <code>SocketChannel</code>. Add this <code>Task</code> to the Keep-Alive
     * sub-system.
     */
    protected void finishConnection(){
        
        if (SelectorThread.logger().isLoggable(Level.FINEST))
            SelectorThread.logger().log(Level.FINEST,"finishConnection"); 
        
        try{
            if (taskContext != null){
                taskContext.recycle();
            }
        } catch (IOException ioe){
            ;
        }

        selectorThread.cancelKey(key);
    }

    
    /**
     * Register the <code>SelectionKey</code> with the <code>Selector</code>
     */
    protected void registerKey(){
        if (key.isValid()){            
            if (SelectorThread.logger().isLoggable(Level.FINEST))
                SelectorThread.logger().log(Level.FINEST,"registerKey");           

            selectorThread.registerKey(key);
        } else {
            terminate(false);
        }
    }
    
    
    // -------------------------------------------------------- getter/setter--/
    

    /**
     * Return the associated <code>ProcessorTask</code>.
     * @return the associated <code>ProcessorTask</code>, null if not used.
     */
    public ProcessorTask getProcessorTask(){
        return processorTask;
    }
    
    
    /**
     * Set the algorithm used by this instance to predict the end of the NIO 
     * Stream.
     */
    public void setStreamAlgorithm(StreamAlgorithm algorithm){
        this.algorithm = algorithm;
    }
    
    
    /**
     * Return the underlying <code>ByteBuffer</code> used by this class.
     */
    public ByteBuffer getByteBuffer(){
        return byteBuffer;
    }
    
    
    /**
     * If the attached byteBuffer was already filled, tell the
     * Algorithm to re-use the bytes.
     */ 
    public void setBytesAvailable(boolean bytesAvailable){
        this.bytesAvailable = bytesAvailable;
    }

}
