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-tests/src/test/java/okhttp3/internal/http/DisconnectTest.java

@@ -19,22 +19,33 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 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.DelegatingServerSocketFactory;
 import okhttp3.DelegatingSocketFactory;
+import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
 import okhttp3.OkUrlFactory;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
 import okio.Buffer;
 import org.junit.Before;
 import org.junit.Test;
 
 import static okhttp3.TestUtil.defaultClient;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 public final class DisconnectTest {
@@ -117,6 +128,46 @@ public final class DisconnectTest {
     responseBody.close();
   }
 
+  @Test public void readStatusForInterruptedRequest() throws Exception {
+    HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+    server.createContext("/", new HttpHandler() {
+      byte[] buffer = new byte[1024];
+
+      @Override
+      public void handle(HttpExchange exchange) throws IOException {
+        InputStream inBody = exchange.getRequestBody();
+        for(int i = 0; i < 10; i++) {
+          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(MediaType.get("text/plain"), new byte[requestBodySize]))
+                    .build()
+    ).execute();
+    assertEquals(413, response.code());
+  }
+
   private void disconnectLater(final HttpURLConnection connection, final int delayMillis) {
     Thread interruptingCow = new Thread() {
       @Override public void run() {

okhttp/src/main/java/okhttp3/internal/http/CallServerInterceptor.java

@@ -69,7 +69,29 @@ public CallServerInterceptor(boolean forWebSocket) {
             new CountingSink(httpCodec.createRequestBody(request, contentLength));
         BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
 
-        request.body().writeTo(bufferedRequestBody);
+        try {
+          request.body().writeTo(bufferedRequestBody);
+        } catch (IOException ioException) {
+          // 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 = httpCodec.readResponseHeaders(false);
+          Response response = responseBuilder
+                  .request(request)
+                  .handshake(streamAllocation.connection().handshake())
+                  .sentRequestAtMillis(sentRequestMillis)
+                  .receivedResponseAtMillis(System.currentTimeMillis())
+                  .build();
+          response = response.newBuilder().body(httpCodec.openResponseBody(response)).build();
+
+          int code = response.code();
+          if (code >= 400 && code < 500) {
+            realChain.eventListener()
+                    .responseHeadersEnd(realChain.call(), response);
+            return response;
+          }
+
+          throw ioException;
+        }
         bufferedRequestBody.close();
         realChain.eventListener()
             .requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);

okhttp/src/main/java/okhttp3/internal/http/RealInterceptorChain.java

@@ -157,7 +157,7 @@ public Response proceed(Request request, StreamAllocation streamAllocation, Http
       throw new NullPointerException("interceptor " + interceptor + " returned null");
     }
 
-    if (response.body() == null) {
+    if ((response.code() < 400 || response.code() >= 500) && response.body() == null) {
       throw new IllegalStateException(
           "interceptor " + interceptor + " returned a response with no body");
     }

okhttp/src/main/java/okhttp3/internal/http1/Http1Codec.java

@@ -181,7 +181,9 @@ public void writeRequest(Headers headers, String requestLine) throws IOException
   }
 
   @Override public Response.Builder readResponseHeaders(boolean expectContinue) throws IOException {
-    if (state != STATE_OPEN_REQUEST_BODY && state != STATE_READ_RESPONSE_HEADERS) {
+    if (state != STATE_WRITING_REQUEST_BODY
+                    && state != STATE_OPEN_REQUEST_BODY
+                    && state != STATE_READ_RESPONSE_HEADERS) {
       throw new IllegalStateException("state: " + state);
     }