Don't report connectionAcquired until actually connected. (square#3566)

Currently our implementation acquires the connection early so that we have
something to close if the call is canceled. Event listeners are simpler if
they only get a connectionAcquired event once the connection has been
actually established.

Author: swankjesse

okhttp-tests/src/test/java/okhttp3/ConnectionPoolTest.java

@@ -85,7 +85,7 @@ public final class ConnectionPoolTest {
     synchronized (pool) {
       StreamAllocation streamAllocation = new StreamAllocation(pool, addressA, null,
           EventListener.NONE, null);
-      streamAllocation.acquire(c1);
+      streamAllocation.acquire(c1, true);
     }
 
     // Running at time 50, the pool returns that nothing can be evicted until time 150.
@@ -179,7 +179,7 @@ private void allocateAndLeakAllocation(ConnectionPool pool, RealConnection conne
     synchronized (pool) {
       StreamAllocation leak = new StreamAllocation(pool, connection.route().address(), null,
           EventListener.NONE, null);
-      leak.acquire(connection);
+      leak.acquire(connection, true);
     }
   }
 

okhttp-tests/src/test/java/okhttp3/EventListenerTest.java

@@ -113,7 +113,7 @@ public final class EventListenerTest {
     response.body().close();
 
     List<String> expectedEvents = Arrays.asList("CallStart", "DnsStart", "DnsEnd",
-        "ConnectionAcquired", "ConnectStart", "ConnectEnd", "RequestHeadersStart",
+        "ConnectStart", "ConnectEnd", "ConnectionAcquired", "RequestHeadersStart",
         "RequestHeadersEnd", "ResponseHeadersStart", "ResponseHeadersEnd", "ResponseBodyStart",
         "ResponseBodyEnd", "ConnectionReleased", "CallEnd");
     assertEquals(expectedEvents, listener.recordedEventTypes());
@@ -144,7 +144,7 @@ public final class EventListenerTest {
     completionLatch.await();
 
     List<String> expectedEvents = Arrays.asList("CallStart", "DnsStart", "DnsEnd",
-        "ConnectionAcquired", "ConnectStart", "ConnectEnd", "RequestHeadersStart",
+        "ConnectStart", "ConnectEnd", "ConnectionAcquired", "RequestHeadersStart",
         "RequestHeadersEnd", "ResponseHeadersStart", "ResponseHeadersEnd", "ResponseBodyStart",
         "ResponseBodyEnd", "ConnectionReleased", "CallEnd");
     assertEquals(expectedEvents, listener.recordedEventTypes());
@@ -166,7 +166,7 @@ public final class EventListenerTest {
     }
 
     List<String> expectedEvents = Arrays.asList("CallStart", "DnsStart", "DnsEnd",
-        "ConnectionAcquired", "ConnectStart", "ConnectEnd", "RequestHeadersStart",
+        "ConnectStart", "ConnectEnd", "ConnectionAcquired", "RequestHeadersStart",
         "RequestHeadersEnd", "ResponseHeadersStart", "ConnectionReleased", "CallFailed");
     assertEquals(expectedEvents, listener.recordedEventTypes());
   }
@@ -198,8 +198,8 @@ private void assertSuccessfulEventOrder(Matcher<Response> responseMatcher) throw
 
     assumeThat(response, responseMatcher);
 
-    List<String> expectedEvents = asList("CallStart", "DnsStart", "DnsEnd", "ConnectionAcquired",
-        "ConnectStart", "SecureConnectStart", "SecureConnectEnd", "ConnectEnd",
+    List<String> expectedEvents = asList("CallStart", "DnsStart", "DnsEnd", "ConnectStart",
+        "SecureConnectStart", "SecureConnectEnd", "ConnectEnd", "ConnectionAcquired",
         "RequestHeadersStart", "RequestHeadersEnd", "ResponseHeadersStart", "ResponseHeadersEnd",
         "ResponseBodyStart", "ResponseBodyEnd", "ConnectionReleased", "CallEnd");
 
@@ -239,7 +239,8 @@ private void assertBytesReadWritten(RecordingEventListener listener,
       RequestHeadersEnd responseHeadersEnd = listener.removeUpToEvent(RequestHeadersEnd.class);
       assertThat("request header length", responseHeadersEnd.headerLength, requestHeaderLength);
     } else {
-      assertFalse("Found RequestHeadersEnd", listener.recordedEventTypes().contains("RequestHeadersEnd"));
+      assertFalse("Found RequestHeadersEnd",
+          listener.recordedEventTypes().contains("RequestHeadersEnd"));
     }
 
     if (requestBodyBytes != null) {
@@ -253,14 +254,16 @@ private void assertBytesReadWritten(RecordingEventListener listener,
       ResponseHeadersEnd responseHeadersEnd = listener.removeUpToEvent(ResponseHeadersEnd.class);
       assertThat("response header length", responseHeadersEnd.headerLength, responseHeaderLength);
     } else {
-      assertFalse("Found ResponseHeadersEnd", listener.recordedEventTypes().contains("ResponseHeadersEnd"));
+      assertFalse("Found ResponseHeadersEnd",
+          listener.recordedEventTypes().contains("ResponseHeadersEnd"));
     }
 
     if (responseBodyBytes != null) {
       ResponseBodyEnd responseBodyEnd = listener.removeUpToEvent(ResponseBodyEnd.class);
       assertThat("response body bytes", responseBodyEnd.bytesRead, responseBodyBytes);
     } else {
-      assertFalse("Found ResponseBodyEnd", listener.recordedEventTypes().contains("ResponseBodyEnd"));
+      assertFalse("Found ResponseBodyEnd",
+          listener.recordedEventTypes().contains("ResponseBodyEnd"));
     }
   }
 

okhttp/src/main/java/okhttp3/ConnectionPool.java

@@ -123,7 +123,7 @@ public synchronized int connectionCount() {
     assert (Thread.holdsLock(this));
     for (RealConnection connection : connections) {
       if (connection.isEligible(address, route)) {
-        streamAllocation.acquire(connection);
+        streamAllocation.acquire(connection, true);
         return connection;
       }
     }

okhttp/src/main/java/okhttp3/internal/connection/StreamAllocation.java

@@ -87,6 +87,7 @@ public final class StreamAllocation {
   private final RouteSelector routeSelector;
   private int refusedStreamCount;
   private RealConnection connection;
+  private boolean reportedAcquired;
   private boolean released;
   private boolean canceled;
   private HttpCodec codec;
@@ -176,6 +177,10 @@ private RealConnection findConnection(int connectTimeout, int readTimeout, int w
         result = this.connection;
         releasedConnection = null;
       }
+      if (!reportedAcquired) {
+        // If the connection was never reported acquired, don't report it as released!
+        releasedConnection = null;
+      }
 
       if (result == null) {
         // Attempt to get a connection from the pool.
@@ -237,15 +242,13 @@ private RealConnection findConnection(int connectTimeout, int readTimeout, int w
         route = selectedRoute;
         refusedStreamCount = 0;
         result = new RealConnection(connectionPool, selectedRoute);
-        acquire(result);
+        acquire(result, false);
       }
     }
 
-    // We have a connection. Either a connected one from the pool, or one we need to connect.
-    eventListener.connectionAcquired(call, result);
-
     // If we found a pooled connection on the 2nd time around, we're done.
     if (foundPooledConnection) {
+      eventListener.connectionAcquired(call, result);
       return result;
     }
 
@@ -256,6 +259,8 @@ private RealConnection findConnection(int connectTimeout, int readTimeout, int w
 
     Socket socket = null;
     synchronized (connectionPool) {
+      reportedAcquired = true;
+
       // Pool the connection.
       Internal.instance.put(connectionPool, result);
 
@@ -268,6 +273,7 @@ private RealConnection findConnection(int connectTimeout, int readTimeout, int w
     }
     closeQuietly(socket);
 
+    eventListener.connectionAcquired(call, result);
     return result;
   }
 
@@ -440,7 +446,7 @@ public void streamFailed(IOException e) {
       }
       releasedConnection = connection;
       socket = deallocate(noNewStreams, false, true);
-      if (connection != null) releasedConnection = null;
+      if (connection != null || !reportedAcquired) releasedConnection = null;
     }
 
     closeQuietly(socket);
@@ -453,11 +459,12 @@ public void streamFailed(IOException e) {
    * Use this allocation to hold {@code connection}. Each call to this must be paired with a call to
    * {@link #release} on the same connection.
    */
-  public void acquire(RealConnection connection) {
+  public void acquire(RealConnection connection, boolean reportedAcquired) {
     assert (Thread.holdsLock(connectionPool));
     if (this.connection != null) throw new IllegalStateException();
 
     this.connection = connection;
+    this.reportedAcquired = reportedAcquired;
     connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
   }