Merge pull request square#1654 from square/jwilson_0516_dont_share_ti…

…meouts

Don't share timeouts between pooled connections.

okhttp-tests/src/test/java/com/squareup/okhttp/CallTest.java

@@ -694,6 +694,72 @@ private void postBodyRetransmittedAfterAuthorizationFail(String body) throws Exc
     }
   }
 
+  @Test public void reusedSinksGetIndependentTimeoutInstances() throws Exception {
+    server.enqueue(new MockResponse());
+    server.enqueue(new MockResponse());
+
+    // Call 1: set a deadline on the request body.
+    RequestBody requestBody1 = new RequestBody() {
+      @Override public MediaType contentType() {
+        return MediaType.parse("text/plain");
+      }
+      @Override public void writeTo(BufferedSink sink) throws IOException {
+        sink.writeUtf8("abc");
+        sink.timeout().deadline(5, TimeUnit.SECONDS);
+      }
+    };
+    Request request1 = new Request.Builder()
+        .url(server.getUrl("/"))
+        .method("POST", requestBody1)
+        .build();
+    Response response1 = client.newCall(request1).execute();
+    assertEquals(200, response1.code());
+
+    // Call 2: check for the absence of a deadline on the request body.
+    RequestBody requestBody2 = new RequestBody() {
+      @Override public MediaType contentType() {
+        return MediaType.parse("text/plain");
+      }
+      @Override public void writeTo(BufferedSink sink) throws IOException {
+        assertFalse(sink.timeout().hasDeadline());
+        sink.writeUtf8("def");
+      }
+    };
+    Request request2 = new Request.Builder()
+        .url(server.getUrl("/"))
+        .method("POST", requestBody2)
+        .build();
+    Response response2 = client.newCall(request2).execute();
+    assertEquals(200, response2.code());
+
+    // Use sequence numbers to confirm the connection was pooled.
+    assertEquals(0, server.takeRequest().getSequenceNumber());
+    assertEquals(1, server.takeRequest().getSequenceNumber());
+  }
+
+  @Test public void reusedSourcesGetIndependentTimeoutInstances() throws Exception {
+    server.enqueue(new MockResponse().setBody("abc"));
+    server.enqueue(new MockResponse().setBody("def"));
+
+    // Call 1: set a deadline on the response body.
+    Request request1 = new Request.Builder().url(server.getUrl("/")).build();
+    Response response1 = client.newCall(request1).execute();
+    BufferedSource body1 = response1.body().source();
+    assertEquals("abc", body1.readUtf8());
+    body1.timeout().deadline(5, TimeUnit.SECONDS);
+
+    // Call 2: check for the absence of a deadline on the request body.
+    Request request2 = new Request.Builder().url(server.getUrl("/")).build();
+    Response response2 = client.newCall(request2).execute();
+    BufferedSource body2 = response2.body().source();
+    assertEquals("def", body2.readUtf8());
+    assertFalse(body2.timeout().hasDeadline());
+
+    // Use sequence numbers to confirm the connection was pooled.
+    assertEquals(0, server.takeRequest().getSequenceNumber());
+    assertEquals(1, server.takeRequest().getSequenceNumber());
+  }
+
   @Test public void tls() throws Exception {
     server.get().useHttps(sslContext.getSocketFactory(), false);
     server.enqueue(new MockResponse()

okhttp/src/main/java/com/squareup/okhttp/internal/http/HttpConnection.java

@@ -30,6 +30,7 @@
 import okio.Buffer;
 import okio.BufferedSink;
 import okio.BufferedSource;
+import okio.ForwardingTimeout;
 import okio.Okio;
 import okio.Sink;
 import okio.Source;
@@ -265,8 +266,21 @@ public BufferedSource rawSource() {
     return source;
   }
 
+  /**
+   * Sets the delegate of {@code timeout} to {@link Timeout#NONE} and resets its underlying timeout
+   * to the default configuration. Use this to avoid unexpected sharing of timeouts between pooled
+   * connections.
+   */
+  private void detachTimeout(ForwardingTimeout timeout) {
+    Timeout oldDelegate = timeout.delegate();
+    timeout.setDelegate(Timeout.NONE);
+    oldDelegate.clearDeadline();
+    oldDelegate.clearTimeout();
+  }
+
   /** An HTTP body with a fixed length known in advance. */
   private final class FixedLengthSink implements Sink {
+    private final ForwardingTimeout timeout = new ForwardingTimeout(sink.timeout());
     private boolean closed;
     private long bytesRemaining;
 
@@ -275,7 +289,7 @@ private FixedLengthSink(long bytesRemaining) {
     }
 
     @Override public Timeout timeout() {
-      return sink.timeout();
+      return timeout;
     }
 
     @Override public void write(Buffer source, long byteCount) throws IOException {
@@ -298,6 +312,7 @@ private FixedLengthSink(long bytesRemaining) {
       if (closed) return;
       closed = true;
       if (bytesRemaining > 0) throw new ProtocolException("unexpected end of stream");
+      detachTimeout(timeout);
       state = STATE_READ_RESPONSE_HEADERS;
     }
   }
@@ -308,10 +323,11 @@ private FixedLengthSink(long bytesRemaining) {
    * sink with this sink.
    */
   private final class ChunkedSink implements Sink {
+    private final ForwardingTimeout timeout = new ForwardingTimeout(sink.timeout());
     private boolean closed;
 
     @Override public Timeout timeout() {
-      return sink.timeout();
+      return timeout;
     }
 
     @Override public void write(Buffer source, long byteCount) throws IOException {
@@ -333,15 +349,17 @@ private final class ChunkedSink implements Sink {
       if (closed) return;
       closed = true;
       sink.writeUtf8("0\r\n\r\n");
+      detachTimeout(timeout);
       state = STATE_READ_RESPONSE_HEADERS;
     }
   }
 
   private abstract class AbstractSource implements Source {
+    protected final ForwardingTimeout timeout = new ForwardingTimeout(source.timeout());
     protected boolean closed;
 
     @Override public Timeout timeout() {
-      return source.timeout();
+      return timeout;
     }
 
     /**
@@ -351,6 +369,8 @@ private abstract class AbstractSource implements Source {
     protected final void endOfInput(boolean recyclable) throws IOException {
       if (state != STATE_READING_RESPONSE_BODY) throw new IllegalStateException("state: " + state);
 
+      detachTimeout(timeout);
+
       state = STATE_IDLE;
       if (recyclable && onIdle == ON_IDLE_POOL) {
         onIdle = ON_IDLE_HOLD; // Set the on idle policy back to the default.