3939/**
4040 * Buffers experimentation events and ships them to the Flagsmith events API.
4141 *
42- * <p>Events are flushed on a fixed interval, when the buffer fills up, and on {@link #close()}.
43- * Exposure events are deduplicated within a flush window. No exception thrown here reaches caller
44- * code: every failure is logged instead, including traits or metadata that cannot be serialised,
45- * which drop only their own event. {@link Error}s such as {@code OutOfMemoryError} are not caught.
42+ * <p>Events are flushed on a timer, when the buffer fills, and on {@link #close()}. Exposures are
43+ * deduplicated within a flush window. Exceptions are logged, never thrown to the caller;
44+ * {@link Error}s are not caught.
4645 */
4746public class EventProcessor {
4847
@@ -58,15 +57,9 @@ public class EventProcessor {
5857 private static final MediaType JSON_MEDIA_TYPE =
5958 MediaType .get ("application/json; charset=utf-8" );
6059 /**
61- * The most events that may be waiting on the events API at once. Once reached, a flush drops
62- * its batch instead of queueing it: while the API is slow or down, batches are produced faster
63- * than they are retried and given up on, and an unbounded queue of them would grow with the
64- * host application's traffic until it ran out of memory.
65- *
66- * <p>It counts events, not batches, so that it bounds memory without capping throughput on a
67- * healthy API: a batch count would throttle harder the smaller the configured buffer. At the
68- * default buffer size it is ten batches. A batch is admitted while the count is below the
69- * limit, so the bound is this plus one buffer's worth.
60+ * Cap on events awaiting the events API. Past it a flush drops its batch rather than queue it,
61+ * so an outage cannot grow memory with the host's traffic. It counts events, not batches, so a
62+ * small buffer is not throttled on a healthy API. The true bound is this plus one buffer.
7063 */
7164 static final int MAX_IN_FLIGHT_EVENTS = 10_000 ;
7265 /** The close timeout when the HTTP client's timeouts do not bound a request. */
@@ -83,8 +76,7 @@ public class EventProcessor {
8376 /** The interval between timed flushes; 0 means there is no timer. */
8477 @ Getter
8578 private final int flushIntervalMillis ;
86- // Everything below is internal state, deliberately not exposed: once published, a getter
87- // would be public API for as long as the SDK is supported.
79+ // No public getters below: each one would become supported API.
8880 private final List <Map <String , Object >> buffer = new ArrayList <>();
8981 private final Set <String > dedupeKeys = new HashSet <>();
9082 private final Object lock = new Object ();
@@ -110,9 +102,9 @@ public class EventProcessor {
110102 private ScheduledFuture <?> scheduledFlush ;
111103
112104 /**
113- * Instantiate with an HTTP client.
105+ * Create a processor that sends batches through {@code client} .
114106 *
115- * @param client client instance
107+ * @param client HTTP client; its timeouts also bound {@link #close()}
116108 * @param eventsUri base URI of the events API, e.g. https://events.api.flagsmith.com/
117109 * @param maxBufferItems number of buffered events that triggers an immediate flush; at
118110 * least 1
@@ -126,14 +118,10 @@ public EventProcessor(OkHttpClient client, HttpUrl eventsUri, int maxBufferItems
126118 new RequestProcessor (client , new FlagsmithLogger (), buildRetry ()));
127119 }
128120
129- /**
130- * Instantiate with a request processor. Package-private: it exists for tests, and is not
131- * something callers should come to depend on.
132- */
121+ /** For tests: sends through the given request processor. */
133122 EventProcessor (HttpUrl eventsUri , int maxBufferItems , int flushIntervalMillis ,
134123 RequestProcessor requestProcessor ) {
135- // Without a positive buffer limit nothing bounds the buffer between timed flushes, and with
136- // the timer off as well it would grow for as long as the process runs.
124+ // The buffer limit is the only bound on the buffer between timed flushes, or with no timer.
137125 if (maxBufferItems < 1 ) {
138126 throw new IllegalArgumentException ("maxBufferItems must be at least 1." );
139127 }
@@ -164,10 +152,9 @@ private static Retry buildRetry() {
164152 }
165153
166154 /**
167- * The longest one batch can take to be delivered or given up on: every attempt the retry
168- * policy allows, each running to the client's timeouts, plus the backoff between them. An
169- * attempt is bounded by the call timeout when one is set, and otherwise by connect, write and
170- * read in turn. When none of those bounds it, neither is the batch.
155+ * The longest one batch can take: every attempt the retry policy allows, each at the client's
156+ * timeouts, plus backoff. An attempt is bounded by the call timeout if set, else by connect,
157+ * write and read in turn.
171158 *
172159 * @return the worst case in milliseconds, or {@link #UNBOUNDED}
173160 */
@@ -194,9 +181,9 @@ static long worstCaseBatchMillis(OkHttpClient client, Retry retry) {
194181 }
195182
196183 /**
197- * Set the logger used by the processor and by its request processor.
184+ * Set the logger, for this processor and its request processor.
198185 *
199- * @param logger logger instance
186+ * @param logger the client's logger, so event failures appear alongside its other output
200187 */
201188 public void setLogger (FlagsmithLogger logger ) {
202189 this .logger = logger ;
@@ -282,9 +269,8 @@ public synchronized void start() {
282269 }
283270
284271 if (closed .get ()) {
285- // The scheduler is shut down, and scheduling on it would throw. This happens when a
286- // FlagsmithConfig, which owns the processor, is reused after a client built from it was
287- // closed.
272+ // Scheduling on the shut-down scheduler would throw. Reached when a FlagsmithConfig is
273+ // reused after a client built from it was closed.
288274 logger .error ("Not starting the event processor: it has been closed." );
289275 return ;
290276 }
@@ -294,10 +280,8 @@ public synchronized void start() {
294280 }
295281
296282 /**
297- * Stop the flush timer, ship whatever is left on a best-effort basis and release the HTTP
298- * resources. Blocks until in-flight batches settle, for at most the worst case of one batch
299- * under the HTTP client's timeouts and the retry policy; with a timeout switched off there is
300- * no such bound, and it waits as long as the request does.
283+ * Stop the timer, flush what is left and release HTTP resources. Blocks until in-flight
284+ * batches settle, for at most the worst case of one batch; unbounded if a client timeout is off.
301285 */
302286 public void close () {
303287 synchronized (this ) {
@@ -322,8 +306,7 @@ public void close() {
322306 logger .error ("Failed to flush events on close." , e );
323307 }
324308
325- // shutdown(), never shutdownNow(): interrupting a POST mid-flight loses its batch, where
326- // letting it finish delivers it if the JVM stays up long enough.
309+ // A graceful shutdown: interrupting a POST mid-flight would lose its batch.
327310 requestProcessor .close ();
328311 }
329312
@@ -344,17 +327,10 @@ private void bufferEvent(String event, String featureName, String identifier, Ob
344327 eventMetadata .put (SDK_VERSION_KEY , Versions .getVersion ());
345328 final Object experimentId = eventMetadata .get (EXPERIMENT_ID_KEY );
346329
347- // Traits and metadata are caller objects of any shape. Serialising them here, rather than
348- // at flush time, means a value Jackson cannot serialise drops this one event (logged
349- // below) instead of failing the whole batch it would later be sent in. The JSON text is
350- // also a deep copy, so a caller mutating a nested map afterwards cannot change a buffered
351- // event, and is more compact to hold than the objects or a JSON tree.
352- //
353- // It must be writeValueAsString, not valueToTree: on a map or list that contains itself,
354- // valueToTree lets a raw StackOverflowError escape, where writeValueAsString reports it as
355- // a JsonMappingException that the catch below handles.
356330 Map <String , Object > eventTraits = eventTraits (traits );
357331
332+ // Serialised now, not at flush: an unserialisable value drops only this event, not its
333+ // batch, and the JSON text is a deep copy that later caller mutation cannot reach.
358334 Map <String , Object > eventPayload = new LinkedHashMap <>();
359335 eventPayload .put ("event" , event );
360336 eventPayload .put ("feature_name" , featureName );
@@ -367,9 +343,8 @@ private void bufferEvent(String event, String featureName, String identifier, Ob
367343 boolean isFull ;
368344
369345 synchronized (lock ) {
370- // Checked again under the lock: close() sets the flag before its final flush takes the
371- // lock, so an event either lands in the buffer ahead of that flush or is refused here.
372- // Checking only above would let an event slip in after the final flush and be lost.
346+ // Re-checked under the lock: close() sets the flag before its final flush takes the
347+ // lock, so an event either makes that flush or is refused here, never stranded.
373348 if (closed .get ()) {
374349 logClosed (event );
375350 return ;
@@ -390,7 +365,10 @@ private void bufferEvent(String event, String featureName, String identifier, Ob
390365 }
391366 }
392367
393- /** JSON text Jackson writes out verbatim when the batch is serialised. */
368+ /**
369+ * JSON text written verbatim into the batch. Uses writeValueAsString, not valueToTree: on a
370+ * self-containing map valueToTree throws a raw StackOverflowError instead of an exception.
371+ */
394372 private static RawValue toJson (Object value ) throws JsonProcessingException {
395373 return new RawValue (MapperFactory .getMapper ().writeValueAsString (value ));
396374 }
@@ -400,10 +378,8 @@ private void logClosed(String event) {
400378 }
401379
402380 /**
403- * Flatten a caller trait map into the flat map of trait values the events API expects. Values
404- * wrapped in a {@link TraitConfig} are unwrapped, and traits the caller marked transient are
405- * dropped: transient means "do not persist this against the identity", and an event store keeps
406- * what it is sent.
381+ * Unwrap {@link TraitConfig} values and drop transient traits: transient means "do not
382+ * persist", and an event store keeps what it is sent.
407383 */
408384 private static Map <String , Object > eventTraits (Map <String , Object > traits ) {
409385 if (traits == null ) {
@@ -437,13 +413,11 @@ private static String nullToEmpty(String value) {
437413 }
438414
439415 /**
440- * Hand a batch to the request processor. {@code tracked} is settled exactly once, on every path
441- * out of here: leaving it pending would wedge every later {@link #flush()}, since those wait on
442- * it.
416+ * Hand a batch to the request processor. {@code tracked} is settled on every path out: left
417+ * pending, it would wedge every later {@link #flush()}.
443418 */
444419 private void send (List <Map <String , Object >> batch , CompletableFuture <Void > tracked ) {
445- // The completion callback outlives this call by as long as the POST takes, so it captures
446- // the size rather than the list: the serialised request already holds the batch's content.
420+ // The callback lives as long as the POST, so it holds the size and lets the list be freed.
447421 final int batchSize = batch .size ();
448422 boolean submitted = false ;
449423
@@ -505,10 +479,8 @@ private void settle(CompletableFuture<Void> tracked, int batchSize) {
505479 }
506480
507481 /**
508- * Count dropped events, and say whether it is time to report them. The first drop is reported
509- * at once; later ones accumulate and are reported at most once per
510- * {@link #DROP_LOG_INTERVAL_NANOS}, so a caller saturating the processor during an outage
511- * cannot turn every flush into an error line. Called under the lock.
482+ * Count dropped events. The first drop is reported at once, later ones at most once per
483+ * {@link #DROP_LOG_INTERVAL_NANOS}, so an outage does not log per flush. Called under the lock.
512484 *
513485 * @return the number of drops to report now, or 0 to stay quiet
514486 */
@@ -528,10 +500,7 @@ private CompletableFuture<Void> awaitInFlight() {
528500 return CompletableFuture .allOf (inFlight .toArray (new CompletableFuture [0 ]));
529501 }
530502
531- /**
532- * A snapshot of the buffer, for tests. Taken under the lock, so a test never iterates the live
533- * list while another thread is appending to it.
534- */
503+ /** A snapshot of the buffer for tests, taken under the lock. */
535504 List <Map <String , Object >> bufferedEvents () {
536505 synchronized (lock ) {
537506 return new ArrayList <>(buffer );
0 commit comments