/*
 * 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.grizzly.asyncweb;

import com.sun.enterprise.web.connector.grizzly.ByteBufferInputStream;
import com.sun.enterprise.web.connector.grizzly.Handler;
import com.sun.enterprise.web.connector.grizzly.OutputWriter;
import com.sun.enterprise.web.connector.grizzly.ProcessorTask;
import com.sun.enterprise.web.connector.grizzly.SelectorFactory;
import com.sun.enterprise.web.connector.grizzly.SelectorThread;
import com.sun.enterprise.web.connector.grizzly.TaskBase;
import com.sun.enterprise.web.connector.grizzly.TaskContext;
import com.sun.enterprise.web.connector.grizzly.TaskEvent;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.util.HashMap;
import java.util.Set;
import java.util.logging.Level;
import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.SessionConfig;
import org.apache.mina.common.TransportType;
import org.apache.mina.io.IoFilterChain;
import org.apache.mina.io.IoHandler;
import org.apache.mina.io.IoSession;
import org.safehaus.asyncweb.transport.nio.HttpIOHandler;

/**
 * <code>ProcessorTask</code> that delegates the request/response handling
 * to AsyncWeb's <code>HttpIoHandler</code>.
 *
 * @author Jeanfrancois Arcand
 */
public class AsyncWebProcessorTask extends TaskBase implements ProcessorTask{


    /**
     * The <code>taskContext</code> associated with this object. The 
     * <code>taskContext</code> contains information about the current
     * connection.
     */
    protected TaskContext taskContext;    
    
    
    /**
     * The <code>TaskEvent</code> associated with this task.
     */
    protected TaskEvent<TaskContext> taskEvent;
    
    /**
     * Not used right since AsyncWeb has its own config file
     */
    private int requestBufferSize;
    private boolean dropConnection = false;
    private Handler handler;
    private int maxHttpHeaderSize = 4096;
    private int timeouts;
    
    
    /**
     * Keep-alive flag.
     */
    private boolean keepAlive = true;
    
    
    /**
     * Has erros occured?
     */
    private boolean error = false;
    
    
    /**
     * The Mina IoSession used to bridge AsyncWeb with Grizzly.
     */
    private MinaIoSession ioSession;
    
    
    /**
     * Mina <code>ByteBuffer</code> implementation.
     */
    private ByteBuffer minaByteBuffer;
    
    
    /**
     * Read as much bytes as we can before invoking AsyncWeb to avoid
     * attaching to the SelectionKey.attach().
     */
    private boolean forcedRead = true;
    
    
    /**
     * The AsyncWeb entry point to handle request.
     */
    private HttpIOHandler httpIOHandler;
    
    
    public AsyncWebProcessorTask() {
        super();
    }
    
    
    /**
     * Used when Grizzly ARP is enabled. 
     * XXX Not Yet supported.
     */
    public void doTask() throws IOException {
        try {
            process(taskContext.getInputStream(),
                    taskContext.getOutputStream());
        } catch(Throwable ex){
            SelectorThread.logger().log(Level.FINE,
                    "processorTask.errorProcessingRequest", ex);
        } finally {
            terminateProcess();        
        }        
    }
    
    
    /**
     * Receive notifications from <code>Task</code>
     */
    public void taskEvent(TaskEvent event){
        if ( event.getStatus() == TaskEvent.START) {
            taskContext = (TaskContext)event.attachement();
            if (  taskEvent == null ) {
                taskEvent = new TaskEvent<TaskContext>();
            }
            
            taskEvent.attach(taskContext);
        }
    }
    
    
    /**
     * Initialize the Mina session.
     */
    public void initialize() {
        ioSession = new MinaIoSession();
    }

 
    /**
     * Delegate the request to <code>AsyncWeb</code> by simulating 
     */
    public boolean process(InputStream input, OutputStream output) 
        throws Exception {
          
        ByteBufferInputStream bbInputStream =
                (ByteBufferInputStream)input;
        java.nio.ByteBuffer byteBuffer = bbInputStream.getByteBuffer();
        
        if ( forcedRead ) {
            readAllBytes(byteBuffer);
        }
        byteBuffer.flip();
        
        httpIOHandler.sessionOpened(ioSession);    
        
        // We MUST avoid creating a wrapper on every request, but recycle the 
        // one used the previous transaction.
        minaByteBuffer = ByteBuffer.wrap(byteBuffer);
        httpIOHandler.dataRead(ioSession,minaByteBuffer);

        return keepAlive;
    }

    
    /**
     * Use the temporary <code>Selector</code> to try to loads as much as we
     * can available bytes before delegating the request processing to 
     * <code>AsyncWeb</code>
     */
    private void readAllBytes(java.nio.ByteBuffer byteBuffer) throws IOException{
        int count = 1;
        int byteRead = 0;
        Selector readSelector = null;
        SelectionKey tmpKey = null;

        try{
            SocketChannel socketChannel = (SocketChannel)key.channel();
            while (count > 0){
                count = socketChannel.read(byteBuffer);
                if ( count > 0 )
                    byteRead += count;
            }            

            if ( byteRead == 0 ){
                readSelector = SelectorFactory.getSelector();

                if ( readSelector == null ){
                    return;
                }
                count = 1;
                tmpKey = socketChannel
                        .register(readSelector,SelectionKey.OP_READ);               
                tmpKey.interestOps(tmpKey.interestOps() | SelectionKey.OP_READ);
                int code = readSelector.selectNow();
                tmpKey.interestOps(
                    tmpKey.interestOps() & (~SelectionKey.OP_READ));

                if ( code == 0 ){
                    return;
                }

                while (count > 0){
                    count = socketChannel.read(byteBuffer);
                    if ( count > 0 )
                        byteRead += count;                 
                }
            }
        } finally {
            if (tmpKey != null)
                tmpKey.cancel();

            if ( readSelector != null){
                // Bug 6403933
                try{
                    readSelector.selectNow();
                } catch (IOException ex){
                    ;
                }
                SelectorFactory.returnSelector(readSelector);
            }
        }
    }
    
    
    /**
     * Set the <code>HttpIoHandler</code>.
     */
    public void setIoHandler(HttpIOHandler httpIOHandler){
        this.httpIOHandler = httpIOHandler;
    }
    
    
    /**
     * Stop keep-aliving the connection.
     */
    public void terminateProcess() {
        keepAlive = false;
    }

    
    /**
     * Always called when the connection is closed.
     */
    public void recycle(){
        super.recycle();
        keepAlive = true;        
        
        httpIOHandler.sessionClosed(ioSession);
        ioSession.recycle();
    }
    
    // ----------------------------------------------- Mina IoSession Hook ---/
    
    
    /**
     * Since AsyncWeb is based on MINA, delegate all Mina calls to Grizzly under
     * the hood.
     */
    private class MinaIoSession implements IoSession{
        private HashMap<String,Object> attributes;
        private Object attachment;
        
        public MinaIoSession(){
            attributes = new HashMap<String,Object>();
        }
        
        public IoFilterChain getFilterChain() {
            return null;
        }

        public void write(ByteBuffer byteBuffer, Object object) {
            try{
                // The Object is a marker used by Mina when 
                // writting asynchronously. We might want to re-introduce
                // the WriteTask in Grizzly to simulate the non-blocking 
                // operation.
                
                OutputWriter.flushChannel(
                        (SocketChannel)key.channel(),byteBuffer.buf());

                // Tell AsyncWeb we are done.
                httpIOHandler.dataWritten(this,object);
            } catch (IOException ex){
                SelectorThread.logger().log(Level.WARNING,"IoSession.write",ex);
            }
        }

        public void close() {
            terminateProcess();
        }

        public void close(boolean b) {
            terminateProcess();        
        }

        public Object getAttachment() {
            return attachment;
        }

        public Object setAttachment(Object attachment) {
            Object oldAttachment = this.attachment;
            this.attachment = attachment;
            return oldAttachment;
        }

        public Object getAttribute(String key) {
            return attributes.get(key);
        }

        public Object setAttribute(String key, Object value) {
            return attributes.put(key,value);
        }

        public Object removeAttribute(String key) {
            return attributes.remove(key);
        }

        public Set getAttributeKeys() {
            return attributes.keySet();
        }

        public TransportType getTransportType() {
            return TransportType.SOCKET;
        }

        public boolean isConnected() {
            if (key == null) return false;
            return ((SocketChannel)key.channel()).isConnected();
        }

        public SessionConfig getConfig() {
            return null;
        }

        public SocketAddress getRemoteAddress() {
            if (key == null) return null;
            return ((SocketChannel)key.channel()).socket().getRemoteSocketAddress();
        }

        public SocketAddress getLocalAddress() {
            if (key == null) return null;
            return ((SocketChannel)key.channel()).socket().getLocalSocketAddress();              
        }

        public long getReadBytes() {
            return -1;
        }

        public long getWrittenBytes() {
            return -1;
        }

        public long getWrittenWriteRequests() {
            return -1;        
        }

        public int getScheduledWriteRequests() {
            return -1;       
        }

        public long getCreationTime() {
            return -1;        
        }

        public long getLastIoTime() {
            return -1;        
        }

        public long getLastReadTime() {
            return -1;        
        }

        public long getLastWriteTime() {
            return -1;        
        }

        public boolean isIdle(IdleStatus idleStatus) {
            return false;
        }

        public int getIdleCount(IdleStatus idleStatus) {
            return -1;
        }

        public long getLastIdleTime(IdleStatus idleStatus) {
            return -1;
        }

        public IoHandler getHandler() {
            return null;
        }
        
        public void recycle(){
            attributes.clear();
            attachment = null;
        }
    }
    
       
    // ------------------------------------------------------ Setter/Getter ---/
    
    
    public void setBufferSize(int requestBufferSize) {
        this.requestBufferSize= requestBufferSize;
    }

    
    public void setDropConnection(boolean dropConnection) {
        this.dropConnection = dropConnection;
    }

    
    public void setHandler(Handler handler) {
        this.handler = handler;
    }

    
    public Handler getHandler() {
        return handler;
    }

    
    public void setMaxHttpHeaderSize(int maxHttpHeaderSize) {
        this.maxHttpHeaderSize = maxHttpHeaderSize;
    }

    public void setTimeout(int timeouts) {
        this.timeouts = timeouts;
    }

    public boolean isKeepAlive() {
        return keepAlive;
    }

    
    public boolean isError() {
        return error;
    }

    
    public int getBufferSize() {
        return requestBufferSize;
    }

    public boolean getDropConnection() {
        return dropConnection;
    }

    public void setSocket(Socket socket) {
    }
    
    // ----------------------------------------- Not used -----------------//

    public int getMaxPostSize() {
         throw new UnsupportedOperationException();
    }  
    
    
    public void setMaxPostSize(int mps) {
         throw new UnsupportedOperationException();
    }

    public String getRequestURI() {
        throw new UnsupportedOperationException();
    }

    
    public long getWorkerThreadID() {
        throw new UnsupportedOperationException();
    }

    
    public void invokeAdapter() {
         throw new UnsupportedOperationException();
    }

    public void parseRequest() throws Exception {
         throw new UnsupportedOperationException();
    }

    public boolean parseRequest(InputStream input, OutputStream output, 
            boolean keptAlive) throws Exception {
         throw new UnsupportedOperationException();
    }

    public void postProcess() throws Exception {
         throw new UnsupportedOperationException();
    }

    public void postProcess(InputStream input, OutputStream output) 
        throws Exception {
         throw new UnsupportedOperationException();
    }

    public void postResponse() throws Exception {
         throw new UnsupportedOperationException();
    }

    public void preProcess() throws Exception {
         throw new UnsupportedOperationException();
    }

    public void preProcess(InputStream input, OutputStream output) 
        throws Exception {
         throw new UnsupportedOperationException();
    }    

    public boolean parseRequest(
            AbstractSelectableChannel abstractSelectableChannel, boolean b) 
                throws Exception {
        return false;
    }

    public void postProcess(AbstractSelectableChannel abstractSelectableChannel) throws Exception {
    }

    public void preProcess(AbstractSelectableChannel abstractSelectableChannel) throws Exception {
    }

    public boolean process(AbstractSelectableChannel abstractSelectableChannel)
        throws Exception {
        return false;
    }

}
