Package ai.attackframework.tools.burp.sinks

Class TrafficExportQueue

java.lang.Object
ai.attackframework.tools.burp.sinks.TrafficExportQueue
public final class TrafficExportQueue extends Object
Bounded queue for traffic documents so the HTTP thread can enqueue and return immediately.

A dedicated worker thread drains the queue and pushes via the OpenSearch Bulk API using chunked request body (NDJSON written incrementally; no full batch list in memory). When full, the oldest document is dropped (backpressure). Batches are limited by count (BatchSizeController), payload size (BulkPayloadEstimator, 5 MB cap), and time (flush after 100 ms). Used only by the traffic export path.

  • Method Details

    • getCurrentSize

      public static int getCurrentSize()
      Returns the current number of documents in the traffic export queue.
      Returns:
      number of documents currently queued (for stats and Exporter-index observability)

    • getCurrentBytesEstimate

      public static long getCurrentBytesEstimate()
      Returns the approximate total bytes of documents currently held in the in-memory queue.

      Computed on demand by iterating a snapshot of the queue and summing BulkPayloadEstimator.estimateBytes(Map). Intended for low-frequency callers (StatsPanel refresh). Does not include spilled documents; those are already tracked as bytes by getCurrentSpillBytes().

    • getCurrentSpillSize

      public static int getCurrentSpillSize()
      Returns current spill queue depth for StatsPanel observability.

    • getCurrentSpillBytes

      public static long getCurrentSpillBytes()
      Returns current spill queue bytes for StatsPanel observability.

    • getCurrentSpillOldestAgeMs

      public static long getCurrentSpillOldestAgeMs()
      Returns oldest spill age in milliseconds (0 when no spilled docs).

    • getRecoveredSpillCount

      public static long getRecoveredSpillCount()
      Returns startup-recovered spill document count.

    • getSpillDirectoryPath

      public static String getSpillDirectoryPath()
      Returns spill directory path for diagnostics.

    • offer

      public static void offer(Map<String,Object> document)
      Offers a traffic document to the queue. Non-blocking.

      If the queue is full, the document is first spilled to disk. Only when spill rejects it does this path fall back to drop-oldest behavior and record a drop in ExportStats. Starts the drain worker on first use. Thread-safe.

      Parameters:
      document - the document to enqueue; null is ignored

    • clearPendingWork

      public static void clearPendingWork()
      Clears in-memory and spilled traffic backlog.

      Used when export is intentionally stopped or a Start attempt fails, so queued traffic does not resume behind a stopped UI.

    • stopWorker

      public static void stopWorker()
      Interrupts and joins the drain worker so the extension unloads cleanly.

      Safe to call from any thread and safe to call more than once. Resets the start flag so the next offer(Map) lazily starts a fresh worker. Delegates termination to Workers so shutdown semantics match every other extension-owned worker; if the worker does not exit within WORKER_SHUTDOWN_TIMEOUT_MS milliseconds, the current thread's interrupt flag is restored and the method returns.