From b676343a1e9ddfb9ec2f39b2dc95bcb9eef015d9 Mon Sep 17 00:00:00 2001
From: Lionell Pack <lionell.pack@thetradedesk.com>
Date: Fri, 15 Mar 2024 17:47:11 +1100
Subject: [PATCH] Switch webhook logic to wait for each optout to succeed
 before sending the next one.

---
 .../com/uid2/optout/vertx/OptOutSender.java   | 32 +++++++++----------
 1 file changed, 16 insertions(+), 16 deletions(-)

diff --git a/src/main/java/com/uid2/optout/vertx/OptOutSender.java b/src/main/java/com/uid2/optout/vertx/OptOutSender.java
index c82c000..81e08c8 100644
--- a/src/main/java/com/uid2/optout/vertx/OptOutSender.java
+++ b/src/main/java/com/uid2/optout/vertx/OptOutSender.java
@@ -407,25 +407,25 @@ private void deltaReplay(Promise<Void> promise, OptOutCollection store, List<Str
             Future<Void> lastOp = Future.succeededFuture();
             for (int i = 0; i < store.size(); ++i) {
                 final OptOutEntry entry = store.get(i);
-                Future<Void> sendOp = this.remotePartner.send(entry);
-                sendOp.onComplete(v -> {
-                    if (v.succeeded()) {
-                        recordEntryReplayStatus("success");
-                        this.lastEntrySent.set(entry.timestamp);
-                    } else {
-                        if (v.cause() instanceof TooManyRetriesException) {
-                            recordEntryReplayStatus("too_many_retries");
-                        } else if (v.cause() instanceof UnexpectedStatusCodeException) {
-                            recordEntryReplayStatus("unexpected_status_code_" + ((UnexpectedStatusCodeException) v.cause()).getStatusCode());
+                lastOp = lastOp.compose(ar -> {
+                    Future<Void> sendOp = this.remotePartner.send(entry);
+                    return sendOp.onComplete(v -> {
+                        if (v.succeeded()) {
+                            recordEntryReplayStatus("success");
+                            this.lastEntrySent.set(entry.timestamp);
                         } else {
-                            recordEntryReplayStatus("unknown_error");
+                            if (v.cause() instanceof TooManyRetriesException) {
+                                recordEntryReplayStatus("too_many_retries");
+                            } else if (v.cause() instanceof UnexpectedStatusCodeException) {
+                                recordEntryReplayStatus("unexpected_status_code_" + ((UnexpectedStatusCodeException) v.cause()).getStatusCode());
+                            } else {
+                                recordEntryReplayStatus("unknown_error");
+                            }
+
+                            this.logger.error("deltaReplay failed sending entry: " + entry.timestamp, v.cause());
                         }
-
-                        this.logger.error("deltaReplay failed sending entry: " + entry.timestamp, v.cause());
-                    }
+                    });
                 });
-
-                lastOp = lastOp.compose(v -> sendOp);
             }
 
             lastOp.onComplete(ar -> {