DVC-4412: implement EventQueueManager and track

build.gradle

@@ -129,6 +129,10 @@ dependencies {
 
     implementation ("io.github.kawamuray.wasmtime:wasmtime-java:$wasmtime_version")
 
+    implementation("com.squareup.okhttp3:logging-interceptor:$okhttp_version") {
+        exclude group: "com.squareup.okhttp3", module: "okhttp"
+    }
+
     compileOnly("org.projectlombok:lombok:$lombok_version")
 
     testAnnotationProcessor("org.projectlombok:lombok:$lombok_version")

src/main/java/com/devcycle/sdk/server/common/api/IDVCApi.java

@@ -1,6 +1,7 @@
 package com.devcycle.sdk.server.common.api;
 
 import com.devcycle.sdk.server.common.model.*;
+import com.devcycle.sdk.server.local.model.EventsBatch;
 import retrofit2.Call;
 import retrofit2.http.*;
 
@@ -62,4 +63,15 @@ public interface IDVCApi {
   @Headers({"Content-Type:application/json"})
   @GET("config/v1/server/{sdkToken}.json")
   Call<ProjectConfig> getConfig(@Path("sdkToken") String sdkToken, @Header("If-None-Match") String etag);
+
+  /**
+   * Post events to DevCycle for user
+   *
+   * @param eventsBatch  (required)
+   * @return Call&lt;DVCResponse&gt;
+   */
+  @Headers({"Content-Type:application/json"})
+  @POST("v1/events/batch")
+  Call<DVCResponse> publishEvents(@Body EventsBatch eventsBatch);
 }
+

src/main/java/com/devcycle/sdk/server/local/api/DVCLocalApiClient.java

@@ -2,6 +2,7 @@
 
 import com.devcycle.sdk.server.common.api.IDVCApi;
 import com.devcycle.sdk.server.local.model.DVCLocalOptions;
+import com.devcycle.sdk.server.local.model.FlushPayload;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.annotation.JsonInclude;
 
@@ -16,6 +17,7 @@ public final class DVCLocalApiClient {
 
   private final OkHttpClient.Builder okBuilder;
   private final Retrofit.Builder adapterBuilder;
+  private final Retrofit.Builder eventsBuilder;
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
@@ -44,9 +46,13 @@ private DVCLocalApiClient(DVCLocalOptions options) {
     adapterBuilder = new Retrofit.Builder()
         .baseUrl(url)
         .addConverterFactory(JacksonConverterFactory.create());
+
+    eventsBuilder = new Retrofit.Builder()
+            .baseUrl("https://events.devcycle.com")
+            .addConverterFactory(JacksonConverterFactory.create());
   }
 
-  public DVCLocalApiClient(String apiKey, DVCLocalOptions options) {
+  public DVCLocalApiClient(String serverKey, DVCLocalOptions options) {
     this(options);
   }
 

src/main/java/com/devcycle/sdk/server/local/api/DVCLocalClient.java

@@ -6,6 +6,7 @@
 import com.devcycle.sdk.server.common.model.*;
 import com.devcycle.sdk.server.local.bucketing.LocalBucketing;
 import com.devcycle.sdk.server.local.managers.EnvironmentConfigManager;
+import com.devcycle.sdk.server.local.managers.EventQueueManager;
 import com.devcycle.sdk.server.local.model.BucketedUserConfig;
 import com.devcycle.sdk.server.local.model.DVCLocalOptions;
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -22,6 +23,8 @@ public final class DVCLocalClient {
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
+  private EventQueueManager eventQueueManager;
+
   public DVCLocalClient(String serverKey) {
     this(serverKey, DVCLocalOptions.builder().build());
   }
@@ -30,6 +33,7 @@ public DVCLocalClient(String serverKey, DVCLocalOptions dvcOptions) {
     configManager = new EnvironmentConfigManager(serverKey, localBucketing, dvcOptions);
     this.serverKey = serverKey;
     OBJECT_MAPPER.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+    eventQueueManager = new EventQueueManager(serverKey, localBucketing, dvcOptions);
   }
 
   /**
@@ -71,36 +75,36 @@ public <T> Variable<T> variable(User user, String key, T defaultValue) {
       System.out.println("Variable called before DVCClient has initialized, returning default value");
     }
 
+    Variable<T> defaultVariable = (Variable<T>) Variable.builder()
+            .key(key)
+            .value(defaultValue)
+            .isDefaulted(true)
+            .reasonUsingDefaultValue("Variable not found")
+            .build();
+
     try {
       String userString = OBJECT_MAPPER.writeValueAsString(user);
 
       BucketedUserConfig bucketedUserConfig = localBucketing.generateBucketedConfig(serverKey, userString);
       if (bucketedUserConfig.variables.containsKey(key)) {
         Variable<T> variable = bucketedUserConfig.variables.get(key);
         variable.setIsDefaulted(false);
+        eventQueueManager.queueAggregateEvent(Event.builder().type("aggVariableEvaluated").target(key).build(), bucketedUserConfig);
         return variable;
+      } else {
+        eventQueueManager.queueAggregateEvent(Event.builder().type("aggVariableDefaulted").target(key).build(), bucketedUserConfig);
+        return defaultVariable;
       }
     } catch (JsonProcessingException e) {
       System.out.printf("Unable to parse JSON for Variable %s due to error: %s", key, e.toString());
     }
 
-    Variable<T> variable;
-
-    variable = (Variable<T>) Variable.builder()
-        .key(key)
-        .value(defaultValue)
-        .isDefaulted(true)
-        .reasonUsingDefaultValue("Variable not found")
-        .build();
-
-    // TODO queue events
-    // eventQueue.queueAggregateEvent(
-    // user,
-    // new Event(type: EventTypes.variableDefaulted, target: key),
-    // null
-    // );
-
-    return variable;
+    try {
+      eventQueueManager.queueAggregateEvent(Event.builder().type("aggVariableDefaulted").target(key).build(), null);
+    } catch (JsonProcessingException e) {
+      System.out.printf("Unable to parse aggVariableDefaulted event for Variable %s due to error: %s", key, e.toString());
+    }
+    return defaultVariable;
   }
 
   /**
@@ -123,19 +127,11 @@ public Map<String, Variable> allVariables(User user) throws JsonProcessingExcept
    * @param user  (required)
    * @param event (required)
    */
-  // TODO: Original return type should match the line below, uncomment once
-  // implemented and delete the void return
-  // public DVCResponse track(User user, Event event) throws DVCException {
-  public void track(User user, Event event) {
+  public void track(User user, Event event) throws Exception {
     validateUser(user);
     localBucketing.setPlatformData(user.getPlatformData().toString());
 
-    UserAndEvents userAndEvents = UserAndEvents.builder()
-        .user(user)
-        .events(Collections.singletonList(event))
-        .build();
-
-    // Call track method to append custom event to queue
+    eventQueueManager.queueEvent(user, event);
   }
 
   private void validateUser(User user) {

src/main/java/com/devcycle/sdk/server/local/api/DVCLocalEventsApiClient.java

@@ -0,0 +1,64 @@
+package com.devcycle.sdk.server.local.api;
+
+import com.devcycle.sdk.server.common.api.IDVCApi;
+import com.devcycle.sdk.server.common.interceptor.AuthorizationHeaderInterceptor;
+import com.devcycle.sdk.server.local.model.DVCLocalOptions;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import okhttp3.OkHttpClient;
+import okhttp3.logging.HttpLoggingInterceptor;
+import okhttp3.logging.HttpLoggingInterceptor.Level;
+import retrofit2.Retrofit;
+import retrofit2.converter.jackson.JacksonConverterFactory;
+
+import java.util.Objects;
+
+public final class DVCLocalEventsApiClient {
+
+    private final OkHttpClient.Builder okBuilder;
+    private final Retrofit.Builder adapterBuilder;
+    private HttpLoggingInterceptor logging = new HttpLoggingInterceptor();
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private static final String EVENTS_API_URL = "https://events.devcycle.com/";
+
+    private String eventsApiUrl;
+
+    private DVCLocalEventsApiClient(DVCLocalOptions options) {
+        String url;
+
+        OBJECT_MAPPER.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        okBuilder = new OkHttpClient.Builder();
+
+        String eventsApiUrlFromOptions = options.getEventsApiBaseUrl();
+
+        eventsApiUrl = checkIfStringNullOrEmpty(eventsApiUrlFromOptions) ? EVENTS_API_URL : eventsApiUrlFromOptions;
+
+        url = this.eventsApiUrl;
+
+        adapterBuilder = new Retrofit.Builder()
+                .baseUrl(url)
+                .addConverterFactory(JacksonConverterFactory.create());
+    }
+
+    public DVCLocalEventsApiClient(String serverKey, DVCLocalOptions options) {
+        this(options);
+        okBuilder.addInterceptor(new AuthorizationHeaderInterceptor(serverKey));
+        logging.setLevel(Level.BODY);
+
+        okBuilder.addInterceptor(logging);
+    }
+
+    public IDVCApi initialize() {
+        return adapterBuilder
+                .client(okBuilder.build())
+                .build()
+                .create(IDVCApi.class);
+    }
+
+    private Boolean checkIfStringNullOrEmpty(String stringToCheck) {
+        return Objects.isNull(stringToCheck) || Objects.equals(stringToCheck, "");
+    }
+}

src/main/java/com/devcycle/sdk/server/local/bucketing/LocalBucketing.java

@@ -12,7 +12,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.devcycle.sdk.server.local.model.BucketedUserConfig;
-import com.devcycle.sdk.server.local.model.EventPayload;
+import com.devcycle.sdk.server.local.model.FlushPayload;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -180,7 +180,7 @@ public void queueAggregateEvent(String token, String event, String variableVaria
         fn.accept(tokenAddress, eventAddress, variableVariationMapAddress);
     }
 
-    public EventPayload[] flushEventQueue(String token) throws JsonProcessingException {
+    public FlushPayload[] flushEventQueue(String token) throws JsonProcessingException {
         int tokenAddress = newWasmString(token);
 
         Func flushEventQueuePtr = linker.get(store, "", "flushEventQueue").get().func();
@@ -197,7 +197,7 @@ public EventPayload[] flushEventQueue(String token) throws JsonProcessingExcepti
         df.setTimeZone(TimeZone.getTimeZone("UTC"));
         objectMapper.setDateFormat(df);
 
-        EventPayload[] payloads = objectMapper.readValue(flushPayloadsStr, EventPayload[].class);
+        FlushPayload[] payloads = objectMapper.readValue(flushPayloadsStr, FlushPayload[].class);
 
         return payloads;
     }

src/main/java/com/devcycle/sdk/server/local/managers/EnvironmentConfigManager.java

@@ -31,20 +31,18 @@ public final class EnvironmentConfigManager {
   private ProjectConfig config;
   private String configETag = "";
 
-  private String environmentKey;
+  private String serverKey;
   private int pollingIntervalMS;
 
-  public EnvironmentConfigManager(String environmentKey, LocalBucketing localBucketing, DVCLocalOptions options) {
-    this.environmentKey = environmentKey;
+  public EnvironmentConfigManager(String serverKey, LocalBucketing localBucketing, DVCLocalOptions options) {
+    this.serverKey = serverKey;
     this.localBucketing = localBucketing;
 
-    configApiClient = new DVCLocalApiClient(environmentKey, options).initialize();
+    configApiClient = new DVCLocalApiClient(serverKey, options).initialize();
 
     int configPollingIntervalMs = options.getConfigPollingIntervalMs();
     pollingIntervalMS = configPollingIntervalMs >= MIN_INTERVALS_MS ? configPollingIntervalMs
         : DEFAULT_POLL_INTERVAL_MS;
-
-    this.localBucketing = localBucketing;
 
     setupScheduler();
   }
@@ -68,7 +66,7 @@ public boolean isConfigInitialized() {
   }
 
   private ProjectConfig getConfig() throws DVCException, JsonProcessingException {
-    Call<ProjectConfig> config = this.configApiClient.getConfig(this.environmentKey, this.configETag);
+    Call<ProjectConfig> config = this.configApiClient.getConfig(this.serverKey, this.configETag);
 
     this.config = getConfigResponse(config);
     return this.config;
@@ -93,7 +91,7 @@ private ProjectConfig getConfigResponse(Call<ProjectConfig> call) throws DVCExce
       ProjectConfig config = response.body();
       try {
         ObjectMapper mapper = new ObjectMapper();
-        localBucketing.storeConfig(environmentKey, mapper.writeValueAsString(config));
+        localBucketing.storeConfig(serverKey, mapper.writeValueAsString(config));
       } catch (JsonProcessingException e) {
         if (this.config != null) {
           System.out.printf("Unable to parse config with etag: %s. Using cache, etag %s%n", currentETag, this.configETag);