[Feat][SDK-594] Compress payload and update payload max size to 1mb (#371)

* feat(java): raise max payload size to 1MB

* feat(java): compress outgoing payloads with gzip by default

* style(java): fix import order in SyncSender

* fix(java): add default implementation to compressPayload to avoid API break

* feat(reactive-streams): wire gzip compression into AsyncSender matching SyncSender behaviour

* fix(java): close raw stream if GZIPOutputStream constructor throws in sendJson

* fix(reactive-streams): disable compression in AsyncSenderTest to avoid null getBody()

* refactor(reactive-streams): use compression-intent pattern for safe third-party extensibility

* fix(reactive-streams): prevent GZIPOutputStream and ByteBuf resource leaks

Author: buongarzoni

rollbar-java/src/integTest/java/com/rollbar/notifier/RollbarITest.java

@@ -352,7 +352,7 @@ public void ifPayloadIsTooLargeItShouldBeTruncated() {
     assertThat(frames, hasSize(20));
 
     assertThat(PayloadTruncator.sizeInBytes(payloadJsonString),
-        lessThanOrEqualTo(512 * 1024));
+        lessThanOrEqualTo(1024 * 1024));
   }
 
   protected Sender buildSender(String url, String accessToken, Proxy proxy) {

rollbar-java/src/main/java/com/rollbar/notifier/RollbarBase.java

@@ -36,7 +36,7 @@
 public abstract class RollbarBase<RESULT, C extends CommonConfig> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RollbarBase.class);
-  private static final int MAX_PAYLOAD_SIZE_BYTES = 512 * 1024; // 512kb
+  private static final int MAX_PAYLOAD_SIZE_BYTES = 1024 * 1024; // 1mb
 
   protected BodyFactory bodyFactory;
   protected PayloadTruncator payloadTruncator;

rollbar-java/src/main/java/com/rollbar/notifier/config/CommonConfig.java

@@ -212,6 +212,17 @@ public interface CommonConfig {
    */
   boolean truncateLargePayloads();
 
+  /**
+   * <p>
+   * If set to true (the default), payloads are gzip-compressed before sending.
+   * Set to false to send uncompressed JSON.
+   * </p>
+   * @return true to compress payloads, false otherwise.
+   */
+  default boolean compressPayload() {
+    return true;
+  }
+
   int maximumTelemetryData();
 
   TelemetryEventTracker telemetryEventTracker();

rollbar-java/src/main/java/com/rollbar/notifier/config/ConfigBuilder.java

@@ -87,6 +87,8 @@ public class ConfigBuilder {
 
   protected boolean truncateLargePayloads;
 
+  protected boolean compressPayload;
+
   private int maximumTelemetryData =
       RollbarTelemetryEventTracker.MAXIMUM_CAPACITY_FOR_TELEMETRY_EVENTS;
 
@@ -101,6 +103,7 @@ protected ConfigBuilder(String accessToken) {
     this.accessToken = accessToken;
     this.handleUncaughtErrors = true;
     this.enabled = true;
+    this.compressPayload = true;
     this.defaultLevels = new DefaultLevels();
   }
 
@@ -136,6 +139,7 @@ private ConfigBuilder(Config config) {
     this.appPackages = config.appPackages();
     this.defaultLevels = new DefaultLevels(config);
     this.truncateLargePayloads = config.truncateLargePayloads();
+    this.compressPayload = config.compressPayload();
     this.maximumTelemetryData = config.maximumTelemetryData();
     this.telemetryEventTracker = config.telemetryEventTracker();
   }
@@ -480,6 +484,18 @@ public ConfigBuilder truncateLargePayloads(boolean truncate) {
     return this;
   }
 
+  /**
+   * <p>
+   * If set to false, payloads will not be gzip-compressed before sending. Default: true.
+   * </p>
+   * @param compress true to gzip-compress payloads.
+   * @return the builder instance.
+   */
+  public ConfigBuilder compressPayload(boolean compress) {
+    this.compressPayload = compress;
+    return this;
+  }
+
   /**
    * <p>
    * Maximum Telemetry events sent in a payload, only for the default TelemetryEventTracker, if
@@ -526,7 +542,8 @@ public Config build() {
       SyncSender.Builder innerSender =
           new SyncSender.Builder(this.endpoint)
           .accessToken(accessToken)
-          .proxy(proxy);
+          .proxy(proxy)
+          .compressPayload(this.compressPayload);
       if (this.jsonSerializer != null) {
         innerSender.jsonSerializer(this.jsonSerializer);
       }
@@ -601,6 +618,8 @@ private static class ConfigImpl implements Config {
 
     private final boolean truncateLargePayloads;
 
+    private final boolean compressPayload;
+
     private final int maximumTelemetryData;
 
     private final TelemetryEventTracker telemetryEventTracker;
@@ -637,6 +656,7 @@ private static class ConfigImpl implements Config {
       this.enabled = builder.enabled;
       this.defaultLevels = builder.defaultLevels;
       this.truncateLargePayloads = builder.truncateLargePayloads;
+      this.compressPayload = builder.compressPayload;
       this.maximumTelemetryData = builder.maximumTelemetryData;
       this.telemetryEventTracker = builder.telemetryEventTracker;
     }
@@ -786,6 +806,11 @@ public boolean truncateLargePayloads() {
       return this.truncateLargePayloads;
     }
 
+    @Override
+    public boolean compressPayload() {
+      return this.compressPayload;
+    }
+
     @Override
     public int maximumTelemetryData() {
       return this.maximumTelemetryData;

rollbar-java/src/main/java/com/rollbar/notifier/sender/SyncSender.java

@@ -15,6 +15,7 @@
 import java.net.MalformedURLException;
 import java.net.Proxy;
 import java.net.URL;
+import java.util.zip.GZIPOutputStream;
 
 /**
  * Synchronous implementation of the {@link Sender sender}.
@@ -33,11 +34,14 @@ public class SyncSender extends AbstractSender {
 
   private final Proxy proxy;
 
+  private final boolean compressPayload;
+
   SyncSender(Builder builder) {
     this.url = builder.url;
     this.jsonSerializer = builder.jsonSerializer;
     this.accessToken = builder.accessToken;
     this.proxy = builder.proxy != null ? builder.proxy : Proxy.NO_PROXY;
+    this.compressPayload = builder.compressPayload;
   }
 
   @Override
@@ -69,6 +73,9 @@ private HttpURLConnection getConnection() throws IOException {
     connection.setRequestProperty("Accept-Charset", UTF_8);
     connection.setRequestProperty("Content-Type", "application/json; charset=" + UTF_8);
     connection.setRequestProperty("Accept", "application/json");
+    if (compressPayload) {
+      connection.setRequestProperty("Content-Encoding", "gzip");
+    }
     connection.setDoOutput(true);
     connection.setRequestMethod("POST");
 
@@ -78,10 +85,14 @@ private HttpURLConnection getConnection() throws IOException {
   private void sendJson(HttpURLConnection connection, byte[] bytes) throws IOException {
     OutputStream out = null;
     try {
-      out = connection.getOutputStream();
+      OutputStream raw = connection.getOutputStream();
+      try {
+        out = compressPayload ? new GZIPOutputStream(raw) : raw;
+      } catch (IOException e) {
+        ObjectsUtils.close(raw);
+        throw e;
+      }
       out.write(bytes, 0, bytes.length);
-    } catch (IOException e) {
-      throw e;
     } finally {
       ObjectsUtils.close(out);
     }
@@ -131,6 +142,8 @@ public static final class Builder {
 
     private Proxy proxy;
 
+    private boolean compressPayload = true;
+
     public Builder() {
       this(DEFAULT_API_ENDPOINT);
     }
@@ -195,6 +208,16 @@ public Builder proxy(Proxy proxy) {
       return this;
     }
 
+    /**
+     * Whether to gzip-compress payloads before sending. Default: true.
+     * @param compress true to enable compression.
+     * @return the builder instance.
+     */
+    public Builder compressPayload(boolean compress) {
+      this.compressPayload = compress;
+      return this;
+    }
+
     /**
      * Builds the {@link SyncSender sync sender}.
      *

rollbar-java/src/test/java/com/rollbar/notifier/sender/SyncSenderTest.java

@@ -17,9 +17,11 @@
 import com.rollbar.notifier.sender.result.Response;
 import com.rollbar.notifier.sender.result.Result;
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.zip.GZIPInputStream;
 import java.net.HttpURLConnection;
 import java.net.Proxy;
 import java.net.URL;
@@ -76,6 +78,7 @@ public void setUp()throws Exception {
     sut = new SyncSender.Builder()
         .url(url)
         .jsonSerializer(serializer)
+        .compressPayload(false)
         .build();
     sut.addListener(listener);
   }
@@ -198,6 +201,57 @@ public void shouldSendThePayloadUsingAProxyIfProvided() throws Exception {
     verify(listener).onResponse(payload, expectedResponse);
   }
 
+  @Test
+  public void shouldSendGzipEncodedPayloadWhenCompressionEnabled() throws Exception {
+    ByteArrayOutputStream capturedBytes = new ByteArrayOutputStream();
+    when(url.openConnection(eq(Proxy.NO_PROXY))).thenReturn(connection);
+    when(connection.getOutputStream()).thenReturn(capturedBytes);
+
+    int responseCode = 200;
+    String responseJson = "simulated_response_json";
+    when(connection.getResponseCode()).thenReturn(responseCode);
+    when(connection.getInputStream())
+        .thenReturn(new ByteArrayInputStream(responseJson.getBytes(UTF_8)));
+    when(serializer.resultFrom(responseJson)).thenReturn(result);
+
+    SyncSender compressingSut = new SyncSender.Builder()
+        .url(url)
+        .jsonSerializer(serializer)
+        .compressPayload(true)
+        .build();
+
+    compressingSut.send(payload);
+
+    verify(connection).setRequestProperty("Content-Encoding", "gzip");
+
+    GZIPInputStream gzipIn = new GZIPInputStream(
+        new ByteArrayInputStream(capturedBytes.toByteArray()));
+    ByteArrayOutputStream decompressedBytes = new ByteArrayOutputStream();
+    byte[] buf = new byte[1024];
+    int n;
+    while ((n = gzipIn.read(buf)) != -1) {
+      decompressedBytes.write(buf, 0, n);
+    }
+    String decompressed = decompressedBytes.toString(UTF_8);
+    assertThat(decompressed, is(PAYLOAD_JSON));
+  }
+
+  @Test
+  public void shouldNotSetContentEncodingWhenCompressionDisabled() throws Exception {
+    int responseCode = 200;
+    String responseJson = "simulated_response_json";
+    when(connection.getResponseCode()).thenReturn(responseCode);
+    when(connection.getInputStream())
+        .thenReturn(new ByteArrayInputStream(responseJson.getBytes(UTF_8)));
+    when(serializer.resultFrom(responseJson)).thenReturn(result);
+
+    sut.send(payload);
+
+    verify(connection, org.mockito.Mockito.never())
+        .setRequestProperty(org.mockito.ArgumentMatchers.eq("Content-Encoding"),
+            org.mockito.ArgumentMatchers.anyString());
+  }
+
   private void verifyHttp() throws Exception {
     verify(connection).setRequestProperty("Accept-Charset", UTF_8);
     verify(connection).setRequestProperty("Content-Type", "application/json; charset=" + UTF_8);

rollbar-reactive-streams-reactor/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/ReactorAsyncHttpClient.java

@@ -1,12 +1,16 @@
 package com.rollbar.reactivestreams.notifier.sender.http;
 
+import com.rollbar.notifier.sender.SyncSender;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.Proxy;
 import java.nio.charset.StandardCharsets;
 import java.util.AbstractMap;
 import java.util.Map;
+import java.util.zip.GZIPOutputStream;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
@@ -47,14 +51,26 @@ public class ReactorAsyncHttpClient implements AsyncHttpClient {
   public Publisher<AsyncHttpResponse> send(AsyncHttpRequest httpRequest) {
 
     ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
-    buffer.writeCharSequence(httpRequest.getBody(), StandardCharsets.UTF_8);
+    try {
+      if (httpRequest.isCompressionRequested()) {
+        buffer.writeBytes(compress(httpRequest.getBody()));
+      } else {
+        buffer.writeCharSequence(httpRequest.getBody(), StandardCharsets.UTF_8);
+      }
+    } catch (Throwable t) {
+      buffer.release();
+      throw t;
+    }
     Mono<ByteBuf> buf = Mono.just(buffer);
 
     return httpClient
         .headers(entries -> {
           for (Map.Entry<String, String> header : httpRequest.getHeaders()) {
             entries.add(header.getKey(), header.getValue());
           }
+          if (httpRequest.isCompressionRequested()) {
+            entries.add("Content-Encoding", "gzip");
+          }
         })
         .post()
         .uri(httpRequest.getUrl())
@@ -140,6 +156,18 @@ private static ProxyProvider.Proxy getProxyType(Proxy proxy) {
     }
   }
 
+  private static byte[] compress(String json) {
+    try {
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
+        gzip.write(json.getBytes(SyncSender.UTF_8));
+      }
+      return baos.toByteArray();
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to gzip-compress payload", e);
+    }
+  }
+
   public static final class Builder {
     private Proxy proxy;
     private ConnectionProvider connectionProvider;