java.net.HttpURLConnection#setChunkedStreamingMode ( )源码实例Demo

@Test
public void testChunkAggregatedUpload() throws IOException {
    //create a random file to be uploaded.
    int size = 69 * 1024;
    File fname = new File(tmpFolder, "testChunkAggregatedUpload.txt");
    fname.createNewFile();
    RandomAccessFile randf = new RandomAccessFile(fname, "rw");
    randf.setLength(size);
    randf.close();

    //test chunked upload
    HttpURLConnection urlConn = request("/test/v1/aggregate/upload", HttpMethod.PUT);
    urlConn.setChunkedStreamingMode(1024);
    Files.copy(Paths.get(fname.toURI()), urlConn.getOutputStream());
    assertEquals(200, urlConn.getResponseCode());

    assertEquals(size, Integer.parseInt(getContent(urlConn).split(":")[1].trim()));
    urlConn.disconnect();
    fname.delete();
}
 

@Test
public void shouldRejectRequestsWithBodyMoreThanContentLength() throws IOException {
    startServer();

    final HttpURLConnection conn = startJsonPostRequest();
    final byte[] bodyBytes = JSON_MAPPER.writeValueAsBytes(EMPTY_JSON_OBJECT);
    // Explicitly lie about the length of the content we're sending.
    conn.setChunkedStreamingMode(1);
    conn.setRequestProperty("content-length", String.valueOf(bodyBytes.length - 1));
    // Sanity check that the underlying implementation didn't discard our content length.
    Assume.assumeTrue("Cannot manipulate content-length headers; try setting 'sun.net.http.allowRestrictedHeaders'.",
            null != conn.getRequestProperty("content-length"));
    try (final OutputStream requestBody = conn.getOutputStream()) {
        requestBody.write(bodyBytes);
    }

    assertEquals(HTTP_BAD_REQUEST, conn.getResponseCode());
}
 

protected OutputStream createOutputStream(Message message,
                                          boolean needToCacheRequest,
                                          boolean isChunking,
                                          int chunkThreshold) throws IOException {
    HttpURLConnection connection = (HttpURLConnection)message.get(KEY_HTTP_CONNECTION);

    if (isChunking && chunkThreshold <= 0) {
        chunkThreshold = 0;
        connection.setChunkedStreamingMode(-1);
    }
    try {
        return new URLConnectionWrappedOutputStream(message, connection,
                                       needToCacheRequest,
                                       isChunking,
                                       chunkThreshold,
                                       getConduitName());
    } catch (URISyntaxException e) {
        throw new IOException(e);
    }
}
 

private static void doOutput(final HttpURLConnection urlConnection, final RequestBody body)
        throws IOException {
    urlConnection.setDoOutput(true);
    final String contentType = body.contentType();
    if (contentType != null) {
        urlConnection.addRequestProperty("Content-Type", contentType);
    }
    final long contentLength = body.contentLength();
    if (contentLength > 0) {
        setFixedLengthStreamingMode(urlConnection, contentLength);
    } else {
        urlConnection.setChunkedStreamingMode(0);
    }
    BufferedOutputStream os = null;
    try {
        os = new BufferedOutputStream(urlConnection.getOutputStream());
        body.writeTo(os);
    } finally {
        Utils.closeQuietly(os);
    }
}
 

private static void doOutput(final HttpURLConnection urlConnection, final RequestBody body)
        throws IOException {
    urlConnection.setDoOutput(true);
    final String contentType = body.contentType().toString();
    if (contentType != null) {
        urlConnection.addRequestProperty("Content-Type", contentType);
    }
    final long contentLength = body.contentLength();
    if (contentLength > 0) {
        setFixedLengthStreamingMode(urlConnection, contentLength);
        urlConnection.addRequestProperty("Content-Length", Long.toString(contentLength));
    } else {
        urlConnection.setChunkedStreamingMode(0);
    }
    BufferedOutputStream os = null;
    try {
        os = new BufferedOutputStream(urlConnection.getOutputStream());
        body.writeTo(os);
    } finally {
        Util.closeQuietly(os);
    }
}
 

@Test
public void testBug49424WithChunking() throws Exception {
    Tomcat tomcat = getTomcatInstance();
    Context root = tomcat.addContext("",
            System.getProperty("java.io.tmpdir"));
    Tomcat.addServlet(root, "Bug37794", new Bug37794Servlet());
    root.addServletMappingDecoded("/", "Bug37794");
    tomcat.start();

    HttpURLConnection conn = getConnection("http://localhost:" + getPort() + "/");
    conn.setChunkedStreamingMode(8 * 1024);
    InputStream is = conn.getInputStream();
    Assert.assertNotNull(is);
}
 

@Test
public void shouldRejectRequestsWithTooLargeBody() throws IOException {
    startServer();

    final HttpURLConnection conn = startJsonPostRequest();
    // Here we're not declaring ahead of time that we will be too large. But the body will be.
    conn.setChunkedStreamingMode(128);
    try (final OutputStream requestBody = conn.getOutputStream()) {
        JSON_MAPPER.writeValue(requestBody, buildBigJsonPayload(JSON_MAXIMUM_BODY_SIZE + 1));
    }

    assertEquals(HTTP_ENTITY_TOO_LARGE, conn.getResponseCode());
}
 

void test(String method) throws Exception {
    ss = new ServerSocket(0);
    ss.setSoTimeout(ACCEPT_TIMEOUT);
    int port = ss.getLocalPort();

    Thread otherThread = new Thread(this);
    otherThread.start();

    try {
        URL url = new URL("http://localhost:" + port + "/");
        HttpURLConnection uc = (HttpURLConnection) url.openConnection();
        uc.setDoOutput(true);
        if (method != null)
            uc.setRequestMethod(method);
        uc.setChunkedStreamingMode(4096);
        OutputStream os = uc.getOutputStream();
        os.write("Hello there".getBytes());

        InputStream is = uc.getInputStream();
        is.close();
    } catch (IOException expected) {
        //expected.printStackTrace();
    } finally {
        ss.close();
        otherThread.join();
    }
}
 

@Override
protected void writeBody(HttpURLConnection connection, ProgressListener listener) {
    try {
        connection.setChunkedStreamingMode(0);
        connection.setDoOutput(true);
        this.outputStream = connection.getOutputStream();

        for (Map.Entry<String, String> entry : this.fields.entrySet()) {
            this.writePartHeader(new String[][] {{"name", entry.getKey()}});
            this.writeOutput(entry.getValue());
        }

        this.writePartHeader(new String[][] {{"name", "file"}, {"filename", this.filename}},
            "application/octet-stream");

        OutputStream fileContentsOutputStream = this.outputStream;
        if (listener != null) {
            fileContentsOutputStream = new ProgressOutputStream(this.outputStream, listener, this.fileSize);
        }
        if (this.inputStream != null) {
            byte[] buffer = new byte[BUFFER_SIZE];
            int n = this.inputStream.read(buffer);
            while (n != -1) {
                fileContentsOutputStream.write(buffer, 0, n);
                n = this.inputStream.read(buffer);
            }
        } else {
            this.callback.writeToStream(this.outputStream);
        }

        if (LOGGER.isLoggable(Level.FINE)) {
            this.loggedRequest.append("<File Contents Omitted>");
        }

        this.writeBoundary();
        this.writeOutput("--");
    } catch (IOException e) {
        throw new BoxAPIException("Couldn't connect to the Box API due to a network error.", e);
    }
}
 

void test(String method) throws Exception {
    ss = new ServerSocket(0);
    ss.setSoTimeout(ACCEPT_TIMEOUT);
    int port = ss.getLocalPort();

    Thread otherThread = new Thread(this);
    otherThread.start();

    try {
        URL url = new URL("http://localhost:" + port + "/");
        HttpURLConnection uc = (HttpURLConnection) url.openConnection();
        uc.setDoOutput(true);
        if (method != null)
            uc.setRequestMethod(method);
        uc.setChunkedStreamingMode(4096);
        OutputStream os = uc.getOutputStream();
        os.write("Hello there".getBytes());

        InputStream is = uc.getInputStream();
        is.close();
    } catch (IOException expected) {
        //expected.printStackTrace();
    } finally {
        ss.close();
        otherThread.join();
    }
}
 

void test(String method) throws Exception {
    ss = new ServerSocket(0);
    ss.setSoTimeout(ACCEPT_TIMEOUT);
    int port = ss.getLocalPort();

    Thread otherThread = new Thread(this);
    otherThread.start();

    try {
        URL url = new URL("http://localhost:" + port + "/");
        HttpURLConnection uc = (HttpURLConnection) url.openConnection();
        uc.setDoOutput(true);
        if (method != null)
            uc.setRequestMethod(method);
        uc.setChunkedStreamingMode(4096);
        OutputStream os = uc.getOutputStream();
        os.write("Hello there".getBytes());

        InputStream is = uc.getInputStream();
        is.close();
    } catch (IOException expected) {
        //expected.printStackTrace();
    } finally {
        ss.close();
        otherThread.join();
    }
}
 

@Test
public void testBug49424WithChunking() throws Exception {
    Tomcat tomcat = getTomcatInstance();
    // No file system docBase required
    Context root = tomcat.addContext("", null);
    Tomcat.addServlet(root, "Bug37794", new Bug37794Servlet());
    root.addServletMapping("/", "Bug37794");
    tomcat.start();

    HttpURLConnection conn = getConnection("http://localhost:" + getPort() + "/");
    conn.setChunkedStreamingMode(8 * 1024);
    InputStream is = conn.getInputStream();
    assertNotNull(is);
}
 

static void setContentTypeAndLengthForStreaming(
        HttpURLConnection connection,
        Request request,
        boolean compress) {

    long length;

    if (request.payload instanceof File) {
        length = compress ? -1L : ((File) request.payload).length();
    } else if (request.payload instanceof InputStream) {
        length = -1L;
    } else {
        throw new IllegalStateException();
    }

    if (length > Integer.MAX_VALUE) {
        length = -1L; // use chunked streaming mode
    }

    WebbUtils.ensureRequestProperty(connection, Const.HDR_CONTENT_TYPE, Const.APP_BINARY);
    if (length < 0) {
        connection.setChunkedStreamingMode(-1); // use default chunk size
        if (compress) {
            connection.setRequestProperty(Const.HDR_CONTENT_ENCODING, "gzip");
        }
    } else {
        connection.setFixedLengthStreamingMode((int) length);
    }
}
 

static void setContentTypeAndLengthForStreaming(final HttpURLConnection connection, final Request request,
		final boolean compress) {

	long length;

	if (request.payload instanceof File) {
		length = compress ? -1L : ((File) request.payload).length();
	} else if (request.payload instanceof InputStream) {
		length = -1L;
	} else {
		throw new IllegalStateException();
	}

	if (length > Integer.MAX_VALUE) {
		length = -1L; // use chunked streaming mode
	}

	WebbUtils.ensureRequestProperty(connection, Webb.HDR_CONTENT_TYPE, Webb.APP_BINARY);
	if (length < 0) {
		connection.setChunkedStreamingMode(-1); // use default chunk size
		if (compress) {
			connection.setRequestProperty(Webb.HDR_CONTENT_ENCODING, "gzip");
		}
	} else {
		connection.setFixedLengthStreamingMode((int) length);
	}
}
 

public static void streamWriter(HttpURLConnection urlc, OutputStreamWriter writer) throws IOException
{
    urlc.setRequestMethod("POST");
    urlc.setDoInput(true);
    urlc.setDoOutput(true);
    urlc.setChunkedStreamingMode(2048);
    OutputStream outputStream = urlc.getOutputStream();
    writeStuff(outputStream, writer);
    outputStream.close();
}
 

public static int postUrlWithDisconnect(boolean stream, BytesStreamer streamer, String path,
        Map<String, List<String>> reqHead, Map<String, List<String>> resHead) throws IOException {

    URL url = new URL(path);
    HttpURLConnection connection = (HttpURLConnection) url.openConnection();
    connection.setDoOutput(true);
    connection.setReadTimeout(1000000);
    if (reqHead != null) {
        for (Map.Entry<String, List<String>> entry : reqHead.entrySet()) {
            StringBuilder valueList = new StringBuilder();
            for (String value : entry.getValue()) {
                if (valueList.length() > 0) {
                    valueList.append(',');
                }
                valueList.append(value);
            }
            connection.setRequestProperty(entry.getKey(), valueList.toString());
        }
    }
    if (streamer != null && stream) {
        if (streamer.getLength() > 0) {
            connection.setFixedLengthStreamingMode(streamer.getLength());
        } else {
            connection.setChunkedStreamingMode(1024);
        }
    }

    connection.connect();

    // Write the request body
    try (OutputStream os = connection.getOutputStream()) {
        while (streamer != null && streamer.available() > 0) {
            byte[] next = streamer.next();
            os.write(next);
            os.flush();
        }
    }

    int rc = connection.getResponseCode();
    if (resHead != null) {
        Map<String, List<String>> head = connection.getHeaderFields();
        resHead.putAll(head);
    }
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {

    }
    if (rc == HttpServletResponse.SC_OK) {
        connection.getInputStream().close();
        connection.disconnect();
    }
    return rc;
}
 

public static void main(String[] args) throws Exception {

        HTTPServer server = new HTTPServer();
        server.start();
        int port = server.getPort();
        out.println("Server listening on " + port);


        URL url = new URL("http://localhost:" + port);
        HttpURLConnection conn = (HttpURLConnection)url.openConnection();
        conn.setRequestMethod("POST");
        conn.setDoOutput(true);
        conn.setChunkedStreamingMode(1024);

        out.println("sending " + TOTAL_BYTES + " bytes");

        int byteAtOnce;
        int sendingBytes = TOTAL_BYTES;
        byte[] buffer = getBuffer(BUFFER_SIZE);
        try (OutputStream toServer = conn.getOutputStream()) {
            while (sendingBytes > 0) {
                if (sendingBytes > BUFFER_SIZE) {
                    byteAtOnce = BUFFER_SIZE;
                } else {
                    byteAtOnce = sendingBytes;
                }
                toServer.write(buffer, 0, byteAtOnce);
                sendingBytes -= byteAtOnce;
                out.print((TOTAL_BYTES - sendingBytes) + " was sent. ");
                toServer.flush();
                // gives the server thread time to read, and eventually close;
                Thread.sleep(500);
            }
        } catch (IOException expected) {
            // Expected IOException due to server.close()
            out.println("PASSED. Caught expected: " + expected);
            return;
        }

        // Expected IOException not received. FAIL
        throw new RuntimeException("Failed: Expected IOException not received");
    }
 

@Override
public Connection connect(Request request) throws IOException {
    URL url = new URL(request.url().toString(true));
    Proxy proxy = request.proxy();
    HttpURLConnection connection = open(url, proxy);

    connection.setConnectTimeout(request.connectTimeout());
    connection.setReadTimeout(request.readTimeout());
    connection.setInstanceFollowRedirects(false);

    if (connection instanceof HttpsURLConnection) {
        SSLSocketFactory sslSocketFactory = request.sslSocketFactory();
        if (sslSocketFactory != null)
            ((HttpsURLConnection) connection).setSSLSocketFactory(sslSocketFactory);
        HostnameVerifier hostnameVerifier = request.hostnameVerifier();
        if (hostnameVerifier != null)
            ((HttpsURLConnection) connection).setHostnameVerifier(hostnameVerifier);
    }

    RequestMethod method = request.method();
    connection.setRequestMethod(method.toString());
    connection.setDoInput(true);
    boolean isAllowBody = method.allowBody();
    connection.setDoOutput(isAllowBody);

    Headers headers = request.headers();

    if (isAllowBody) {
        long contentLength = headers.getContentLength();
        if (contentLength <= Integer.MAX_VALUE) connection.setFixedLengthStreamingMode((int) contentLength);
        else if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.KITKAT) connection.setFixedLengthStreamingMode(contentLength);
        else connection.setChunkedStreamingMode(256 * 1024);
    }

    Map<String, String> requestHeaders = Headers.getRequestHeaders(headers);
    for (Map.Entry<String, String> headerEntry : requestHeaders.entrySet()) {
        String headKey = headerEntry.getKey();
        String headValue = headerEntry.getValue();
        connection.setRequestProperty(headKey, headValue);
    }

    connection.connect();
    return new URLConnection(connection);
}
 

@VisibleForTesting
LowLevelHttpResponse execute(final OutputWriter outputWriter) throws IOException {
  HttpURLConnection connection = this.connection;
  // write content
  if (getStreamingContent() != null) {
    String contentType = getContentType();
    if (contentType != null) {
      addHeader("Content-Type", contentType);
    }
    String contentEncoding = getContentEncoding();
    if (contentEncoding != null) {
      addHeader("Content-Encoding", contentEncoding);
    }
    long contentLength = getContentLength();
    if (contentLength >= 0) {
      connection.setRequestProperty("Content-Length", Long.toString(contentLength));
    }
    String requestMethod = connection.getRequestMethod();
    if ("POST".equals(requestMethod) || "PUT".equals(requestMethod)) {
      connection.setDoOutput(true);
      // see http://developer.android.com/reference/java/net/HttpURLConnection.html
      if (contentLength >= 0 && contentLength <= Integer.MAX_VALUE) {
        connection.setFixedLengthStreamingMode((int) contentLength);
      } else {
        connection.setChunkedStreamingMode(0);
      }
      final OutputStream out = connection.getOutputStream();

      boolean threw = true;
      try {
        writeContentToOutputStream(outputWriter, out);

        threw = false;
      } catch (IOException e) {
        // If we've gotten a response back, continue on and try to parse the response. Otherwise,
        // re-throw the IOException
        if (!hasResponse(connection)) {
          throw e;
        }
      } finally {
        try {
          out.close();
        } catch (IOException exception) {
          // When writeTo() throws an exception, chances are that the close call will also fail.
          // In such case, swallow exception from close call so that the underlying cause can
          // propagate.
          if (!threw) {
            throw exception;
          }
        }
      }
    } else {
      // cannot call setDoOutput(true) because it would change a GET method to POST
      // for HEAD, OPTIONS, DELETE, or TRACE it would throw an exceptions
      Preconditions.checkArgument(
          contentLength == 0, "%s with non-zero content length is not supported", requestMethod);
    }
  }
  // connect
  boolean successfulConnection = false;
  try {
    connection.connect();
    NetHttpResponse response = new NetHttpResponse(connection);
    successfulConnection = true;
    return response;
  } finally {
    if (!successfulConnection) {
      connection.disconnect();
    }
  }
}
 

HttpURLConnection convertAndSend(Request request, Options options) throws IOException {
  final URL url = new URL(request.url());
  final HttpURLConnection connection = this.getConnection(url);
  if (connection instanceof HttpsURLConnection) {
    HttpsURLConnection sslCon = (HttpsURLConnection) connection;
    if (sslContextFactory != null) {
      sslCon.setSSLSocketFactory(sslContextFactory);
    }
    if (hostnameVerifier != null) {
      sslCon.setHostnameVerifier(hostnameVerifier);
    }
  }
  connection.setConnectTimeout(options.connectTimeoutMillis());
  connection.setReadTimeout(options.readTimeoutMillis());
  connection.setAllowUserInteraction(false);
  connection.setInstanceFollowRedirects(options.isFollowRedirects());
  connection.setRequestMethod(request.httpMethod().name());

  Collection<String> contentEncodingValues = request.headers().get(CONTENT_ENCODING);
  boolean gzipEncodedRequest =
      contentEncodingValues != null && contentEncodingValues.contains(ENCODING_GZIP);
  boolean deflateEncodedRequest =
      contentEncodingValues != null && contentEncodingValues.contains(ENCODING_DEFLATE);

  boolean hasAcceptHeader = false;
  Integer contentLength = null;
  for (String field : request.headers().keySet()) {
    if (field.equalsIgnoreCase("Accept")) {
      hasAcceptHeader = true;
    }
    for (String value : request.headers().get(field)) {
      if (field.equals(CONTENT_LENGTH)) {
        if (!gzipEncodedRequest && !deflateEncodedRequest) {
          contentLength = Integer.valueOf(value);
          connection.addRequestProperty(field, value);
        }
      } else {
        connection.addRequestProperty(field, value);
      }
    }
  }
  // Some servers choke on the default accept string.
  if (!hasAcceptHeader) {
    connection.addRequestProperty("Accept", "*/*");
  }

  if (request.body() != null) {
    if (disableRequestBuffering) {
      if (contentLength != null) {
        connection.setFixedLengthStreamingMode(contentLength);
      } else {
        connection.setChunkedStreamingMode(8196);
      }
    }
    connection.setDoOutput(true);
    OutputStream out = connection.getOutputStream();
    if (gzipEncodedRequest) {
      out = new GZIPOutputStream(out);
    } else if (deflateEncodedRequest) {
      out = new DeflaterOutputStream(out);
    }
    try {
      out.write(request.body());
    } finally {
      try {
        out.close();
      } catch (IOException suppressed) { // NOPMD
      }
    }
  }
  return connection;
}