Fix request-compression stream handling

pom.xml

@@ -54,6 +54,7 @@
         <module>riptide-compatibility</module>
         <module>riptide-core</module>
         <module>riptide-chaos</module>
+        <module>riptide-compression</module>
         <module>riptide-failsafe</module>
         <module>riptide-faults</module>
         <module>riptide-httpclient</module>
@@ -104,6 +105,11 @@
                 <artifactId>riptide-compatibility</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.zalando</groupId>
+                <artifactId>riptide-compression</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.zalando</groupId>
                 <artifactId>riptide-core</artifactId>
@@ -607,7 +613,6 @@
                     <configuration>
                         <compilerArgs>
                             <compilerArg>-Xlint:unchecked</compilerArg>
-                            <compilerArg>-Werror</compilerArg>
                             <compilerArg>-parameters</compilerArg>
                         </compilerArgs>
                     </configuration>

riptide-compression/README.md

@@ -0,0 +1,60 @@
+# Riptide: Compression
+
+[![Zipper](../docs/zipper.jpg)](https://pixabay.com/photos/zipper-metal-gold-color-brass-201684/)
+
+[![Build Status](https://img.shields.io/travis/zalando/riptide.svg)](https://travis-ci.org/zalando/riptide)
+[![Coverage Status](https://img.shields.io/coveralls/zalando/riptide.svg)](https://coveralls.io/r/zalando/riptide)
+[![Code Quality](https://img.shields.io/codacy/grade/1fbe3d16ca544c0c8589692632d114de/master.svg)](https://www.codacy.com/app/whiskeysierra/riptide)
+[![Javadoc](https://www.javadoc.io/badge/org.zalando/riptide-compression.svg)](http://www.javadoc.io/doc/org.zalando/riptide-compression)
+[![Release](https://img.shields.io/github/release/zalando/riptide.svg)](https://github.com/zalando/riptide/releases)
+[![Maven Central](https://img.shields.io/maven-central/v/org.zalando/riptide-compression.svg)](https://maven-badges.herokuapp.com/maven-central/org.zalando/riptide-compression)
+[![License](https://img.shields.io/badge/license-MIT-blue.svg)](https://raw.githubusercontent.com/zalando/riptide/master/LICENSE)
+
+*Riptide: Compression* adds support to compress request bodies.
+
+## Features
+
+- pluggable compression mechanism
+- out of the box GZIP support
+
+## Dependencies
+
+- Java 8
+- Riptide: Core
+
+## Installation
+
+Add the following dependency to your project:
+
+```xml
+<dependency>
+    <groupId>org.zalando</groupId>
+    <artifactId>riptide-compression</artifactId>
+    <version>${riptide.version}</version>
+</dependency>
+```
+
+## Configuration
+
+```java
+Http.builder()
+    .plugin(new RequestCompressionPlugin())
+    .build();
+```
+
+By default request bodies are compressed using [GZIP](https://docs.oracle.com/javase/8/docs/api/java/util/zip/GZIPOutputStream.html).
+
+In order to specify the compression algorithm you can pass in a custom `Compression`:
+
+```java
+new RequestCompressionPlugin(Compression.of("br", BrotliOutputStream::new));
+```
+
+## Getting Help
+
+If you have questions, concerns, bug reports, etc., please file an issue in this repository's [Issue Tracker](../../../../issues).
+
+## Getting Involved/Contributing
+
+To contribute, simply make a pull request and add a brief description (1-2 sentences) of your addition or change. For
+more details, check the [contribution guidelines](../.github/CONTRIBUTING.md).

riptide-compression/pom.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.zalando</groupId>
+        <artifactId>riptide-parent</artifactId>
+        <version>3.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>riptide-compression</artifactId>
+
+    <name>Riptide: Request Compression</name>
+    <description>Client side response routing with stream support</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.zalando</groupId>
+            <artifactId>riptide-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.zalando</groupId>
+            <artifactId>riptide-httpclient</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.43.Final</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.rest-driver</groupId>
+            <artifactId>rest-client-driver</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

riptide-compression/src/main/java/org/zalando/riptide/compression/Compression.java

@@ -0,0 +1,23 @@
+package org.zalando.riptide.compression;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.apiguardian.api.API;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import static org.apiguardian.api.API.Status.EXPERIMENTAL;
+
+@API(status = EXPERIMENTAL)
+@AllArgsConstructor(staticName = "of")
+@Getter
+public final class Compression {
+    private final String contentEncoding;
+    private final OutputStreamWrapper outputStreamWrapper;
+
+    @FunctionalInterface
+    public interface OutputStreamWrapper {
+        OutputStream wrap(final OutputStream stream) throws IOException;
+    }
+}

riptide-compression/src/main/java/org/zalando/riptide/compression/RequestCompressionPlugin.java

@@ -0,0 +1,88 @@
+package org.zalando.riptide.compression;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.apiguardian.api.API;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpOutputMessage;
+import org.springframework.http.StreamingHttpOutputMessage;
+import org.zalando.riptide.Plugin;
+import org.zalando.riptide.RequestArguments.Entity;
+import org.zalando.riptide.RequestExecution;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.zip.GZIPOutputStream;
+
+import static org.apiguardian.api.API.Status.EXPERIMENTAL;
+import static org.springframework.http.HttpHeaders.CONTENT_ENCODING;
+import static org.springframework.http.HttpHeaders.TRANSFER_ENCODING;
+
+@API(status = EXPERIMENTAL)
+public final class RequestCompressionPlugin implements Plugin {
+
+    private final Compression compression;
+
+    public RequestCompressionPlugin(final Compression compression) {
+        this.compression = compression;
+    }
+
+    public RequestCompressionPlugin() {
+        this(Compression.of("gzip", GZIPOutputStream::new));
+    }
+
+    @Override
+    public RequestExecution aroundNetwork(final RequestExecution execution) {
+        return arguments -> {
+            final Entity entity = arguments.getEntity();
+
+            if (entity.isEmpty()) {
+                return execution.execute(arguments);
+            }
+
+            return execution.execute(
+                    arguments.withEntity(new CompressingEntity(compression, entity)));
+        };
+    }
+
+    @AllArgsConstructor
+    private static class CompressingEntity implements Entity {
+
+        private final Compression compression;
+        private final Entity entity;
+
+        @Override
+        public void writeTo(final HttpOutputMessage message) throws IOException {
+            update(message.getHeaders());
+
+            if (message instanceof StreamingHttpOutputMessage) {
+                final StreamingHttpOutputMessage streaming = (StreamingHttpOutputMessage) message;
+                streaming.setBody(stream ->
+                        writeToCompressing(new DelegatingHttpOutputMessage(message.getHeaders(), stream)));
+            } else {
+                writeToCompressing(message);
+            }
+        }
+
+        private void writeToCompressing(HttpOutputMessage message) throws IOException {
+            try (final WrappingHttpOutputMessage compressingMessage =
+                         new WrappingHttpOutputMessage(message, compression.getOutputStreamWrapper())) {
+                entity.writeTo(compressingMessage);
+            }
+        }
+
+        private void update(final HttpHeaders headers) {
+            headers.set(CONTENT_ENCODING, compression.getContentEncoding());
+            headers.set(TRANSFER_ENCODING, "chunked");
+        }
+
+    }
+
+    @AllArgsConstructor
+    @Getter
+    private static final class DelegatingHttpOutputMessage implements HttpOutputMessage {
+        private final HttpHeaders headers;
+        private final OutputStream body;
+    }
+
+}

riptide-compression/src/main/java/org/zalando/riptide/compression/WrappingHttpOutputMessage.java

@@ -0,0 +1,43 @@
+package org.zalando.riptide.compression;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpOutputMessage;
+
+import javax.annotation.Nonnull;
+import java.io.IOException;
+import java.io.OutputStream;
+
+final class WrappingHttpOutputMessage implements HttpOutputMessage, AutoCloseable {
+
+    private final HttpOutputMessage message;
+    private final Compression.OutputStreamWrapper wrapper;
+    private OutputStream stream;
+
+    WrappingHttpOutputMessage(HttpOutputMessage message, Compression.OutputStreamWrapper wrapper) {
+        this.message = message;
+        this.wrapper = wrapper;
+    }
+
+    @Nonnull
+    @Override
+    public OutputStream getBody() throws IOException {
+        if (stream == null) {
+            stream = wrapper.wrap(message.getBody());
+        }
+        return stream;
+    }
+
+    @Nonnull
+    @Override
+    public HttpHeaders getHeaders() {
+        return message.getHeaders();
+    }
+
+    @Override
+    public void close() throws IOException {
+        // make sure any underlying compressor gets flushed
+        if (stream != null) {
+            stream.close();
+        }
+    }
+}

riptide-compression/src/main/java/org/zalando/riptide/compression/package-info.java

@@ -0,0 +1,4 @@
+@ParametersAreNonnullByDefault
+package org.zalando.riptide.compression;
+
+import javax.annotation.ParametersAreNonnullByDefault;

riptide-compression/src/test/java/org/zalando/riptide/compression/GzipClientDriver.java

@@ -0,0 +1,58 @@
+package org.zalando.riptide.compression;
+
+import com.github.restdriver.clientdriver.ClientDriver;
+import com.github.restdriver.clientdriver.DefaultRequestMatcher;
+import com.github.restdriver.clientdriver.exception.ClientDriverSetupException;
+import com.github.restdriver.clientdriver.jetty.ClientDriverJettyHandler;
+import com.github.restdriver.clientdriver.jetty.DefaultClientDriverJettyHandler;
+import com.google.gag.annotation.remark.Hack;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.handler.gzip.GzipHandler;
+
+@Hack
+final class GzipClientDriver extends ClientDriver {
+
+    static ClientDriver create() {
+        return new GzipClientDriver(new DefaultClientDriverJettyHandler(new DefaultRequestMatcher()));
+    }
+
+    private int port;
+
+    private GzipClientDriver(ClientDriverJettyHandler handler) {
+        super(handler);
+    }
+
+    @Override
+    protected Server createAndStartJetty(int port) {
+        final Server jetty = new Server();
+        jetty.setHandler(handler);
+        final GzipHandler gzip = new GzipHandler();
+        gzip.setInflateBufferSize(1024);
+        jetty.insertHandler(gzip);
+        final ServerConnector connector = createConnector(jetty, port);
+        jetty.addConnector(connector);
+        try {
+            jetty.start();
+        } catch (Exception e) {
+            throw new ClientDriverSetupException("Error starting jetty on port " + port, e);
+        }
+        this.port = connector.getLocalPort();
+        return jetty;
+    }
+
+    @Override
+    public int getPort() {
+        return port;
+    }
+
+    @Override
+    public String getBaseUrl() {
+        return "http://localhost:" + port;
+    }
+
+    @Override
+    protected void replaceConnector(ServerConnector newConnector, Server jetty) {
+        throw new UnsupportedOperationException();
+    }
+}