This commit takes jetty-7 back to the 2.5 servlet API, but leaves much of the

3.0 asynchronous machinery available.

Introduces a new improved Continuation API that is an evolution of the 
original Continuation API, while taking some good ideas from the 3.0 spec.
The intention is that this API will simplify the use of the 3.0
API, while making it available in jetty-7 (scalably) and in all other
2.5 containers with the ContinuationFilter.





git-svn-id: svn+ssh://dev.eclipse.org/svnroot/rt/org.eclipse.jetty/jetty/trunk@27 7e9141cc-0065-0410-87d8-b60c137991c4

6 files changed, 775 insertions, 0 deletions

diff --git a/jetty-continuation/pom.xml b/jetty-continuation/pom.xml
new file mode 100644
index 0000000000..8f6c35ea8f
--- /dev/null
+++ b/jetty-continuation/pom.xml
@@ -0,0 +1,83 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <parent>
+    <groupId>org.eclipse.jetty</groupId>
+    <artifactId>jetty-project</artifactId>
+    <version>7.0.0.incubation0-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+  <artifactId>jetty-continuation</artifactId>
+  <name>Jetty :: Continuation</name>
+  <description>Asynchronous API</description>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.felix</groupId>
+        <artifactId>maven-bundle-plugin</artifactId>
+        <version>1.4.2</version>
+        <extensions>true</extensions>
+        <executions>
+          <execution>
+            <goals>
+              <goal>manifest</goal>
+            </goals>
+            <configuration>
+              <instructions>
+                <Bundle-SymbolicName>org.eclipse.jetty.continuation</Bundle-SymbolicName>
+                <Bundle-RequiredExecutionEnvironment>J2SE-1.5</Bundle-RequiredExecutionEnvironment>
+                <Bundle-DocURL>http://jetty.eclipse.org</Bundle-DocURL>
+                <Import-Package></Import-Package>
+              </instructions>
+            </configuration>
+           </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>artifact-jar</id>
+            <goals>
+              <goal>jar</goal>
+            </goals>
+          </execution>
+          <execution>
+            <id>test-jar</id>
+            <goals>
+              <goal>test-jar</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <archive>               
+            <manifestFile>${project.build.outputDirectory}/META-INF/MANIFEST.MF</manifestFile>
+          </archive>
+        </configuration>
+      </plugin>
+      <!-- always include sources since jetty-xbean makes use of them -->
+       <plugin>
+         <groupId>org.apache.maven.plugins</groupId>
+         <artifactId>maven-source-plugin</artifactId>
+         <executions>
+           <execution>
+             <id>attach-sources</id>
+             <goals>
+               <goal>jar</goal>
+             </goals>
+           </execution>
+         </executions>
+      </plugin>
+    </plugins>
+  </build>
+  <dependencies>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mortbay.jetty</groupId>
+      <artifactId>servlet-api</artifactId>
+    </dependency>
+  </dependencies>
+</project>
diff --git a/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/Continuation.java b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/Continuation.java
new file mode 100644
index 0000000000..f4043ef23b
--- /dev/null
+++ b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/Continuation.java
@@ -0,0 +1,207 @@
+// ========================================================================
+// Copyright (c) 2004-2009 Mort Bay Consulting Pty. Ltd.
+// ------------------------------------------------------------------------
+// All rights reserved. This program and the accompanying materials
+// are made available under the terms of the Eclipse Public License v1.0
+// and Apache License v2.0 which accompanies this distribution.
+// The Eclipse Public License is available at 
+// http://www.eclipse.org/legal/epl-v10.html
+// The Apache License v2.0 is available at
+// http://www.opensource.org/licenses/apache2.0.php
+// You may elect to redistribute this code under either of these licenses. 
+// ========================================================================
+
+package org.eclipse.jetty.continuation;
+
+import java.io.IOException;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.Servlet;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+
+/* ------------------------------------------------------------ */
+/**
+ * Continuation.
+ * 
+ * A continuation is a mechanism by which a HTTP Request can be suspended and
+ * restarted after a timeout or an asynchronous event has occurred.
+ * 
+ */
+public interface Continuation
+{
+    public final static String ATTRIBUTE = "org.eclipse.jetty.continuation";
+
+    /**
+     * Set the continuation timeout
+     * 
+     * @param timeoutMs
+     *            The time in milliseconds to wait before expiring this
+     *            continuation.
+     */
+    void setTimeout(long timeoutMs);
+
+    /**
+     * Suspend the processing of the request and associated
+     * {@link ServletResponse}.
+     * 
+     * <p>
+     * After this method has been called, the lifecycle of the request will be
+     * extended beyond the return to the container from the
+     * {@link Servlet#service(ServletRequest, ServletResponse)} method and
+     * {@link Filter#doFilter(ServletRequest, ServletResponse, FilterChain)}
+     * calls. If a request is suspended, then the container will not commit the
+     * associated response when the call to the filter chain and/or servlet
+     * service method returns to the container. Any exceptions thrown to the
+     * container by a filter chain and/or servlet for a suspended requests are
+     * silently ignored.
+     * </p>
+     * 
+     * <p>
+     * When the thread calling the filter chain and/or servlet has returned to
+     * the container with a suspended request, the thread is freed for other
+     * tasks and the request is held pending either:
+     * <ul>
+     * <li>a call to {@link ServletRequest#resume()}.</li>
+     * <li>a call to {@link ServletRequest#complete()}.</li>
+     * <li>the passed or default timeout expires.</li>
+     * <li>there is IO activity on the connection that received the request,
+     * such as the close of the connection or the receipt of a pipelined
+     * request.
+     * </ul>
+     * <p>
+     * After any of the events listed above, the suspended request will be
+     * redispatched via the filter and servlet processing.
+     * </p>
+     * 
+     * <p>
+     * Suspend may only be called by a thread that is within the service calling
+     * stack of
+     * {@link Filter#doFilter(ServletRequest, ServletResponse, FilterChain)}
+     * and/or {@link Servlet#service(ServletRequest, ServletResponse)}. A
+     * request that has been dispatched for error handling may not be suspended.
+     * </p>
+     * 
+     * @see {@link #resume()}
+     * @see {@link #complete()}
+     * 
+     * @exception IllegalStateException
+     *                If the calling thread is not within the calling stack of
+     *                {@link Filter#doFilter(ServletRequest, ServletResponse, FilterChain)}
+     *                and/or
+     *                {@link Servlet#service(ServletRequest, ServletResponse)}
+     *                or if the request has been dispatched for error handling.
+     */
+    void suspend();
+
+    /**
+     * Resume a suspended request.
+     * 
+     * <p>
+     * This method can be called by any thread that has been passed a reference
+     * to a suspended request. When called the request is redispatched to the
+     * normal filter chain and servlet processing.
+     * </p>
+     * 
+     * <p>
+     * If resume is called before a suspended request is returned to the
+     * container (ie the thread that called {@link #suspend(long)} is still
+     * within the filter chain and/or servlet service method), then the resume
+     * does not take effect until the call to the filter chain and/or servlet
+     * returns to the container. In this case both {@link #isSuspended()} and
+     * {@link isResumed()} return true.
+     * </p>
+     * 
+     * <p>
+     * Multiple calls to resume are ignored
+     * </p>
+     * 
+     * @see {@link #suspend()}
+     * @exception IllegalStateException
+     *                if the request is not suspended.
+     * 
+     */
+    void resume();
+
+    /**
+     * Complete a suspended request.
+     * 
+     * <p>
+     * This method can be called by any thread that has been passed a reference
+     * to a suspended request. When a request is completed, the associated
+     * response object commited and flushed. The request is not redispatched.
+     * </p>
+     * 
+     * <p>
+     * If complete is called before a suspended request is returned to the
+     * container (ie the thread that called {@link #suspend(long)} is still
+     * within the filter chain and/or servlet service method), then the complete
+     * does not take effect until the call to the filter chain and/or servlet
+     * returns to the container. In this case both {@link #isSuspended()} and
+     * {@link isResumed()} return true.
+     * </p>
+     * 
+     * <p>
+     * Once complete has been called and any thread calling the filter chain
+     * and/or servlet chain has returned to the container, the request lifecycle
+     * is complete. The container is able to recycle request objects, so it is
+     * not valid hold a request reference after the end of the life cycle or to
+     * call any request methods.
+     * 
+     * @see {@link #suspend()}
+     * @exception IllegalStateException
+     *                if the request is not suspended.
+     * 
+     */
+    void complete();
+
+    /**
+     * @return true after {@link #suspend(long)} has been called and before the
+     *         request has been redispatched due to being resumed, completed or
+     *         timed out.
+     */
+    boolean isSuspended();
+
+    /**
+     * @return true if the request has been redispatched by a call to
+     *         {@link #resume()}. Returns false after any subsequent call to
+     *         suspend
+     */
+    boolean isResumed();
+
+    /**
+     * @return true after a request has been redispatched as the result of a
+     *         timeout. Returns false after any subsequent call to suspend.
+     */
+    boolean isExpired();
+
+    /**
+     * @return true while the request is within the initial dispatch to the
+     *         filter chain and/or servlet. Will return false once the calling
+     *         thread has returned to the container after suspend has been
+     *         called and during any subsequent redispatch.
+     */
+    boolean isInitial();
+    
+    
+    /* ------------------------------------------------------------ */
+    /**
+     * Call to signal that the suspending filter/servlet has kept a 
+     * reference to any passed requests wrappers, and that these should 
+     * not be finalized until a onComplete event has been seen.
+     */
+    void keepWrappers();
+    
+    /**
+     * @return True if {@link #keepWrappers()} has been called.
+     */
+    boolean wrappersKept();
+
+    /* ------------------------------------------------------------ */
+    /**
+     * @param listener
+     */
+    void addContinuationListener(ContinuationListener listener);
+
+}
diff --git a/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationEvent.java b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationEvent.java
new file mode 100644
index 0000000000..ab49733b9a
--- /dev/null
+++ b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationEvent.java
@@ -0,0 +1,10 @@
+package org.eclipse.jetty.continuation;
+
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+
+public interface ContinuationEvent
+{
+    public ServletRequest getRequest();
+    public ServletResponse getResponse();
+}
diff --git a/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationFilter.java b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationFilter.java
new file mode 100644
index 0000000000..230a386536
--- /dev/null
+++ b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationFilter.java
@@ -0,0 +1,433 @@
+package org.eclipse.jetty.continuation;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletContext;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+
+public class ContinuationFilter implements Filter
+{
+    final boolean _jetty=true;
+    ServletContext _context;
+
+    public void init(FilterConfig filterConfig) throws ServletException
+    {
+        _context = filterConfig.getServletContext();
+    }
+
+    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException
+    {
+        if (_jetty)
+            chain.doFilter(request,response);
+        else
+        {
+            final FauxContinuation fc = new FauxContinuation();
+            request.setAttribute(Continuation.ATTRIBUTE,fc);
+            boolean complete=false;
+      
+            while (!complete)
+            {
+                try
+                {
+                    chain.doFilter(request,response);
+                }
+                catch(IOException e)
+                {
+                    _context.log("OpenServletFilter caught ",e);
+                }
+                catch(ServletException e)
+                {
+                    _context.log("OpenServletFilter caught ",e);
+                }
+                finally
+                {
+                    complete=fc.handleSuspension();
+                }
+            }
+        }
+    }
+    
+    public void destroy()
+    {
+    }
+
+
+    
+    private class FauxContinuation implements Continuation
+    {
+        private static final int __HANDLING=1;   // Request dispatched to filter/servlet
+        private static final int __SUSPENDING=2;   // Suspend called, but not yet returned to container
+        private static final int __RESUMING=3;     // resumed while suspending
+        private static final int __COMPLETING=4;   // resumed while suspending or suspended
+        private static final int __SUSPENDED=5;    // Suspended and parked
+        private static final int __UNSUSPENDING=6;
+        private static final int __COMPLETE=7;
+        
+        private int _state=__HANDLING;
+        private boolean _initial=true;
+        private boolean _resumed=false;
+        private boolean _timeout=false;
+        private boolean _keepWrappers=false;
+        
+        private  long _timeoutMs=30000; // TODO configure
+        
+        private ArrayList<ContinuationListener> _listeners; 
+
+        
+        /* ------------------------------------------------------------ */
+        /**
+         * @see org.eclipse.jetty.continuation.Continuation#keepWrappers()
+         */
+        public void keepWrappers()
+        {
+            _keepWrappers=true;
+        }
+
+        /* ------------------------------------------------------------ */
+        /**
+         * @see org.eclipse.jetty.continuation.Continuation#wrappersKept()
+         */
+        public boolean wrappersKept()
+        {
+            return _keepWrappers;
+        }
+
+        /* ------------------------------------------------------------ */
+        public boolean isInitial()
+        {
+            synchronized(this)
+            {
+                return _initial;
+            }
+        }
+           
+        public boolean isResumed()
+        {
+            synchronized(this)
+            {
+                return _resumed;
+            }
+        }
+        
+        public boolean isSuspended()
+        {
+            synchronized(this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        return false;
+                    case __SUSPENDING:
+                    case __RESUMING:
+                    case __COMPLETING:
+                    case __SUSPENDED:
+                        return true;
+                    case __UNSUSPENDING:
+                    default:
+                        return false;   
+                }
+            }
+        }
+
+        public boolean isExpired()
+        {
+            synchronized(this)
+            {
+                return _timeout;
+            }
+        }
+
+        public void setTimeout(long timeoutMs)
+        {
+            _timeoutMs = timeoutMs;
+        }
+
+        public void suspend()
+        {
+            synchronized (this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        _timeout=false;
+                        _resumed=false;
+                        _state=__SUSPENDING;
+                        return;
+
+                    case __SUSPENDING:
+                    case __RESUMING:
+                        return;
+
+                    case __COMPLETING:
+                    case __SUSPENDED:
+                    case __UNSUSPENDING:
+                        throw new IllegalStateException(this.getStatusString());
+
+                    default:
+                        throw new IllegalStateException(""+_state);
+                }
+
+            }
+        }
+
+
+        /* ------------------------------------------------------------ */
+        /* (non-Javadoc)
+         * @see org.mortbay.jetty.Suspendor#resume()
+         */
+        public void resume()
+        {
+            synchronized (this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        _resumed=true;
+                        return;
+                        
+                    case __SUSPENDING:
+                        _resumed=true;
+                        _state=__RESUMING;
+                        return;
+
+                    case __RESUMING:
+                    case __COMPLETING:
+                        return;
+                        
+                    case __SUSPENDED:
+                        fauxResume();
+                        _resumed=true;
+                        _state=__UNSUSPENDING;
+                        break;
+                        
+                    case __UNSUSPENDING:
+                        _resumed=true;
+                        return;
+                        
+                    default:
+                        throw new IllegalStateException(this.getStatusString());
+                }
+            }
+            
+        }
+        
+        
+        public void complete()
+        {
+            // just like resume, except don't set _resumed=true;
+            synchronized (this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        throw new IllegalStateException(this.getStatusString());
+                        
+                    case __SUSPENDING:
+                        _state=__COMPLETING;
+                        break;
+                        
+                    case __RESUMING:
+                        break;
+
+                    case __COMPLETING:
+                        return;
+                        
+                    case __SUSPENDED:
+                        _state=__COMPLETING;
+                        fauxResume();
+                        break;
+                        
+                    case __UNSUSPENDING:
+                        return;
+                        
+                    default:
+                        throw new IllegalStateException(this.getStatusString());
+                }
+            }
+            
+        }
+        
+
+        
+        
+        void handling()
+        {
+            synchronized (this)
+            {
+                _keepWrappers=false;
+                switch(_state)
+                {
+                    case __HANDLING:
+                        throw new IllegalStateException(this.getStatusString());
+
+                    case __SUSPENDING:
+                    case __RESUMING:
+                        throw new IllegalStateException(this.getStatusString());
+
+                    case __COMPLETING:
+                        return;
+
+                    case __SUSPENDED:
+                        fauxResume();
+                    case __UNSUSPENDING:
+                        _state=__HANDLING;
+                        return;
+
+                    default:
+                        throw new IllegalStateException(""+_state);
+                }
+
+            }
+        }
+
+        /* ------------------------------------------------------------ */
+        /**
+         * @return true if handling is complete
+         */
+        public boolean handleSuspension()
+        {
+            synchronized (this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        _state=__COMPLETE;
+                        return true;
+
+                    case __SUSPENDING:
+                        _initial=false;
+                        _state=__SUSPENDED;
+                        fauxSuspend(); // could block and change state.
+                        if (_state==__SUSPENDED || _state==__COMPLETING)
+                            return true;
+                        
+                        _initial=false;
+                        _state=__HANDLING;
+                        return false; 
+
+                    case __RESUMING:
+                        _initial=false;
+                        _state=__HANDLING;
+                        return false; 
+
+                    case __COMPLETING:
+                        _initial=false;
+                        _state=__COMPLETE;
+                        return true;
+
+                    case __SUSPENDED:
+                    case __UNSUSPENDING:
+                    default:
+                        throw new IllegalStateException(this.getStatusString());
+                }
+            }
+        }
+
+        /* ------------------------------------------------------------ */
+        protected void expire()
+        {
+            // just like resume, except don't set _resumed=true;
+
+            synchronized (this)
+            {
+                switch(_state)
+                {
+                    case __HANDLING:
+                        return;
+                        
+                    case __SUSPENDING:
+                        _timeout=true;
+                        _state=__RESUMING;
+                        fauxResume();
+                        return;
+                        
+                    case __RESUMING:
+                        return;
+                        
+                    case __COMPLETING:
+                        return;
+                        
+                    case __SUSPENDED:
+                        _timeout=true;
+                        _state=__UNSUSPENDING;
+                        break;
+                        
+                    case __UNSUSPENDING:
+                        _timeout=true;
+                        return;
+                        
+                    default:
+                        throw new IllegalStateException(this.getStatusString());
+                }
+            }
+        }
+
+        private void fauxSuspend()
+        {
+            long expire_at = System.currentTimeMillis()+_timeoutMs;
+            long wait=_timeoutMs;
+            while (_timeoutMs>0 && wait>0)
+            {
+                try
+                {
+                    this.wait(wait);
+                }
+                catch (InterruptedException e)
+                {
+                    _context.log("OpenServletFilter caught ",e);
+                }
+                wait=expire_at-System.currentTimeMillis();
+            }
+
+            if (_timeoutMs>0 && wait<=0)
+                expire();
+        }
+        
+        private void fauxResume()
+        {
+            _timeoutMs=0;
+            this.notifyAll();
+        }
+        
+        public String toString()
+        {
+            return getStatusString();
+        }
+        
+        String getStatusString()
+        {
+            synchronized (this)
+            {
+                return
+                ((_state==__HANDLING)?"HANDLING":
+                        (_state==__SUSPENDING)?"SUSPENDING":
+                            (_state==__SUSPENDED)?"SUSPENDED":
+                                (_state==__RESUMING)?"RESUMING":
+                                    (_state==__UNSUSPENDING)?"UNSUSPENDING":
+                                        (_state==__COMPLETING)?"COMPLETING":
+                                        ("???"+_state))+
+                (_initial?",initial":"")+
+                (_resumed?",resumed":"")+
+                (_timeout?",timeout":"");
+            }
+        }
+
+        
+        public void addContinuationListener(ContinuationListener listener)
+        {
+            if (_listeners==null)
+                _listeners=new ArrayList<ContinuationListener>();
+            _listeners.add(listener);
+            
+            // TODO Call the listeners
+        }
+        
+        
+    }
+}
diff --git a/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationListener.java b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationListener.java
new file mode 100644
index 0000000000..cf4e926d78
--- /dev/null
+++ b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationListener.java
@@ -0,0 +1,11 @@
+
+package org.eclipse.jetty.continuation;
+
+import java.io.IOException;
+import java.util.EventListener;
+
+public interface ContinuationListener extends EventListener 
+{    
+    public void onComplete(ContinuationEvent event) throws IOException;
+    public void onTimeout(ContinuationEvent event) throws IOException;
+}
diff --git a/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationSupport.java b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationSupport.java
new file mode 100644
index 0000000000..22faf15a4f
--- /dev/null
+++ b/jetty-continuation/src/main/java/org/eclipse/jetty/continuation/ContinuationSupport.java
@@ -0,0 +1,31 @@
+// ========================================================================
+// Copyright (c) 2004-2009 Mort Bay Consulting Pty. Ltd.
+// ------------------------------------------------------------------------
+// All rights reserved. This program and the accompanying materials
+// are made available under the terms of the Eclipse Public License v1.0
+// and Apache License v2.0 which accompanies this distribution.
+// The Eclipse Public License is available at 
+// http://www.eclipse.org/legal/epl-v10.html
+// The Apache License v2.0 is available at
+// http://www.opensource.org/licenses/apache2.0.php
+// You may elect to redistribute this code under either of these licenses. 
+// ========================================================================
+
+package org.eclipse.jetty.continuation;
+
+import javax.servlet.ServletRequest;
+
+/* ------------------------------------------------------------ */
+/** ContinuationSupport.
+ * 
+ */
+public class ContinuationSupport
+{
+    public static Continuation getContinuation(ServletRequest request)
+    {
+        Continuation continuation = (Continuation) request.getAttribute(Continuation.ATTRIBUTE);
+        if (continuation==null)
+            throw new IllegalStateException("!Jetty && !ContinuationFilter installed");
+        return continuation;
+    }
+}