Add Monitoring Connections for Error Status Messages

As per https://tools.ietf.org/html/rfc2616#section-8.2.2 An
HTTP/1.1 (or later) client sending a message-body SHOULD
monitor the network connection for an error status while it
is transmitting the request.

Currently, okhttp implements a send request, then read response
if/when request was sent successfully.
This hides early responses such as 413 Payload Too Large
errors from clients.

This commit fixes that by adding a response read in case of an
exception happening while the request is being sent.

This closes square#1001 .

okhttp/src/main/kotlin/okhttp3/internal/http/CallServerInterceptor.kt

@@ -21,6 +21,7 @@ import okhttp3.Interceptor
 import okhttp3.Response
 import okhttp3.internal.EMPTY_RESPONSE
 import okio.buffer
+import java.net.SocketException
 
 /** This is the last interceptor in the chain. It makes a network call to the server. */
 class CallServerInterceptor(private val forWebSocket: Boolean) : Interceptor {
@@ -56,7 +57,30 @@ class CallServerInterceptor(private val forWebSocket: Boolean) : Interceptor {
         } else {
           // Write the request body if the "Expect: 100-continue" expectation was met.
           val bufferedRequestBody = exchange.createRequestBody(request, false).buffer()
-          requestBody.writeTo(bufferedRequestBody)
+          try {
+            requestBody.writeTo(bufferedRequestBody)
+          } catch (socketException: SocketException) {
+            // As per https://tools.ietf.org/html/rfc2616#section-8.2.2 it might happen that the server sends an early
+            // response such as 413. Try and collect an early response
+            responseBuilder = exchange.readResponseHeaders(expectContinue = false)!!
+            var response = responseBuilder
+              .request(request)
+              .handshake(exchange.connection.handshake())
+              .sentRequestAtMillis(sentRequestMillis)
+              .receivedResponseAtMillis(System.currentTimeMillis())
+              .build()
+            response = response.newBuilder()
+              .body(exchange.openResponseBody(response))
+              .build()
+
+            val code = response.code
+            if (code in 400..599) {
+              exchange.responseHeadersEnd(response)
+              return response
+            }
+
+            throw socketException
+          }
           bufferedRequestBody.close()
         }
       } else {

okhttp/src/main/kotlin/okhttp3/internal/http1/Http1ExchangeCodec.kt

@@ -170,7 +170,8 @@ class Http1ExchangeCodec(
   }
 
   override fun readResponseHeaders(expectContinue: Boolean): Response.Builder? {
-    check(state == STATE_OPEN_REQUEST_BODY || state == STATE_READ_RESPONSE_HEADERS) {
+    check(state == STATE_OPEN_REQUEST_BODY || state == STATE_READ_RESPONSE_HEADERS ||
+      state == STATE_WRITING_REQUEST_BODY) {
       "state: $state"
     }
 

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

@@ -1118,7 +1118,7 @@ private void writeChunk(BufferedSink sink) throws IOException {
     assertThat(listener.recordedEventTypes()).containsExactly(
         "CallStart", "ProxySelectStart", "ProxySelectEnd", "DnsStart", "DnsEnd", "ConnectStart",
         "ConnectEnd", "ConnectionAcquired", "RequestHeadersStart", "RequestHeadersEnd",
-        "RequestBodyStart", "RequestFailed", "ConnectionReleased", "CallFailed");
+        "RequestBodyStart", "RequestFailed", "ResponseFailed", "ConnectionReleased", "CallFailed");
   }
 
   @Test public void requestBodySuccessHttp1OverHttps() throws IOException {

okhttp/src/test/java/okhttp3/internal/http/ThreadInterruptTest.java

@@ -17,12 +17,19 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
 import okhttp3.Call;
 import okhttp3.DelegatingServerSocketFactory;
 import okhttp3.DelegatingSocketFactory;
@@ -42,6 +49,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 public final class ThreadInterruptTest {
@@ -139,6 +147,47 @@ protected Socket configureSocket(Socket socket) throws IOException {
     responseBody.close();
   }
 
+  @Test public void readStatusForInterruptedRequest() throws Exception {
+    HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+    server.createContext("/", new HttpHandler() {
+      final byte[] buffer = new byte[1024];
+
+      @Override
+      public void handle(HttpExchange exchange) throws IOException {
+        InputStream inBody = exchange.getRequestBody();
+        for (int i = 0; i < 10; i++) {
+          //noinspection ResultOfMethodCallIgnored
+          inBody.read(buffer);
+        }
+        inBody.close();
+        OutputStream outBody = exchange.getResponseBody();
+        exchange.sendResponseHeaders(413, 65);
+        outBody.write("{\"error\":\"too_large\",\"reason\":\"the request entity is too large\"}\r\n".getBytes());
+
+        outBody.flush();
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
+        outBody.close();
+      }
+    });
+    server.setExecutor(Executors.newSingleThreadExecutor());
+    server.start();
+
+    int requestBodySize = 20 * 1024 * 1024; // 20 MiB
+
+    OkHttpClient client = new OkHttpClient();
+    Response response = client.newCall(
+            new Request.Builder()
+                    .url(String.format("http://localhost:%d/", server.getAddress().getPort()))
+                    .post(RequestBody.create(new byte[requestBodySize]))
+                    .build()
+    ).execute();
+    assertEquals(413, response.code());
+  }
+
   private void sleep(int delayMillis) {
     try {
       Thread.sleep(delayMillis);