Merge pull request #252 from IABTechLab/ian-UID2-6337-asynchronous-full-queue-process

process async with polling, no message limit

Author: Ian-Nara

src/main/java/com/uid2/optout/Const.java

@@ -22,6 +22,16 @@ public static class Config extends com.uid2.shared.Const.Config {
         public static final String OptOutSqsS3FolderProp = "optout_sqs_s3_folder"; // Default: "sqs-delta" - folder within same S3 bucket as regular optout
         public static final String OptOutSqsMaxMessagesPerPollProp = "optout_sqs_max_messages_per_poll";
         public static final String OptOutSqsVisibilityTimeoutProp = "optout_sqs_visibility_timeout";
+        public static final String OptOutDeltaJobTimeoutSecondsProp = "optout_delta_job_timeout_seconds";
+        public static final String OptOutS3BucketDroppedRequestsProp = "optout_s3_bucket_dropped_requests";
+        public static final String OptOutMaxMessagesPerFileProp = "optout_max_messages_per_file";
+        public static final String TrafficFilterConfigPathProp = "traffic_filter_config_path";
+        public static final String TrafficCalcConfigPathProp = "traffic_calc_config_path";
+        public static final String ManualOverrideS3PathProp = "manual_override_s3_path";
+        public static final String OptOutTrafficCalcBaselineTrafficProp = "traffic_calc_baseline_traffic";
+        public static final String OptOutTrafficCalcThresholdMultiplierProp = "traffic_calc_threshold_multiplier";
+        public static final String OptOutTrafficCalcEvaluationWindowSecondsProp = "traffic_calc_evaluation_window_seconds";
+        public static final String OptOutTrafficCalcAllowlistRangesProp = "traffic_calc_allowlist_ranges";
     }
 
     public static class Event {

src/main/java/com/uid2/optout/vertx/DeltaProduceJobStatus.java

@@ -0,0 +1,85 @@
+package com.uid2.optout.vertx;
+
+import io.vertx.core.json.JsonObject;
+import java.time.Instant;
+
+/**
+ * Represents the status and result of an async delta production job on a pod.
+ * 
+ * This class tracks the lifecycle of a delta production job including its state
+ * (running, completed, failed), timing information, and result or error details.
+ * 
+ */
+public class DeltaProduceJobStatus {
+    private final Instant startTime;
+    private volatile JobState state;
+    private volatile JsonObject result;
+    private volatile String errorMessage;
+    private volatile Instant endTime;
+
+    public enum JobState {
+        RUNNING,
+        COMPLETED,
+        FAILED
+    }
+
+    public DeltaProduceJobStatus() {
+        this.startTime = Instant.now();
+        this.state = JobState.RUNNING;
+    }
+
+    /**
+     * Mark the job as completed with the given result.
+     * @param result The result details as a JsonObject
+     */
+    public void complete(JsonObject result) {
+        this.result = result;
+        this.state = JobState.COMPLETED;
+        this.endTime = Instant.now();
+    }
+
+    /**
+     * Mark the job as failed with the given error message.
+     * @param errorMessage Description of the failure
+     */
+    public void fail(String errorMessage) {
+        this.errorMessage = errorMessage;
+        this.state = JobState.FAILED;
+        this.endTime = Instant.now();
+    }
+
+    /**
+     * Get the current state of the job.
+     * @return The job state
+     */
+    public JobState getState() {
+        return state;
+    }
+
+    /**
+     * Convert the job status to a JSON representation for API responses.
+     * @return JsonObject with state, timing, and result/error information
+     */
+    public JsonObject toJson() {
+        JsonObject json = new JsonObject()
+                .put("state", state.name().toLowerCase())
+                .put("start_time", startTime.toString());
+
+        if (endTime != null) {
+            json.put("end_time", endTime.toString());
+            long durationSeconds = endTime.getEpochSecond() - startTime.getEpochSecond();
+            json.put("duration_seconds", durationSeconds);
+        }
+
+        if (state == JobState.COMPLETED && result != null) {
+            json.put("result", result);
+        }
+
+        if (state == JobState.FAILED && errorMessage != null) {
+            json.put("error", errorMessage);
+        }
+
+        return json;
+    }
+}
+

src/main/java/com/uid2/optout/vertx/DeltaProductionResult.java

@@ -7,9 +7,17 @@ public class DeltaProductionResult {
     private final int deltasProduced;
     private final int entriesProcessed;
 
-    public DeltaProductionResult(int deltasProduced, int entriesProcessed) {
+    /* 
+     * indicates that there are still messages in the queue, however,
+     * not enough time has elapsed to produce a delta file.
+     * We produce in batches of (5 minutes) 
+     */
+    private final boolean stoppedDueToMessagesTooRecent;
+
+    public DeltaProductionResult(int deltasProduced, int entriesProcessed, boolean stoppedDueToMessagesTooRecent) {
         this.deltasProduced = deltasProduced;
         this.entriesProcessed = entriesProcessed;
+        this.stoppedDueToMessagesTooRecent = stoppedDueToMessagesTooRecent;
     }
 
     public int getDeltasProduced() {
@@ -19,5 +27,9 @@ public int getDeltasProduced() {
     public int getEntriesProcessed() {
         return entriesProcessed;
     }
+
+    public boolean stoppedDueToMessagesTooRecent() {
+        return stoppedDueToMessagesTooRecent;
+    }
 }