405570 spdy push - send push resources in the order they have been requested. Do not interleave sending the bytes of the push streams.

jetty-spdy/spdy-core/src/main/java/org/eclipse/jetty/spdy/IStream.java

@@ -59,7 +59,9 @@ public interface IStream extends Stream, Callback
      */
     public void setStreamFrameListener(StreamFrameListener listener);
 
-    //TODO: javadoc thomas
+    /**
+     * @return the stream frame listener associated to this stream
+     */
     public StreamFrameListener getStreamFrameListener();
 
     /**

jetty-spdy/spdy-http-server/src/main/java/org/eclipse/jetty/spdy/server/http/HttpTransportOverSPDY.java

@@ -20,6 +20,7 @@ package org.eclipse.jetty.spdy.server.http;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -46,6 +47,7 @@ import org.eclipse.jetty.spdy.api.StreamStatus;
 import org.eclipse.jetty.util.BlockingCallback;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.ConcurrentArrayQueue;
 import org.eclipse.jetty.util.Fields;
 import org.eclipse.jetty.util.Promise;
 import org.eclipse.jetty.util.log.Log;
@@ -206,28 +208,100 @@ public class HttpTransportOverSPDY implements HttpTransport
         short version = stream.getSession().getVersion();
         if (responseHeaders.get(HTTPSPDYHeader.STATUS.name(version)).value().startsWith("200") && !stream.isClosed())
         {
-            // We have a 200 OK with some content to send, check the push strategy
+            Set<String> pushResources = pushStrategy.apply(stream, requestHeaders, responseHeaders);
+            if (pushResources.size() > 0)
+            {
+                PushResourceCoordinator pushResourceCoordinator = new PushResourceCoordinator(pushResources);
+                pushResourceCoordinator.coordinate();
+            }
+        }
+    }
+
+    private class PushHttpTransportOverSPDY extends HttpTransportOverSPDY
+    {
+        private final PushResourceCoordinator pushResourceCoordinator;
+
+        private PushHttpTransportOverSPDY(Connector connector, HttpConfiguration configuration, EndPoint endPoint,
+                                          PushStrategy pushStrategy, Stream stream, Fields requestHeaders,
+                                          PushResourceCoordinator pushResourceCoordinator)
+        {
+            super(connector, configuration, endPoint, pushStrategy, stream, requestHeaders);
+            this.pushResourceCoordinator = pushResourceCoordinator;
+        }
+
+        @Override
+        public void completed()
+        {
+            pushResourceCoordinator.complete();
+        }
+    }
+
+    private class PushResourceCoordinator
+    {
+        private final Queue<PushResource> queue = new ConcurrentArrayQueue<>();
+        private final AtomicBoolean channelActive = new AtomicBoolean(false);
+        private final Set<String> pushResources;
+
+        private PushResourceCoordinator(Set<String> pushResources)
+        {
+            this.pushResources = pushResources;
+        }
+
+        private void coordinate()
+        {
+            for (String pushResource : pushResources)
+                pushResource(pushResource);
+        }
+
+        private void sendNextResourceData()
+        {
+            if (channelActive.compareAndSet(false, true))
+            {
+                PushResource pushResource = queue.poll();
+                if (pushResource != null)
+                {
+                    HttpChannelOverSPDY pushChannel = newHttpChannelOverSPDY(pushResource.getPushStream(),
+                            pushResource.getPushRequestHeaders());
+                    pushChannel.requestStart(pushResource.getPushRequestHeaders(), true);
+                }
+            }
+        }
+
+        private HttpChannelOverSPDY newHttpChannelOverSPDY(Stream pushStream, Fields pushRequestHeaders)
+        {
+            HttpTransport transport = new PushHttpTransportOverSPDY(connector, configuration, endPoint, pushStrategy,
+                    pushStream, pushRequestHeaders, this);
+            HttpInputOverSPDY input = new HttpInputOverSPDY();
+            return new HttpChannelOverSPDY(connector, configuration, endPoint, transport, input, pushStream);
+        }
+
+        private void pushResource(String pushResource)
+        {
+            short version = stream.getSession().getVersion();
             Fields.Field scheme = requestHeaders.get(HTTPSPDYHeader.SCHEME.name(version));
             Fields.Field host = requestHeaders.get(HTTPSPDYHeader.HOST.name(version));
             Fields.Field uri = requestHeaders.get(HTTPSPDYHeader.URI.name(version));
-            Set<String> pushResources = pushStrategy.apply(stream, requestHeaders, responseHeaders);
+            Fields pushHeaders = createPushHeaders(scheme, host, pushResource);
+            final Fields pushRequestHeaders = createRequestHeaders(scheme, host, uri, pushResource);
 
-            for (String pushResource : pushResources)
+            // TODO: handle the timeout better
+            stream.push(new PushInfo(0, TimeUnit.MILLISECONDS, pushHeaders, false), new Promise.Adapter<Stream>()
             {
-                Fields pushHeaders = createPushHeaders(scheme, host, pushResource);
+                @Override
-                final Fields pushRequestHeaders = createRequestHeaders(scheme, host, uri, pushResource);
+                public void succeeded(Stream pushStream)
-
-                // TODO: handle the timeout better
-                stream.push(new PushInfo(0, TimeUnit.MILLISECONDS, pushHeaders, false), new Promise.Adapter<Stream>()
                 {
-                    @Override
+                    queue.offer(new PushResource(pushStream, pushRequestHeaders));
-                    public void succeeded(Stream pushStream)
+                    sendNextResourceData();
-                    {
+                }
-                        HttpChannelOverSPDY pushChannel = newHttpChannelOverSPDY(pushStream, pushRequestHeaders);
+            });
-                        pushChannel.requestStart(pushRequestHeaders, true);
+        }
-                    }
+
-                });
+        public void complete()
-            }
+        {
+            if (channelActive.compareAndSet(true, false))
+                sendNextResourceData();
+            else
+                throw new IllegalStateException("No channel was active when complete has been called.");
         }
     }
 
@@ -261,10 +335,25 @@ public class HttpTransportOverSPDY implements HttpTransport
         return pushHeaders;
     }
 
-    private HttpChannelOverSPDY newHttpChannelOverSPDY(Stream pushStream, Fields pushRequestHeaders)
+    private class PushResource
     {
-        HttpTransport transport = new HttpTransportOverSPDY(connector, configuration, endPoint, pushStrategy, pushStream, pushRequestHeaders);
+        private final Stream pushStream;
-        HttpInputOverSPDY input = new HttpInputOverSPDY();
+        private final Fields pushRequestHeaders;
-        return new HttpChannelOverSPDY(connector, configuration, endPoint, transport, input, pushStream);
+
+        public PushResource(Stream pushStream, Fields pushRequestHeaders)
+        {
+            this.pushStream = pushStream;
+            this.pushRequestHeaders = pushRequestHeaders;
+        }
+
+        public Stream getPushStream()
+        {
+            return pushStream;
+        }
+
+        public Fields getPushRequestHeaders()
+        {
+            return pushRequestHeaders;
+        }
     }
 }

jetty-spdy/spdy-http-server/src/main/java/org/eclipse/jetty/spdy/server/http/PushStrategy.java

@@ -34,6 +34,8 @@ public interface PushStrategy
     /**
      * <p>Applies the SPDY push logic for the primary resource.</p>
      *
+     *
+     *
      * @param stream the primary resource stream
      * @param requestHeaders the primary resource request headers
      * @param responseHeaders the primary resource response headers

jetty-spdy/spdy-http-server/src/main/java/org/eclipse/jetty/spdy/server/http/ReferrerPushStrategy.java

@@ -27,6 +27,7 @@ import java.util.Locale;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Pattern;
@@ -262,7 +263,7 @@ public class ReferrerPushStrategy implements PushStrategy
     private class MainResource
     {
         private final String name;
-        private final Set<String> resources = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+        private final CopyOnWriteArraySet<String> resources = new CopyOnWriteArraySet<>();
         private final AtomicLong firstResourceAdded = new AtomicLong(-1);
 
         private MainResource(String name)

jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/http/ReferrerPushStrategyTest.java

@@ -22,9 +22,13 @@ import java.io.IOException;
 import java.io.PrintWriter;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import javax.servlet.ServletException;
+import javax.servlet.ServletOutputStream;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
@@ -41,6 +45,8 @@ import org.eclipse.jetty.spdy.api.RstInfo;
 import org.eclipse.jetty.spdy.api.SPDY;
 import org.eclipse.jetty.spdy.api.Session;
 import org.eclipse.jetty.spdy.api.SessionFrameListener;
+import org.eclipse.jetty.spdy.api.Settings;
+import org.eclipse.jetty.spdy.api.SettingsInfo;
 import org.eclipse.jetty.spdy.api.Stream;
 import org.eclipse.jetty.spdy.api.StreamFrameListener;
 import org.eclipse.jetty.spdy.api.StreamStatus;
@@ -56,6 +62,7 @@ import org.junit.Before;
 import org.junit.Test;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.core.IsEqual.equalTo;
 import static org.junit.Assert.assertThat;
 
 public class ReferrerPushStrategyTest extends AbstractHTTPSPDYTest
@@ -159,25 +166,187 @@ public class ReferrerPushStrategyTest extends AbstractHTTPSPDYTest
     public void testMaxConcurrentStreamsToDisablePush() throws Exception
     {
         final CountDownLatch pushReceivedLatch = new CountDownLatch(1);
-        Session session = sendMainRequestAndCSSRequest(new SessionFrameListener.Adapter()
+
+        Session pushCacheBuildSession = startClient(version, serverAddress, null);
+
+        pushCacheBuildSession.syn(new SynInfo(mainRequestHeaders, true), new StreamFrameListener.Adapter());
+        pushCacheBuildSession.syn(new SynInfo(associatedCSSRequestHeaders, true), new StreamFrameListener.Adapter());
+
+        Session session = startClient(version, serverAddress, null);
+
+        Settings settings = new Settings();
+        settings.put(new Settings.Setting(Settings.ID.MAX_CONCURRENT_STREAMS, 0));
+        SettingsInfo settingsInfo = new SettingsInfo(settings);
+        session.settings(settingsInfo);
+
+        session.syn(new SynInfo(mainRequestHeaders, true), new StreamFrameListener.Adapter()
         {
             @Override
-            public StreamFrameListener onSyn(Stream stream, SynInfo synInfo)
+            public StreamFrameListener onPush(Stream stream, PushInfo pushInfo)
             {
                 pushReceivedLatch.countDown();
+                return super.onPush(stream, pushInfo);
+            }
+        });
+
+        assertThat("No push stream is received", pushReceivedLatch.await(1, TimeUnit.SECONDS), is(false));
+    }
+
+    @Test
+    public void testPushResourceOrder() throws Exception
+    {
+        final CountDownLatch allExpectedPushesReceivedLatch = new CountDownLatch(4);
+
+        Session pushCacheBuildSession = startClient(version, serverAddress, null);
+
+        sendRequest(pushCacheBuildSession, mainRequestHeaders, null, null);
+        sendRequest(pushCacheBuildSession, associatedCSSRequestHeaders, null, null);
+        sendRequest(pushCacheBuildSession, associatedJSRequestHeaders, null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/image1.jpg", mainResource), null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/image2.jpg", mainResource), null, null);
+
+        Session session = startClient(version, serverAddress, null);
+
+        session.syn(new SynInfo(mainRequestHeaders, true), new StreamFrameListener.Adapter()
+        {
+            @Override
+            public StreamFrameListener onPush(Stream stream, PushInfo pushInfo)
+            {
+                LOG.info("onPush: stream: {}, pushInfo: {}", stream, pushInfo);
+                String uriHeader = pushInfo.getHeaders().get(HTTPSPDYHeader.URI.name(version)).value();
+                switch ((int)allExpectedPushesReceivedLatch.getCount())
+                {
+                    case 4:
+                        assertThat("1st pushed resource is the css", uriHeader.endsWith("css"), is(true));
+                        break;
+                    case 3:
+                        assertThat("2nd pushed resource is the js", uriHeader.endsWith("js"), is(true));
+                        break;
+                    case 2:
+                        assertThat("3rd pushed resource is image1", uriHeader.endsWith("image1.jpg"),
+                                is(true));
+                        break;
+                    case 1:
+                        assertThat("4th pushed resource is image2", uriHeader.endsWith("image2.jpg"),
+                                is(true));
+                        break;
+                }
+                allExpectedPushesReceivedLatch.countDown();
+                return super.onPush(stream, pushInfo);
+            }
+        });
+
+        assertThat("All expected push resources have been received", allExpectedPushesReceivedLatch.await(5,
+                TimeUnit.SECONDS), is(true));
+    }
+
+    @Test
+    public void testThatPushResourcesAreUnique() throws Exception
+    {
+        final CountDownLatch pushReceivedLatch = new CountDownLatch(2);
+        sendMainRequestAndCSSRequest(null);
+        sendMainRequestAndCSSRequest(null);
+
+        Session session = startClient(version, serverAddress, null);
+
+        session.syn(new SynInfo(mainRequestHeaders, true), new StreamFrameListener.Adapter()
+        {
+            @Override
+            public StreamFrameListener onPush(Stream stream, PushInfo pushInfo)
+            {
+                pushReceivedLatch.countDown();
+                LOG.info("Push received: {}", pushInfo);
                 return null;
             }
         });
 
-//        Settings settings = new Settings();
+        assertThat("style.css has been pushed only once", pushReceivedLatch.await(1, TimeUnit.SECONDS), is(false));
-//        settings.put(new Settings.Setting(Settings.ID.MAX_CONCURRENT_STREAMS, 0));
+    }
-//        SettingsInfo settingsInfo = new SettingsInfo(settings);
-//
-//        session.settings(settingsInfo);
 
-        sendRequest(session, mainRequestHeaders, null, null);
+    @Test
+    public void testPushResourceAreSentNonInterleaved() throws Exception
+    {
+        final CountDownLatch allExpectedPushesReceivedLatch = new CountDownLatch(4);
+        final CountDownLatch allPushDataReceivedLatch = new CountDownLatch(4);
+        final CopyOnWriteArrayList<Integer> dataReceivedOrder = new CopyOnWriteArrayList<>();
 
-        assertThat(pushReceivedLatch.await(1, TimeUnit.SECONDS), is(false));
+        InetSocketAddress bigResponseServerAddress = startHTTPServer(version, new AbstractHandler()
+        {
+            @Override
+            public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
+            {
+                byte[] bytes = new byte[32768];
+                new Random().nextBytes(bytes);
+                ServletOutputStream outputStream = response.getOutputStream();
+                outputStream.write(bytes);
+                baseRequest.setHandled(true);
+            }
+        });
+        Session pushCacheBuildSession = startClient(version, bigResponseServerAddress, null);
+
+        Fields mainResourceHeaders = createHeadersWithoutReferrer(mainResource);
+        sendRequest(pushCacheBuildSession, mainResourceHeaders, null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/style.css", mainResource), null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/javascript.js", mainResource), null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/image1.jpg", mainResource), null, null);
+        sendRequest(pushCacheBuildSession, createHeaders("/image2.jpg", mainResource), null, null);
+
+        Session session = startClient(version, bigResponseServerAddress, null);
+
+        session.syn(new SynInfo(mainResourceHeaders, true), new StreamFrameListener.Adapter()
+        {
+            AtomicInteger currentStreamId = new AtomicInteger(2);
+
+            @Override
+            public StreamFrameListener onPush(Stream stream, PushInfo pushInfo)
+            {
+                LOG.info("Received push for stream: {} {}", stream.getId(), pushInfo);
+                String uriHeader = pushInfo.getHeaders().get(HTTPSPDYHeader.URI.name(version)).value();
+                switch ((int)allExpectedPushesReceivedLatch.getCount())
+                {
+                    case 4:
+                        assertThat("1st pushed resource is the css", uriHeader.endsWith("css"), is(true));
+                        break;
+                    case 3:
+                        assertThat("2nd pushed resource is the js", uriHeader.endsWith("js"), is(true));
+                        break;
+                    case 2:
+                        assertThat("3rd pushed resource is image1", uriHeader.endsWith("image1.jpg"),
+                                is(true));
+                        break;
+                    case 1:
+                        assertThat("4th pushed resource is image2", uriHeader.endsWith("image2.jpg"),
+                                is(true));
+                        break;
+                }
+                allExpectedPushesReceivedLatch.countDown();
+                return new Adapter()
+                {
+                    @Override
+                    public void onData(Stream stream, DataInfo dataInfo)
+                    {
+                        if (stream.getId() != currentStreamId.get())
+                            throw new IllegalStateException("Streams interleaved. Expected StreamId: " +
+                                    currentStreamId + " but was: " + stream.getId());
+                        dataInfo.consume(dataInfo.available());
+                        if (dataInfo.isClose())
+                        {
+                            currentStreamId.compareAndSet(currentStreamId.get(), currentStreamId.get() + 2);
+                            allPushDataReceivedLatch.countDown();
+                            dataReceivedOrder.add(stream.getId());
+                        }
+
+                        LOG.info(stream.getId() + ":" + dataInfo);
+                    }
+                };
+            }
+        });
+
+        assertThat("All push resources received", allExpectedPushesReceivedLatch.await(5, TimeUnit.SECONDS), is(true));
+        assertThat("All pushData received", allPushDataReceivedLatch.await(5, TimeUnit.SECONDS), is(true));
+        assertThat("The data for different push streams has not been interleaved",
+                dataReceivedOrder.toString(), equalTo("[2, 4, 6, 8]"));
+        LOG.info(dataReceivedOrder.toString());
     }
 
     private InetSocketAddress createServer() throws Exception

jetty-spdy/spdy-http-server/src/test/java/org/eclipse/jetty/spdy/server/http/ReferrerPushStrategyUnitTest.java

@@ -18,10 +18,6 @@
 
 package org.eclipse.jetty.spdy.server.http;
 
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
-import static org.mockito.Mockito.when;
-
 import java.util.Arrays;
 import java.util.Set;
 
@@ -35,6 +31,10 @@ import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.when;
+
 @RunWith(MockitoJUnitRunner.class)
 public class ReferrerPushStrategyUnitTest
 {

jetty-spdy/spdy-http-server/src/test/resources/jetty-logging.properties

@@ -4,4 +4,5 @@ org.eclipse.jetty.util.log.class=org.eclipse.jetty.util.log.StdErrLog
 #org.eclipse.jetty.io.ssl.LEVEL=DEBUG
 #org.eclipse.jetty.spdy.LEVEL=DEBUG
 #org.eclipse.jetty.client.LEVEL=DEBUG
+#org.eclipse.jetty.spdy.server.http.ReferrerPushStrategy.LEVEL=DEBUG
 #org.mortbay.LEVEL=DEBUG