feat: onConfigurationChange (#97)

* feat: callback manager class and test
* feat: onConfigurationChange

Author: typotter

src/main/java/cloud/eppo/BaseEppoClient.java

@@ -19,6 +19,7 @@
 import java.util.Map;
 import java.util.Timer;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
@@ -580,6 +581,17 @@ public void setIsGracefulFailureMode(boolean isGracefulFailureMode) {
     this.isGracefulMode = isGracefulFailureMode;
   }
 
+  /**
+   * Subscribe to changes to the configuration.
+   *
+   * @param callback A function to be executed when the configuration changes.
+   * @return a Runnable which, when called unsubscribes the callback from configuration change
+   *     events.
+   */
+  public Runnable onConfigurationChange(Consumer<Configuration> callback) {
+    return requestor.onConfigurationChange(callback);
+  }
+
   /**
    * Returns the configuration object used by the EppoClient for assignment and bandit evaluation.
    *

src/main/java/cloud/eppo/ConfigurationRequestor.java

@@ -1,8 +1,10 @@
 package cloud.eppo;
 
 import cloud.eppo.api.Configuration;
+import cloud.eppo.callback.CallbackManager;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -19,6 +21,8 @@ public class ConfigurationRequestor {
   private CompletableFuture<Boolean> configurationFuture = null;
   private boolean initialConfigSet = false;
 
+  private final CallbackManager<Configuration> configChangeManager = new CallbackManager<>();
+
   public ConfigurationRequestor(
       @NotNull IConfigurationStore configurationStore,
       @NotNull EppoHttpClient client,
@@ -36,8 +40,7 @@ public void setInitialConfiguration(@NotNull Configuration configuration) {
       throw new IllegalStateException("Initial configuration has already been set");
     }
 
-    initialConfigSet =
-        configurationStore.saveConfiguration(configuration).thenApply(v -> true).join();
+    initialConfigSet = saveConfigurationAndNotify(configuration).thenApply(v -> true).join();
   }
 
   /**
@@ -65,10 +68,7 @@ public CompletableFuture<Boolean> setInitialConfiguration(
                       return false;
                     } else {
                       initialConfigSet =
-                          configurationStore
-                              .saveConfiguration(config)
-                              .thenApply((s) -> true)
-                              .join();
+                          saveConfigurationAndNotify(config).thenApply((s) -> true).join();
                       return true;
                     }
                   }
@@ -98,7 +98,7 @@ void fetchAndSaveFromRemote() {
       configBuilder.banditParameters(banditParametersJsonBytes);
     }
 
-    configurationStore.saveConfiguration(configBuilder.build()).join();
+    saveConfigurationAndNotify(configBuilder.build()).join();
   }
 
   /** Loads configuration asynchronously from the API server, off-thread. */
@@ -115,7 +115,7 @@ CompletableFuture<Void> fetchAndSaveFromRemoteAsync() {
     remoteFetchFuture =
         client
             .getAsync(Constants.FLAG_CONFIG_ENDPOINT)
-            .thenApply(
+            .thenCompose(
                 flagConfigJsonBytes -> {
                   synchronized (this) {
                     Configuration.Builder configBuilder =
@@ -137,9 +137,23 @@ CompletableFuture<Void> fetchAndSaveFromRemoteAsync() {
                       }
                     }
 
-                    return configurationStore.saveConfiguration(configBuilder.build()).join();
+                    return saveConfigurationAndNotify(configBuilder.build());
                   }
                 });
     return remoteFetchFuture;
   }
+
+  private CompletableFuture<Void> saveConfigurationAndNotify(Configuration configuration) {
+    CompletableFuture<Void> saveFuture = configurationStore.saveConfiguration(configuration);
+    return saveFuture.thenRun(
+        () -> {
+          synchronized (configChangeManager) {
+            configChangeManager.notifyCallbacks(configuration);
+          }
+        });
+  }
+
+  public Runnable onConfigurationChange(Consumer<Configuration> callback) {
+    return configChangeManager.subscribe(callback);
+  }
 }

src/main/java/cloud/eppo/callback/CallbackManager.java

@@ -0,0 +1,58 @@
+package cloud.eppo.callback;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A generic callback manager that allows registration and notification of callbacks.
+ *
+ * @param <T> The type of data that will be passed to the callbacks
+ */
+public class CallbackManager<T> {
+  private static final Logger log = LoggerFactory.getLogger(CallbackManager.class);
+  private final Map<String, Consumer<T>> subscribers;
+
+  public CallbackManager() {
+    this.subscribers = new ConcurrentHashMap<>();
+  }
+
+  /**
+   * Register a callback to be notified when events occur.
+   *
+   * @param callback The callback function to be called with event data
+   * @return A Runnable that can be called to unsubscribe the callback
+   */
+  public Runnable subscribe(Consumer<T> callback) {
+    String id = UUID.randomUUID().toString();
+    subscribers.put(id, callback);
+
+    return () -> subscribers.remove(id);
+  }
+
+  /**
+   * Notify all subscribers with the provided data.
+   *
+   * @param data The data to pass to all callbacks
+   */
+  public void notifyCallbacks(T data) {
+    subscribers
+        .values()
+        .forEach(
+            callback -> {
+              try {
+                callback.accept(data);
+              } catch (Exception e) {
+                log.error("Eppo SDK: Error thrown by callback: {}", e.getMessage());
+              }
+            });
+  }
+
+  /** Remove all subscribers. */
+  public void clear() {
+    subscribers.clear();
+  }
+}

src/test/java/cloud/eppo/ConfigurationRequestorTest.java

@@ -2,15 +2,20 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import cloud.eppo.api.Configuration;
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
@@ -160,4 +165,136 @@ public void testCacheWritesAfterBrokenFetch() throws IOException {
 
     assertNull(configStore.getConfiguration().getFlag("boolean_flag"));
   }
+
+  private ConfigurationStore mockConfigStore;
+  private EppoHttpClient mockHttpClient;
+  private ConfigurationRequestor requestor;
+
+  @BeforeEach
+  public void setup() {
+    mockConfigStore = mock(ConfigurationStore.class);
+    mockHttpClient = mock(EppoHttpClient.class);
+    requestor = new ConfigurationRequestor(mockConfigStore, mockHttpClient, false, true);
+  }
+
+  @Test
+  public void testConfigurationChangeListener() throws IOException {
+    // Setup mock response
+    String flagConfig = FileUtils.readFileToString(initialFlagConfigFile, StandardCharsets.UTF_8);
+    when(mockHttpClient.get(anyString())).thenReturn(flagConfig.getBytes());
+    when(mockConfigStore.saveConfiguration(any()))
+        .thenReturn(CompletableFuture.completedFuture(null));
+
+    List<Configuration> receivedConfigs = new ArrayList<>();
+
+    // Subscribe to configuration changes
+    Runnable unsubscribe = requestor.onConfigurationChange(receivedConfigs::add);
+
+    // Initial fetch should trigger the callback
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(1, receivedConfigs.size());
+
+    // Another fetch should trigger the callback again (fetches aren't optimized with eTag yet).
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(2, receivedConfigs.size());
+
+    // Unsubscribe should prevent further callbacks
+    unsubscribe.run();
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(2, receivedConfigs.size()); // Count should remain the same
+  }
+
+  @Test
+  public void testMultipleConfigurationChangeListeners() {
+    // Setup mock response
+    when(mockHttpClient.get(anyString())).thenReturn("{}".getBytes());
+    when(mockConfigStore.saveConfiguration(any()))
+        .thenReturn(CompletableFuture.completedFuture(null));
+
+    AtomicInteger callCount1 = new AtomicInteger(0);
+    AtomicInteger callCount2 = new AtomicInteger(0);
+
+    // Subscribe multiple listeners
+    Runnable unsubscribe1 = requestor.onConfigurationChange(v -> callCount1.incrementAndGet());
+    Runnable unsubscribe2 = requestor.onConfigurationChange(v -> callCount2.incrementAndGet());
+
+    // Fetch should trigger both callbacks
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(1, callCount1.get());
+    assertEquals(1, callCount2.get());
+
+    // Unsubscribe first listener
+    unsubscribe1.run();
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(1, callCount1.get()); // Should not increase
+    assertEquals(2, callCount2.get()); // Should increase
+
+    // Unsubscribe second listener
+    unsubscribe2.run();
+    requestor.fetchAndSaveFromRemote();
+    assertEquals(1, callCount1.get()); // Should not increase
+    assertEquals(2, callCount2.get()); // Should not increase
+  }
+
+  @Test
+  public void testConfigurationChangeListenerIgnoresFailedFetch() {
+    // Setup mock response to simulate failure
+    when(mockHttpClient.get(anyString())).thenThrow(new RuntimeException("Fetch failed"));
+
+    AtomicInteger callCount = new AtomicInteger(0);
+    requestor.onConfigurationChange(v -> callCount.incrementAndGet());
+
+    // Failed fetch should not trigger the callback
+    try {
+      requestor.fetchAndSaveFromRemote();
+    } catch (Exception e) {
+      // Expected
+    }
+    assertEquals(0, callCount.get());
+  }
+
+  @Test
+  public void testConfigurationChangeListenerIgnoresFailedSave() {
+    // Setup mock responses
+    when(mockHttpClient.get(anyString())).thenReturn("{}".getBytes());
+    when(mockConfigStore.saveConfiguration(any()))
+        .thenReturn(
+            CompletableFuture.supplyAsync(
+                () -> {
+                  throw new RuntimeException("Save failed");
+                }));
+
+    AtomicInteger callCount = new AtomicInteger(0);
+    requestor.onConfigurationChange(v -> callCount.incrementAndGet());
+
+    // Failed save should not trigger the callback
+    try {
+      requestor.fetchAndSaveFromRemote();
+    } catch (RuntimeException e) {
+      // Pass
+    }
+    assertEquals(0, callCount.get());
+  }
+
+  @Test
+  public void testConfigurationChangeListenerAsyncSave() {
+    // Setup mock responses
+    when(mockHttpClient.getAsync(anyString()))
+        .thenReturn(CompletableFuture.completedFuture("{\"flags\":{}}".getBytes()));
+
+    CompletableFuture<Void> saveFuture = new CompletableFuture<>();
+    when(mockConfigStore.saveConfiguration(any())).thenReturn(saveFuture);
+
+    AtomicInteger callCount = new AtomicInteger(0);
+    requestor.onConfigurationChange(v -> callCount.incrementAndGet());
+
+    // Start fetch
+    CompletableFuture<Void> fetch = requestor.fetchAndSaveFromRemoteAsync();
+    assertEquals(0, callCount.get()); // Callback should not be called yet
+
+    // Complete the save
+    saveFuture.complete(null);
+    fetch.join();
+    assertEquals(1, callCount.get()); // Callback should be called after save completes
+  }
 }