BlockingStreamingHttp[Response|Response] to support InputStream setter (

#1175) Motivation: BlockingStreamingHttp[Response|Response] allow a user to get the payload body as an InputStream. However the user is not able to set the payload body as an InputStream. This requires manual conversion from the user perspective and raises the bar to entry. Modifications: - BlockingStreamingHttp[Response|Response] supports an additional method to set the payloadBody as an InputStream Result: BlockingStreamingHttp[Response|Response] APIs are more InputStream friendly.
apple · Oct 12, 2020 · a323b86 · a323b86
1 parent db7d8fb
commit a323b86
Show file tree

Hide file tree

Showing 5 changed files with 97 additions and 16 deletions.
diff --git a/servicetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingStreamingHttpRequest.java b/servicetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingStreamingHttpRequest.java
@@ -57,12 +57,12 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpRequest} with its underlying payload set to {@code payloadBody}.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link Iterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
      * control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link Iterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body.
      * @return {@code this}
      */
@@ -71,26 +71,40 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpRequest} with its underlying payload set to {@code payloadBody}.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link CloseableIterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
      * control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link CloseableIterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body.
      * @return {@code this}
      */
     BlockingStreamingHttpRequest payloadBody(CloseableIterable<Buffer> payloadBody);
 
+    /**
+     * Returns a {@link BlockingStreamingHttpRequest} with its underlying payload set to {@code payloadBody}.
+     * <p>
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
+     * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
+     * control.
+     * <p>
+     * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
+     * combination with the existing payload body that is being replaced.
+     * @param payloadBody The new payload body.
+     * @return {@code this}
+     */
+    BlockingStreamingHttpRequest payloadBody(InputStream payloadBody);
+
     /**
      * Returns a {@link BlockingStreamingHttpRequest} with its underlying payload set to the result of serialization.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link Iterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(Function, HttpSerializer)} for more
      * fine grain control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link Iterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body, prior to serialization.
      * @param serializer Used to serialize the payload body.
      * @param <T> The type of objects to serialize.
@@ -101,12 +115,12 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpRequest} with its underlying payload set to the result of serialization.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link CloseableIterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(Function, HttpSerializer)} for more
      * fine grain control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link CloseableIterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body, prior to serialization.
      * @param serializer Used to serialize the payload body.
      * @param <T> The type of objects to serialize.

diff --git a/...icetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingStreamingHttpResponse.java b/...icetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingStreamingHttpResponse.java
@@ -57,12 +57,12 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpResponse} with its underlying payload set to {@code payloadBody}.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link Iterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
      * control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link Iterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body.
      * @return {@code this}
      */
@@ -71,26 +71,40 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpResponse} with its underlying payload set to {@code payloadBody}.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link CloseableIterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
      * control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link CloseableIterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body.
      * @return {@code this}
      */
     BlockingStreamingHttpResponse payloadBody(CloseableIterable<Buffer> payloadBody);
 
+    /**
+     * Returns a {@link BlockingStreamingHttpResponse} with its underlying payload set to {@code payloadBody}.
+     * <p>
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
+     * default policy is not sufficient you can use {@link #transformPayloadBody(UnaryOperator)} for more fine grain
+     * control.
+     * <p>
+     * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
+     * combination with the existing payload body that is being replaced.
+     * @param payloadBody The new payload body.
+     * @return {@code this}
+     */
+    BlockingStreamingHttpResponse payloadBody(InputStream payloadBody);
+
     /**
      * Returns a {@link BlockingStreamingHttpResponse} with its underlying payload set to the result of serialization.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link Iterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(Function, HttpSerializer)} for more
      * fine grain control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link Iterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body, prior to serialization.
      * @param serializer Used to serialize the payload body.
      * @param <T> The type of objects to serialize.
@@ -101,12 +115,12 @@ default <T> BlockingIterable<T> payloadBody(HttpDeserializer<T> deserializer) {
     /**
      * Returns a {@link BlockingStreamingHttpResponse} with its underlying payload set to the result of serialization.
      * <p>
-     * A best effort will be made to apply back pressure to the existing {@link CloseableIterable} payload body. If this
+     * A best effort will be made to apply back pressure to the existing payload body which is being replaced. If this
      * default policy is not sufficient you can use {@link #transformPayloadBody(Function, HttpSerializer)} for more
      * fine grain control.
      * <p>
      * This method reserves the right to delay completion/consumption of {@code payloadBody}. This may occur due to the
-     * combination with the existing {@link CloseableIterable} payload body.
+     * combination with the existing payload body that is being replaced.
      * @param payloadBody The new payload body, prior to serialization.
      * @param serializer Used to serialize the payload body.
      * @param <T> The type of objects to serialize.

diff --git a/...k-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpRequest.java b/...k-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpRequest.java
@@ -20,9 +20,11 @@
 import io.servicetalk.concurrent.CloseableIterable;
 import io.servicetalk.concurrent.api.Single;
 
+import java.io.InputStream;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
+import static io.servicetalk.concurrent.api.Publisher.fromInputStream;
 import static io.servicetalk.concurrent.api.Publisher.fromIterable;
 
 final class DefaultBlockingStreamingHttpRequest extends AbstractDelegatingHttpRequest
@@ -127,6 +129,13 @@ public BlockingStreamingHttpRequest payloadBody(final CloseableIterable<Buffer>
         return this;
     }
 
+    @Override
+    public BlockingStreamingHttpRequest payloadBody(final InputStream payloadBody) {
+        original.payloadBody(fromInputStream(payloadBody)
+                .map(bytes -> original.payloadHolder().allocator().wrap(bytes)));
+        return this;
+    }
+
     @Override
     public <T> BlockingStreamingHttpRequest payloadBody(final Iterable<T> payloadBody,
                                                               final HttpSerializer<T> serializer) {

diff --git a/...-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpResponse.java b/...-http-api/src/main/java/io/servicetalk/http/api/DefaultBlockingStreamingHttpResponse.java
@@ -20,9 +20,11 @@
 import io.servicetalk.concurrent.CloseableIterable;
 import io.servicetalk.concurrent.api.Single;
 
+import java.io.InputStream;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
+import static io.servicetalk.concurrent.api.Publisher.fromInputStream;
 import static io.servicetalk.concurrent.api.Publisher.fromIterable;
 
 final class DefaultBlockingStreamingHttpResponse extends AbstractDelegatingHttpResponse
@@ -49,6 +51,13 @@ public BlockingStreamingHttpResponse payloadBody(final CloseableIterable<Buffer>
         return this;
     }
 
+    @Override
+    public BlockingStreamingHttpResponse payloadBody(final InputStream payloadBody) {
+        original.payloadBody(fromInputStream(payloadBody)
+                .map(bytes -> original.payloadHolder().allocator().wrap(bytes)));
+        return this;
+    }
+
     @Override
     public <T> BlockingStreamingHttpResponse payloadBody(final Iterable<T> payloadBody,
                                                          final HttpSerializer<T> serializer) {

diff --git a/...-http-netty/src/test/java/io/servicetalk/http/netty/BlockingStreamingHttpServiceTest.java b/...-http-netty/src/test/java/io/servicetalk/http/netty/BlockingStreamingHttpServiceTest.java
@@ -34,6 +34,7 @@
 import org.junit.rules.ExpectedException;
 import org.junit.rules.Timeout;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -142,6 +143,40 @@ public void receivePayloadBody() throws Exception {
         assertThat(receivedPayloadBody.toString(), is(HELLO_WORLD));
     }
 
+    @Test
+    public void clientRequestInputStreamPayloadBody() throws Exception {
+        StringBuilder receivedPayloadBody = new StringBuilder();
+
+        BlockingStreamingHttpClient client = context((ctx, request, response) -> {
+            request.payloadBody().forEach(chunk -> receivedPayloadBody.append(chunk.toString(US_ASCII)));
+            response.sendMetaData().close();
+        });
+
+        BlockingStreamingHttpResponse response = client.request(client.post("/")
+                .payloadBody(new ByteArrayInputStream(HELLO_WORLD.getBytes(US_ASCII))));
+        assertResponse(response);
+        assertThat(response.toResponse().toFuture().get().payloadBody(), is(EMPTY_BUFFER));
+        assertThat(receivedPayloadBody.toString(), is(HELLO_WORLD));
+    }
+
+    @Test
+    public void clientResponseInputStreamPayloadBody() throws Exception {
+        StringBuilder receivedPayloadBody = new StringBuilder();
+
+        BlockingStreamingHttpClient client = context((ctx, request, response) -> {
+            request.payloadBody().forEach(chunk -> receivedPayloadBody.append(chunk.toString(US_ASCII)));
+            response.sendMetaData().close();
+        });
+
+        String expectedBody = "overwritten";
+        BlockingStreamingHttpResponse response = client.request(client.post("/")
+                .payloadBody(new ByteArrayInputStream(HELLO_WORLD.getBytes(US_ASCII))));
+        assertResponse(response);
+        response.payloadBody(new ByteArrayInputStream(expectedBody.getBytes(US_ASCII)));
+        assertThat(response.toResponse().toFuture().get().payloadBody().toString(US_ASCII), is(expectedBody));
+        assertThat(receivedPayloadBody.toString(), is(HELLO_WORLD));
+    }
+
     @Test
     public void respondWithPayloadBodyAndTrailersUsingPayloadWriter() throws Exception {
         respondWithPayloadBodyAndTrailers((ctx, request, response) -> {