fix: align dedup example with upstream end-of-stream semantics

src/main/java/com/github/wrossmorrow/extproc/processors/DedupRequestProcessor.java

@@ -4,6 +4,10 @@
 import com.github.wrossmorrow.extproc.RequestContext;
 import com.github.wrossmorrow.extproc.RequestProcessor;
 import com.github.wrossmorrow.extproc.RequestProcessorHealthManager;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -30,38 +34,88 @@ public void shutdown() {
   }
 
   public void processRequestHeaders(RequestContext ctx, Map<String, String> headers) {
-    if (ctx.getRequestHeaders().containsKey("x-extproc-request-digest")) {
-      String digest = ctx.getRequestHeaders().get("x-extproc-request-digest");
+    if (!isDedupable(ctx)) {
+      return;
+    }
+    if (ctx.streamComplete()) {
+      String digest = sha256Hex(ctx.getMethod() + ":" + ctx.getPath());
+      ctx.setValue("digest", digest);
+      ctx.addHeader("x-extproc-request-digest", digest);
       if (inflight.containsKey(digest)) {
-        String requestId = inflight.get(digest);
         ctx.cancelRequest(
             409,
             "{\"message\":\"Duplicate request already in flight\",\"requestId\":\""
-                + requestId
+                + inflight.get(digest)
                 + "\"}");
         return;
       }
       inflight.put(digest, ctx.getRequestId());
     }
   }
 
-  public void processRequestBody(RequestContext ctx, String body) {}
+  public void processRequestBody(RequestContext ctx, String body) {
+    if (!isDedupable(ctx)) {
+      return;
+    }
+    if (ctx.streamComplete()) {
+      String digest = sha256Hex(ctx.getMethod() + ":" + ctx.getPath() + ":" + body);
+      ctx.setValue("digest", digest);
+      ctx.addHeader("x-extproc-request-digest", digest);
+      if (inflight.containsKey(digest)) {
+        ctx.cancelRequest(
+            409,
+            "{\"message\":\"Duplicate request already in flight\",\"requestId\":\""
+                + inflight.get(digest)
+                + "\"}");
+        return;
+      }
+      inflight.put(digest, ctx.getRequestId());
+    }
+  }
 
   public void processRequestTrailers(RequestContext ctx, Map<String, String> trailers) {}
 
   public void processResponseHeaders(RequestContext ctx, Map<String, String> headers) {
     if (ctx.streamComplete()) {
-      String digest = ctx.getRequestHeaders().get("x-extproc-request-digest");
-      inflight.remove(digest);
+      String digest = (String) ctx.getValue("digest");
+      if (digest != null) {
+        inflight.remove(digest);
+      }
     }
   }
 
   public void processResponseBody(RequestContext ctx, String body) {
     if (ctx.streamComplete()) {
-      String digest = ctx.getRequestHeaders().get("x-extproc-request-digest");
-      inflight.remove(digest);
+      String digest = (String) ctx.getValue("digest");
+      if (digest != null) {
+        inflight.remove(digest);
+      }
     }
   }
 
   public void processResponseTrailers(RequestContext ctx, Map<String, String> trailers) {}
+
+  private static boolean isDedupable(RequestContext ctx) {
+    String method = ctx.getMethod();
+    return "PUT".equals(method) || "POST".equals(method) || "PATCH".equals(method);
+  }
+
+  private static String sha256Hex(String input) {
+    try {
+      MessageDigest md = MessageDigest.getInstance("SHA-256");
+      byte[] d = md.digest(input.getBytes(StandardCharsets.UTF_8));
+      return hexEncode(d);
+    } catch (NoSuchAlgorithmException e) {
+      return "";
+    }
+  }
+
+  private static String hexEncode(final byte[] messageDigest) {
+    BigInteger bigint = new BigInteger(1, messageDigest);
+    String hexText = bigint.toString(16);
+    while (hexText.length() < 32) {
+      hexText = "0".concat(hexText);
+    }
+    return hexText;
+  }
 }