Merge pull request #75 from square/jwilson/timeouts

Make timeouts work for SPDY.

src/main/java/com/squareup/okhttp/Connection.java

@@ -156,6 +156,7 @@ public final class Connection implements Closeable {
         if (modernTls
                 && (selectedProtocol = platform.getNpnSelectedProtocol(sslSocket)) != null) {
             if (Arrays.equals(selectedProtocol, SPDY2)) {
+                sslSocket.setSoTimeout(0); // SPDY timeouts are set per-stream.
                 spdyConnection = new SpdyConnection.Builder(true, in, out).build();
             } else if (!Arrays.equals(selectedProtocol, HTTP_11)) {
                 throw new IOException("Unexpected NPN transport "

src/main/java/com/squareup/okhttp/internal/net/http/SpdyTransport.java

@@ -55,6 +55,7 @@ public final class SpdyTransport implements Transport {
         boolean hasResponseBody = true;
         stream = spdyConnection.newStream(requestHeaders.toNameValueBlock(),
                 hasRequestBody, hasResponseBody);
+        stream.setReadTimeout(httpEngine.policy.getReadTimeout());
     }
 
     @Override public void writeRequestBody(RetryableOutputStream requestBody) throws IOException {

src/main/java/com/squareup/okhttp/internal/net/spdy/SpdyStream.java

@@ -24,6 +24,7 @@ import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.io.OutputStream;
 import java.net.ProtocolException;
+import java.net.SocketTimeoutException;
 import static java.nio.ByteOrder.BIG_ENDIAN;
 import java.util.List;
 
@@ -60,6 +61,7 @@ public final class SpdyStream {
 
     private final int id;
     private final SpdyConnection connection;
+    private long readTimeoutMillis = 0;
 
     /** Headers sent by the stream initiator. Immutable and non null. */
     private final List<String> requestHeaders;
@@ -188,6 +190,18 @@ public final class SpdyStream {
         connection.writeSynReply(id, flags, responseHeaders);
     }
 
+    /**
+     * Sets the maximum time to wait on input stream reads before failing with a
+     * {@code SocketTimeoutException}, or {@code 0} to wait indefinitely.
+     */
+    public void setReadTimeout(long readTimeoutMillis) {
+        this.readTimeoutMillis = readTimeoutMillis;
+    }
+
+    public long getReadTimeoutMillis() {
+        return readTimeoutMillis;
+    }
+
     /**
      * Returns an input stream that can be used to read data from the peer.
      */
@@ -345,13 +359,7 @@ public final class SpdyStream {
         @Override public int read(byte[] b, int offset, int count) throws IOException {
             synchronized (SpdyStream.this) {
                 checkOffsetAndCount(b.length, offset, count);
-                while (pos == -1 && !finished && !closed && rstStatusCode == -1) {
-                    try {
-                        SpdyStream.this.wait();
-                    } catch (InterruptedException e) {
-                        throw new InterruptedIOException();
-                    }
-                }
+                waitUntilReadable();
                 checkNotClosed();
 
                 if (pos == -1) {
@@ -390,6 +398,34 @@ public final class SpdyStream {
             }
         }
 
+        /**
+         * Returns once the input stream is either readable or finished. Throws
+         * a {@link SocketTimeoutException} if the read timeout elapses before
+         * that happens.
+         */
+        private void waitUntilReadable() throws IOException {
+            long start = 0;
+            long remaining = 0;
+            if (readTimeoutMillis != 0) {
+                start = (System.nanoTime() / 1000000);
+                remaining = readTimeoutMillis;
+            }
+            try {
+                while (pos == -1 && !finished && !closed && rstStatusCode == -1) {
+                    if (readTimeoutMillis == 0) {
+                        SpdyStream.this.wait();
+                    } else if (remaining > 0) {
+                        SpdyStream.this.wait(remaining);
+                        remaining = start + readTimeoutMillis - (System.nanoTime() / 1000000);
+                    } else {
+                        throw new SocketTimeoutException();
+                    }
+                }
+            } catch (InterruptedException e) {
+                throw new InterruptedIOException();
+            }
+        }
+
         void receive(InputStream in, int byteCount) throws IOException {
             assert (!Thread.holdsLock(SpdyStream.this));
 

src/test/java/com/squareup/okhttp/internal/net/spdy/MockSpdyPeer.java

@@ -71,7 +71,7 @@ public final class MockSpdyPeer {
                 try {
                     readAndWriteFrames(serverSocket);
                 } catch (IOException e) {
-                    e.printStackTrace(); // TODO
+                    throw new RuntimeException(e);
                 }
             }
         });

src/test/java/com/squareup/okhttp/internal/net/spdy/SpdyConnectionTest.java

@@ -823,6 +823,32 @@ public final class SpdyConnectionTest {
         assertEquals(-1, ping.roundTripTime());
     }
 
+    @Test public void readTimeoutExpires() throws Exception {
+        // write the mocking script
+        peer.acceptFrame(); // SYN STREAM
+        peer.sendFrame().synReply(0, 1, Arrays.asList("a", "android"));
+        peer.play();
+
+        // play it back
+        SpdyConnection connection = new SpdyConnection.Builder(true, peer.openSocket()).build();
+        SpdyStream stream = connection.newStream(Arrays.asList("b", "banana"), true, true);
+        stream.setReadTimeout(1000);
+        InputStream in = stream.getInputStream();
+        long startNanos = System.nanoTime();
+        try {
+            in.read();
+            fail();
+        } catch (IOException expected) {
+        }
+        long elapsedNanos = System.nanoTime() - startNanos;
+        assertEquals(1000d, TimeUnit.NANOSECONDS.toMillis(elapsedNanos), 200d /* 200ms delta */);
+        assertEquals(1, connection.openStreamCount());
+
+        // verify the peer received what was expected
+        MockSpdyPeer.InFrame synStream = peer.takeFrame();
+        assertEquals(TYPE_SYN_STREAM, synStream.type);
+    }
+
     private void writeAndClose(SpdyStream stream, String data) throws IOException {
         OutputStream out = stream.getOutputStream();
         out.write(data.getBytes("UTF-8"));