Skip to content

Commit 05d2fc2

Browse files
Zaimwa9Zaimwa9
committed
fix: give each client its own event processor
FlagsmithConfig built and held the EventProcessor, so clients sharing one config shared a processor: the last build rebound its API key, closing either client stopped events for both, and a custom API wrapper with its own config left the processor in use unstarted. The config now carries only the event settings; FlagsmithClient.build() creates, binds and starts a processor per client from the builder's configuration. An injected processor is still used as given.
1 parent 9213caa commit 05d2fc2

4 files changed

Lines changed: 145 additions & 31 deletions

File tree

‎src/main/java/com/flagsmith/FlagsmithClient.java‎

Lines changed: 23 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,11 @@
2828
import java.util.concurrent.CompletableFuture;
2929
import java.util.function.Function;
3030
import java.util.stream.Collectors;
31+
import lombok.AccessLevel;
3132
import lombok.Data;
33+
import lombok.Getter;
3234
import lombok.NonNull;
35+
import lombok.Setter;
3336
import org.apache.commons.lang3.StringUtils;
3437
import org.slf4j.Logger;
3538
import org.slf4j.LoggerFactory;
@@ -44,6 +47,9 @@ public class FlagsmithClient {
4447
private FlagsmithSdk flagsmithSdk;
4548
private EvaluationContext evaluationContext;
4649
private PollingManager pollingManager;
50+
@Getter(AccessLevel.PACKAGE)
51+
@Setter(AccessLevel.NONE)
52+
private EventProcessor eventProcessor;
4753

4854
private FlagsmithClient() {
4955
}
@@ -375,13 +381,11 @@ public void trackExposureEvent(String featureName, String identifier, Object val
375381
* are not enabled
376382
*/
377383
public CompletableFuture<Void> flushEvents() {
378-
EventProcessor processor = getEventProcessor();
379-
380-
if (processor == null) {
384+
if (eventProcessor == null) {
381385
return CompletableFuture.completedFuture(null);
382386
}
383387

384-
return processor.flush();
388+
return eventProcessor.flush();
385389
}
386390

387391
/**
@@ -393,28 +397,20 @@ public void close() {
393397
pollingManager.stopPolling();
394398
}
395399

396-
EventProcessor eventProcessor = getEventProcessor();
397400
if (eventProcessor != null) {
398401
eventProcessor.close();
399402
}
400403

401404
flagsmithSdk.close();
402405
}
403406

404-
private EventProcessor getEventProcessor() {
405-
FlagsmithConfig config = getConfig();
406-
return config != null ? config.getEventProcessor() : null;
407-
}
408-
409407
private EventProcessor requireEventProcessor(String action) {
410-
EventProcessor processor = getEventProcessor();
411-
412-
if (processor == null) {
408+
if (eventProcessor == null) {
413409
throw new FlagsmithRuntimeError(
414410
"Events must be enabled to " + action + ". Use withEnableEvents(true).");
415411
}
416412

417-
return processor;
413+
return eventProcessor;
418414
}
419415

420416
private Flags getEnvironmentFlagsFromEvaluationContext() throws FlagsmithClientError {
@@ -710,7 +706,7 @@ public FlagsmithClient build() {
710706
if (configuration.getOfflineHandler() == null) {
711707
throw new FlagsmithRuntimeError("Offline handler must be provided to use offline mode.");
712708
}
713-
if (configuration.getEventProcessor() != null) {
709+
if (configuration.getEnableEvents()) {
714710
throw new FlagsmithRuntimeError("Events cannot be enabled in offline mode.");
715711
}
716712
}
@@ -771,10 +767,18 @@ public FlagsmithClient build() {
771767

772768
// Last, once nothing else can throw: starting the processor starts its flush timer, which a
773769
// failed build would otherwise leave running with no client to close it.
774-
if (configuration.getEventProcessor() != null) {
775-
configuration.getEventProcessor().setApi(client.flagsmithSdk);
776-
configuration.getEventProcessor().setLogger(client.logger);
777-
configuration.getEventProcessor().start();
770+
if (configuration.getEnableEvents()) {
771+
EventProcessor processor = configuration.getEventProcessor() != null
772+
? configuration.getEventProcessor()
773+
: new EventProcessor(
774+
configuration.getHttpClient(),
775+
configuration.getEventsUri(),
776+
configuration.getEventsMaxBufferItems(),
777+
configuration.getEventsFlushIntervalMillis());
778+
processor.setApi(client.flagsmithSdk);
779+
processor.setLogger(client.logger);
780+
processor.start();
781+
client.eventProcessor = processor;
778782
}
779783

780784
return this.client;

‎src/main/java/com/flagsmith/config/FlagsmithConfig.java‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -42,11 +42,15 @@ public final class FlagsmithConfig {
4242
private final OkHttpClient httpClient;
4343
private final HttpUrl baseUri;
4444
private final HttpUrl eventsUri;
45+
private final Boolean enableEvents;
46+
private final int eventsMaxBufferItems;
47+
private final int eventsFlushIntervalMillis;
4548

4649
private final Retry retries;
4750
private Boolean enableLocalEvaluation;
4851
private Integer environmentRefreshIntervalSeconds;
4952
private AnalyticsProcessor analyticsProcessor;
53+
/** The processor from withEventProcessor, or null; each client otherwise builds its own. */
5054
private EventProcessor eventProcessor;
5155
private FlagsmithFlagDefaults flagsmithFlagDefaults = null;
5256
private Boolean raiseUpdateEnvironmentErrorsOnStartup = true;
@@ -96,12 +100,18 @@ protected FlagsmithConfig(Builder builder) {
96100
}
97101

98102
this.eventsUri = builder.eventsUri;
103+
this.enableEvents = Boolean.TRUE.equals(builder.enableEvents);
104+
this.eventsMaxBufferItems = builder.eventsMaxBufferItems;
105+
this.eventsFlushIntervalMillis = builder.eventsFlushIntervalMillis;
99106

100-
if (Boolean.TRUE.equals(builder.enableEvents)) {
101-
eventProcessor = builder.eventProcessor != null
102-
? builder.eventProcessor
103-
: new EventProcessor(httpClient, eventsUri, builder.eventsMaxBufferItems,
104-
builder.eventsFlushIntervalMillis);
107+
if (enableEvents) {
108+
if (eventsMaxBufferItems < 1) {
109+
throw new IllegalArgumentException("maxBufferItems must be at least 1.");
110+
}
111+
if (eventsFlushIntervalMillis < 0) {
112+
throw new IllegalArgumentException("flushIntervalMillis must not be negative.");
113+
}
114+
eventProcessor = builder.eventProcessor;
105115
} else if (builder.eventsConfigured) {
106116
throw new IllegalArgumentException(
107117
"Events must be enabled with withEnableEvents(true) to configure the event processor.");
@@ -324,7 +334,8 @@ public Builder withEnableEvents(Boolean enable) {
324334
}
325335

326336
/**
327-
* Use a custom event processor. Also enables events.
337+
* Use a custom event processor. Also enables events. Unlike the default, the processor is
338+
* shared by every client built from this configuration.
328339
*
329340
* @param processor the processor that buffers and sends events
330341
* @return the Builder

‎src/test/java/com/flagsmith/FlagsmithClientTest.java‎

Lines changed: 101 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import static org.mockito.Mockito.*;
66
import static org.junit.jupiter.api.Assertions.assertEquals;
77
import static org.junit.jupiter.api.Assertions.assertNotNull;
8+
import static org.junit.jupiter.api.Assertions.assertNotSame;
89
import static org.junit.jupiter.api.Assertions.assertNull;
910
import static org.junit.jupiter.api.Assertions.assertThrows;
1011
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -40,9 +41,11 @@
4041

4142
import java.io.IOException;
4243
import java.util.ArrayList;
44+
import java.util.Collections;
4345
import java.util.HashMap;
4446
import java.util.List;
4547
import java.util.Map;
48+
import java.util.concurrent.TimeUnit;
4649
import java.util.regex.Pattern;
4750
import java.util.stream.Collectors;
4851
import java.util.stream.Stream;
@@ -1032,7 +1035,7 @@ public void testEventsUriWithoutEnableEventsIsHarmless() {
10321035
.eventsUri("http://events-uri")
10331036
.build();
10341037

1035-
assertNull(config.getEventProcessor());
1038+
assertFalse(config.getEnableEvents());
10361039
assertEquals("http://events-uri/", config.getEventsUri().toString());
10371040
}
10381041

@@ -1226,18 +1229,23 @@ public void testEventsSettingsReachTheProcessor() {
12261229
.withEventsMaxBufferItems(5)
12271230
.withEventsFlushIntervalMillis(0)
12281231
.build();
1232+
FlagsmithClient client = FlagsmithClient.newBuilder()
1233+
.withConfiguration(config)
1234+
.setApiKey("api-key")
1235+
.build();
12291236

1230-
EventProcessor processor = config.getEventProcessor();
1237+
EventProcessor processor = client.getEventProcessor();
12311238
assertEquals("http://events-uri/v1/events", processor.getEventsEndpoint().toString());
12321239
assertEquals(5, processor.getMaxBufferItems());
12331240
assertEquals(0, processor.getFlushIntervalMillis());
1241+
client.close();
12341242
}
12351243

12361244
@Test
12371245
public void testNullEnableEventsLeavesEventsDisabled() {
12381246
FlagsmithConfig config = FlagsmithConfig.newBuilder().withEnableEvents(null).build();
12391247

1240-
assertNull(config.getEventProcessor());
1248+
assertFalse(config.getEnableEvents());
12411249
}
12421250

12431251
@Test
@@ -1447,4 +1455,94 @@ public void testCloseFlushesBufferedEvents() throws FlagsmithClientError, IOExce
14471455
assertEquals("treatment", events.get(0).get("value").asText());
14481456
assertEquals(42, events.get(0).get("metadata").get("experiment_id").asInt());
14491457
}
1458+
1459+
/** An events-enabled config that records each events batch as "environment key|body". */
1460+
private static FlagsmithConfig recordingEventsConfig(List<String> batches) {
1461+
MockInterceptor interceptor = new MockInterceptor();
1462+
interceptor.addRule()
1463+
.post("http://events-uri/v1/events")
1464+
.anyTimes()
1465+
.respond("{\"accepted\": 1, \"rejected\": []}", MEDIATYPE_JSON);
1466+
return FlagsmithConfig.newBuilder()
1467+
.baseUri("http://bad-url")
1468+
.addHttpInterceptor((chain) -> {
1469+
Request request = chain.request();
1470+
if (request.url().toString().endsWith("/v1/events")) {
1471+
Buffer buffer = new Buffer();
1472+
request.body().writeTo(buffer);
1473+
batches.add(request.header("X-Environment-Key") + "|" + buffer.readUtf8());
1474+
}
1475+
return chain.proceed(request);
1476+
})
1477+
.addHttpInterceptor(interceptor)
1478+
.eventsUri("http://events-uri")
1479+
.withEnableEvents(Boolean.TRUE)
1480+
.withEventsFlushIntervalMillis(0)
1481+
.build();
1482+
}
1483+
1484+
@Test
1485+
public void testClientsSharingAConfigSendEventsUnderTheirOwnKeys() throws Exception {
1486+
List<String> batches = Collections.synchronizedList(new ArrayList<>());
1487+
FlagsmithConfig config = recordingEventsConfig(batches);
1488+
FlagsmithClient clientA = FlagsmithClient.newBuilder()
1489+
.withConfiguration(config).setApiKey("key-a").build();
1490+
FlagsmithClient clientB = FlagsmithClient.newBuilder()
1491+
.withConfiguration(config).setApiKey("key-b").build();
1492+
1493+
assertNotSame(clientA.getEventProcessor(), clientB.getEventProcessor());
1494+
1495+
clientA.trackEvent("purchase", "user-a");
1496+
clientB.trackEvent("purchase", "user-b");
1497+
clientA.flushEvents().get(5, TimeUnit.SECONDS);
1498+
clientB.flushEvents().get(5, TimeUnit.SECONDS);
1499+
1500+
assertEquals(2, batches.size());
1501+
assertTrue(batches.get(0).startsWith("key-a|"));
1502+
assertTrue(batches.get(0).contains("user-a") && !batches.get(0).contains("user-b"));
1503+
assertTrue(batches.get(1).startsWith("key-b|"));
1504+
assertTrue(batches.get(1).contains("user-b") && !batches.get(1).contains("user-a"));
1505+
clientA.close();
1506+
clientB.close();
1507+
}
1508+
1509+
@Test
1510+
public void testClosingOneClientLeavesAnotherOnTheSameConfigTracking() throws Exception {
1511+
List<String> batches = Collections.synchronizedList(new ArrayList<>());
1512+
FlagsmithConfig config = recordingEventsConfig(batches);
1513+
FlagsmithClient clientA = FlagsmithClient.newBuilder()
1514+
.withConfiguration(config).setApiKey("key-a").build();
1515+
FlagsmithClient clientB = FlagsmithClient.newBuilder()
1516+
.withConfiguration(config).setApiKey("key-b").build();
1517+
1518+
clientA.close();
1519+
clientB.trackEvent("purchase", "user-b");
1520+
clientB.flushEvents().get(5, TimeUnit.SECONDS);
1521+
1522+
assertEquals(1, batches.size());
1523+
assertTrue(batches.get(0).startsWith("key-b|"));
1524+
assertTrue(batches.get(0).contains("user-b"));
1525+
clientB.close();
1526+
}
1527+
1528+
@Test
1529+
public void testCustomApiWrapperWithAnEventsConfigDeliversEvents() throws Exception {
1530+
List<String> batches = Collections.synchronizedList(new ArrayList<>());
1531+
FlagsmithApiWrapper wrapper = new FlagsmithApiWrapper(
1532+
FlagsmithConfig.newBuilder().baseUri("http://bad-url").build(),
1533+
null, new FlagsmithLogger(), "wrapper-key");
1534+
FlagsmithClient client = FlagsmithClient.newBuilder()
1535+
.withFlagsmithApiWrapper(wrapper)
1536+
.withConfiguration(recordingEventsConfig(batches))
1537+
.setApiKey("wrapper-key")
1538+
.build();
1539+
1540+
client.trackEvent("purchase", "user-1");
1541+
client.flushEvents().get(5, TimeUnit.SECONDS);
1542+
1543+
assertEquals(1, batches.size());
1544+
assertTrue(batches.get(0).startsWith("wrapper-key|"));
1545+
assertTrue(batches.get(0).contains("user-1"));
1546+
client.close();
1547+
}
14501548
}

‎src/test/java/com/flagsmith/threads/EventProcessorTest.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -679,12 +679,13 @@ public void closeTimeout_followsTheConfiguredTimeouts() {
679679
.connectTimeout(1000)
680680
.writeTimeout(2000)
681681
.readTimeout(30000)
682-
.withEnableEvents(Boolean.TRUE)
683682
.build();
683+
EventProcessor processor =
684+
new EventProcessor(config.getHttpClient(), config.getEventsUri(), 1, 0);
684685

685686
// The read timeout the caller configured, not the SDK default.
686-
assertEquals(2 * (1000 + 2000 + 30000) + 200,
687-
config.getEventProcessor().getCloseTimeoutMillis());
687+
assertEquals(2 * (1000 + 2000 + 30000) + 200, processor.getCloseTimeoutMillis());
688+
processor.close();
688689
}
689690

690691
@Test

0 commit comments

Comments
 (0)