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

buongarzoni · buongarzoni · commit 1f5f214e0f51 · 2026-05-04T02:38:53.000-03:00
diff --git a/rollbar-reactive-streams-reactor/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/ReactorAsyncHttpClient.java b/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,8 +51,8 @@ public class ReactorAsyncHttpClient implements AsyncHttpClient {
   public Publisher<AsyncHttpResponse> send(AsyncHttpRequest httpRequest) {
 
     ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
-    if (httpRequest.getBodyBytes() != null) {
-      buffer.writeBytes(httpRequest.getBodyBytes());
+    if (httpRequest.isCompressionRequested()) {
+      buffer.writeBytes(compress(httpRequest.getBody()));
     } else {
       buffer.writeCharSequence(httpRequest.getBody(), StandardCharsets.UTF_8);
     }
@@ -59,6 +63,9 @@ public Publisher<AsyncHttpResponse> send(AsyncHttpRequest httpRequest) {
           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())
@@ -144,6 +151,18 @@ private static ProxyProvider.Proxy getProxyType(Proxy proxy) {
     }
   }
 
+  private static byte[] compress(String json) {
+    try {
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      GZIPOutputStream gzip = new GZIPOutputStream(baos);
+      gzip.write(json.getBytes(SyncSender.UTF_8));
+      gzip.close();
+      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;
diff --git a/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/AsyncSender.java b/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/AsyncSender.java
@@ -10,12 +10,9 @@
 import com.rollbar.reactivestreams.notifier.sender.http.AsyncHttpClient;
 import com.rollbar.reactivestreams.notifier.sender.http.AsyncHttpRequest;
 import com.rollbar.reactivestreams.notifier.sender.http.AsyncHttpResponse;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.LinkedHashMap;
-import java.util.zip.GZIPOutputStream;
 import org.reactivestreams.Publisher;
 
 /**
@@ -56,13 +53,8 @@ public Publisher<Response> send(Payload payload) {
 
     String json = jsonSerializer.toJson(payload);
 
-    AsyncHttpRequest request;
-    if (compressPayload) {
-      headers.put("Content-Encoding", "gzip");
-      request = AsyncHttpRequest.Builder.build(this.url, headers.entrySet(), compress(json));
-    } else {
-      request = AsyncHttpRequest.Builder.build(this.url, headers.entrySet(), json);
-    }
+    AsyncHttpRequest request =
+        AsyncHttpRequest.Builder.build(this.url, headers.entrySet(), json, compressPayload);
 
     return Utils.map(httpClient.send(request),
         new Utils.Converter<AsyncHttpResponse, Response>() {
@@ -74,18 +66,6 @@ public Response convert(AsyncHttpResponse from) {
         });
   }
 
-  private static byte[] compress(String json) {
-    try {
-      ByteArrayOutputStream baos = new ByteArrayOutputStream();
-      GZIPOutputStream gzip = new GZIPOutputStream(baos);
-      gzip.write(json.getBytes(SyncSender.UTF_8));
-      gzip.close();
-      return baos.toByteArray();
-    } catch (IOException e) {
-      throw new RuntimeException("Failed to gzip-compress payload", e);
-    }
-  }
-
   @Override
   public void close(boolean wait) {
     httpClient.close(wait);
diff --git a/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/ApacheRequestPublisher.java b/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/ApacheRequestPublisher.java
@@ -1,10 +1,14 @@
 package com.rollbar.reactivestreams.notifier.sender.http;
 
+import com.rollbar.notifier.sender.SyncSender;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.net.URI;
 import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPOutputStream;
 import org.apache.hc.client5.http.async.HttpAsyncClient;
 import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
 import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
@@ -102,8 +106,9 @@ private SimpleRequestProducer buildRequest() {
         req.setHeader(header.getKey(), header.getValue());
       }
 
-      if (request.getBodyBytes() != null) {
-        req.setBody(request.getBodyBytes(), ContentType.APPLICATION_JSON);
+      if (request.isCompressionRequested()) {
+        req.addHeader("Content-Encoding", "gzip");
+        req.setBody(compress(request.getBody()), ContentType.APPLICATION_JSON);
       } else {
         req.setBody(request.getBody(), ContentType.APPLICATION_JSON);
       }
@@ -120,4 +125,16 @@ public void cancel() {
       }
     }
   }
+
+  private static byte[] compress(String json) {
+    try {
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      GZIPOutputStream gzip = new GZIPOutputStream(baos);
+      gzip.write(json.getBytes(SyncSender.UTF_8));
+      gzip.close();
+      return baos.toByteArray();
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to gzip-compress payload", e);
+    }
+  }
 }
diff --git a/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/AsyncHttpRequest.java b/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/AsyncHttpRequest.java
@@ -13,19 +13,19 @@ public interface AsyncHttpRequest {
 
   String getBody();
 
-  default byte[] getBodyBytes() {
-    return null;
+  default boolean isCompressionRequested() {
+    return false;
   }
 
   class Builder {
     public static AsyncHttpRequest build(String url, Set<Map.Entry<String, String>> headers,
                                          String reqBody) {
-      return new AsyncHttpRequestImpl(url, headers, reqBody);
+      return new AsyncHttpRequestImpl(url, headers, reqBody, false);
     }
 
     public static AsyncHttpRequest build(String url, Set<Map.Entry<String, String>> headers,
-                                         byte[] body) {
-      return new AsyncHttpRequestImpl(url, headers, body);
+                                         String reqBody, boolean compressionRequested) {
+      return new AsyncHttpRequestImpl(url, headers, reqBody, compressionRequested);
     }
   }
 }
diff --git a/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/AsyncHttpRequestImpl.java b/rollbar-reactive-streams/src/main/java/com/rollbar/reactivestreams/notifier/sender/http/AsyncHttpRequestImpl.java
@@ -9,22 +9,14 @@ class AsyncHttpRequestImpl implements AsyncHttpRequest {
   private final String url;
   private final Iterable<Map.Entry<String, String>> headers;
   private final String body;
-  private final byte[] bodyBytes;
+  private final boolean compressionRequested;
 
   public AsyncHttpRequestImpl(String url, Iterable<Map.Entry<String, String>> headers,
-                              String body) {
+                              String body, boolean compressionRequested) {
     this.url = url;
     this.headers = headers;
     this.body = body;
-    this.bodyBytes = null;
-  }
-
-  public AsyncHttpRequestImpl(String url, Iterable<Map.Entry<String, String>> headers,
-                              byte[] bodyBytes) {
-    this.url = url;
-    this.headers = headers;
-    this.body = null;
-    this.bodyBytes = bodyBytes;
+    this.compressionRequested = compressionRequested;
   }
 
   @Override
@@ -43,7 +35,7 @@ public String getBody() {
   }
 
   @Override
-  public byte[] getBodyBytes() {
-    return bodyBytes;
+  public boolean isCompressionRequested() {
+    return compressionRequested;
   }
 }
diff --git a/rollbar-reactive-streams/src/test/java/com/rollbar/reactivestreams/notifier/sender/http/ApacheAsyncHttpTckTest.java b/rollbar-reactive-streams/src/test/java/com/rollbar/reactivestreams/notifier/sender/http/ApacheAsyncHttpTckTest.java
@@ -74,7 +74,7 @@ public Publisher<SimpleHttpResponse> createPublisher(long elements) {
     });
 
     return new ApacheRequestPublisher(client, new AsyncHttpRequestImpl(url,
-        new LinkedHashMap<String, String>().entrySet(), ""));
+        new LinkedHashMap<String, String>().entrySet(), "", false));
   }
 
   @Override
@@ -85,7 +85,7 @@ public Publisher<SimpleHttpResponse> createFailedPublisher() {
     });
 
     return new ApacheRequestPublisher(client, new AsyncHttpRequestImpl(url,
-        new LinkedHashMap<String, String>().entrySet(), ""));
+        new LinkedHashMap<String, String>().entrySet(), "", false));
   }
 
   @Override

Original file line number	Diff line number	Diff line change
`@@ -13,19 +13,19 @@ public interface AsyncHttpRequest {`
`13`	`13`
`14`	`14`	`String getBody();`
`15`	`15`
`16`		`- default byte[] getBodyBytes() {`
`17`		`- return null;`
	`16`	`+ default boolean isCompressionRequested() {`
	`17`	`+ return false;`
`18`	`18`	`}`
`19`	`19`
`20`	`20`	`class Builder {`
`21`	`21`	`public static AsyncHttpRequest build(String url, Set<Map.Entry<String, String>> headers,`
`22`	`22`	`String reqBody) {`
`23`		`- return new AsyncHttpRequestImpl(url, headers, reqBody);`
	`23`	`+ return new AsyncHttpRequestImpl(url, headers, reqBody, false);`
`24`	`24`	`}`
`25`	`25`
`26`	`26`	`public static AsyncHttpRequest build(String url, Set<Map.Entry<String, String>> headers,`
`27`		`- byte[] body) {`
`28`		`- return new AsyncHttpRequestImpl(url, headers, body);`
	`27`	`+ String reqBody, boolean compressionRequested) {`
	`28`	`+ return new AsyncHttpRequestImpl(url, headers, reqBody, compressionRequested);`
`29`	`29`	`}`
`30`	`30`	`}`
`31`	`31`	`}`
Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ public Publisher<SimpleHttpResponse> createPublisher(long elements) {`
`74`	`74`	`});`
`75`	`75`
`76`	`76`	`return new ApacheRequestPublisher(client, new AsyncHttpRequestImpl(url,`
`77`		`- new LinkedHashMap<String, String>().entrySet(), ""));`
	`77`	`+ new LinkedHashMap<String, String>().entrySet(), "", false));`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`@Override`
`@@ -85,7 +85,7 @@ public Publisher<SimpleHttpResponse> createFailedPublisher() {`
`85`	`85`	`});`
`86`	`86`
`87`	`87`	`return new ApacheRequestPublisher(client, new AsyncHttpRequestImpl(url,`
`88`		`- new LinkedHashMap<String, String>().entrySet(), ""));`
	`88`	`+ new LinkedHashMap<String, String>().entrySet(), "", false));`
`89`	`89`	`}`
`90`	`90`
`91`	`91`	`@Override`