Initial draft for SPDY push.

3 files changed, 167 insertions, 3 deletions

diff --git a/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/PushStrategy.java b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/PushStrategy.java
new file mode 100644
index 0000000000..14032f7b62
--- /dev/null
+++ b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/PushStrategy.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.eclipse.jetty.spdy.http;
+
+import java.util.Set;
+
+import org.eclipse.jetty.spdy.api.Headers;
+import org.eclipse.jetty.spdy.api.Stream;
+
+/**
+ *
+ */
+public interface PushStrategy
+{
+    public Set<String> apply(Stream stream, Headers requestHeaders, Headers responseHeaders);
+}
diff --git a/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ReferrerPushStrategy.java b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ReferrerPushStrategy.java
new file mode 100644
index 0000000000..1f6fb95667
--- /dev/null
+++ b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ReferrerPushStrategy.java
@@ -0,0 +1,88 @@
+package org.eclipse.jetty.spdy.http;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.eclipse.jetty.spdy.api.Headers;
+import org.eclipse.jetty.spdy.api.Stream;
+import org.eclipse.jetty.util.log.Log;
+import org.eclipse.jetty.util.log.Logger;
+
+/**
+ * TODO: this class is kind-of leaking since the resources map is always adding entries
+ * TODO: although these entries will be limited by the application pages
+ * TODO: however, there is no ConcurrentLinkedHashMap yet in JDK (there is in Guava though)
+ * TODO: so we cannot use the built-in LRU features of LinkedHashMap
+ */
+public class ReferrerPushStrategy implements PushStrategy
+{
+    private static final Logger logger = Log.getLogger(ReferrerPushStrategy.class);
+    private final ConcurrentMap<String, Set<String>> resources = new ConcurrentHashMap<>();
+    private List<String> mainSuffixes = new ArrayList<>();
+    private List<String> pushSuffixes = new ArrayList<>();
+
+    @Override
+    public Set<String> apply(Stream stream, Headers requestHeaders, Headers responseHeaders)
+    {
+        String url = requestHeaders.get("url").value();
+        if (!hasQueryString(url))
+        {
+            if (isMainResource(url))
+            {
+                return pushResources(url);
+            }
+            else if (isPushResource(url))
+            {
+                String referrer = requestHeaders.get("referer").value();
+                Set<String> pushResources = resources.get(referrer);
+                if (pushResources == null || !pushResources.contains(url))
+                {
+                    buildPushResources(url, referrer);
+                }
+                else
+                {
+                    return pushResources(url);
+                }
+            }
+        }
+        return Collections.emptySet();
+    }
+
+    private boolean hasQueryString(String url)
+    {
+        return url.contains("?");
+    }
+
+    private boolean isMainResource(String url)
+    {
+        // TODO
+        return false;
+    }
+
+    private boolean isPushResource(String url)
+    {
+        // TODO
+        return false;
+    }
+
+    private Set<String> pushResources(String url)
+    {
+        Set<String> pushResources = resources.get(url);
+        if (pushResources == null)
+            return Collections.emptySet();
+        return Collections.unmodifiableSet(pushResources);
+    }
+
+    private void buildPushResources(String url, String referrer)
+    {
+        Set<String> pushResources = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+        Set<String> existing = resources.putIfAbsent(referrer, pushResources);
+        if (existing != null)
+            pushResources = existing;
+        pushResources.add(url);
+    }
+}
diff --git a/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ServerHTTPSPDYAsyncConnection.java b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ServerHTTPSPDYAsyncConnection.java
index 2882bc4346..a069937706 100644
--- a/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ServerHTTPSPDYAsyncConnection.java
+++ b/jetty-spdy/spdy-jetty-http/src/main/java/org/eclipse/jetty/spdy/http/ServerHTTPSPDYAsyncConnection.java
@@ -22,6 +22,7 @@ import java.io.InterruptedIOException;
 import java.nio.ByteBuffer;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -50,6 +51,7 @@ import org.eclipse.jetty.spdy.api.DataInfo;
 import org.eclipse.jetty.spdy.api.Headers;
 import org.eclipse.jetty.spdy.api.ReplyInfo;
 import org.eclipse.jetty.spdy.api.Stream;
+import org.eclipse.jetty.spdy.api.SynInfo;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 
@@ -117,7 +119,7 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
             {
                 dispatched = true;
                 logger.debug("Dispatching task {}", task);
-                getServer().getThreadPool().dispatch(new Runnable()
+                execute(new Runnable()
                 {
                     @Override
                     public void run()
@@ -133,6 +135,11 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
         }
     }
 
+    protected void execute(Runnable task)
+    {
+        getServer().getThreadPool().dispatch(task);
+    }
+
     @Override
     public Connection handle()
     {
@@ -157,7 +164,7 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
                     String m = method.value();
                     String u = uri.value();
                     String v = version.value();
-                    logger.debug("HTTP > {} {} {}", new Object[]{m, u, v});
+                    logger.debug("HTTP > {} {} {}", m, u, v);
                     startRequest(new ByteArrayBuffer(m), new ByteArrayBuffer(u), new ByteArrayBuffer(v));
 
                     updateState(State.HEADERS);
@@ -363,6 +370,31 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
         });
     }
 
+    protected void reply(Stream stream, ReplyInfo replyInfo)
+    {
+        if (!stream.isUnidirectional())
+            stream.reply(replyInfo);
+        if (replyInfo.getHeaders().get("status").value().startsWith("200") && !stream.isClosed())
+        {
+            // We have a 200 OK with some content to send
+            Set<String> pushResources = pushStrategy.apply(stream, this.headers, replyInfo.getHeaders());
+            for (String url : pushResources)
+            {
+                Headers pushHeaders = new Headers();
+                pushHeaders.put("method", "GET");
+                pushHeaders.put("url", url);
+                pushHeaders.put("version", "HTTP/1.1");
+                Headers.Header acceptEncoding = headers.get("accept-encoding");
+                if (acceptEncoding != null)
+                    pushHeaders.put(acceptEncoding);
+                Stream pushStream = stream.syn(new SynInfo(pushHeaders, true));
+                Synchronous connection = new Synchronous(getConnector(), getEndPoint(), getServer(), , pushStream);
+                connection.beginRequest(pushHeaders);
+                connection.endRequest();
+            }
+        }
+    }
+
     private Buffer consumeContent(long maxIdleTime) throws IOException, InterruptedException
     {
         while (true)
@@ -566,7 +598,7 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
             // We have to query the HttpGenerator and its buffers to know
             // whether there is content buffered; if so, send the data frame
             Buffer content = getContentBuffer();
-            stream.reply(new ReplyInfo(headers, content == null));
+            reply(stream, new ReplyInfo(headers, content == null));
             if (content != null)
             {
                 closed = allContentAdded || isAllContentWritten();
@@ -674,4 +706,18 @@ public class ServerHTTPSPDYAsyncConnection extends AbstractHttpConnection implem
             }
         }
     }
+
+    private static class Synchronous extends ServerHTTPSPDYAsyncConnection
+    {
+        private Synchronous(Connector connector, AsyncEndPoint endPoint, Server server, SPDYAsyncConnection connection, Stream stream)
+        {
+            super(connector, endPoint, server, connection, stream);
+        }
+
+        @Override
+        protected void execute(Runnable task)
+        {
+            task.run();
+        }
+    }
 }