Thread safety around LocalBucketing (#63)

* adding some thread safety around LocalBucketing

* removed old, unused getVariable() method

Author: chris-hoefgen

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

@@ -203,7 +203,7 @@ private byte[] readAssemblyScriptUint8Array(int address)
         return bufferData;
     }
 
-    public void storeConfig(String sdkKey, String config) {
+    public synchronized void storeConfig(String sdkKey, String config) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int configAddress = newUint8ArrayParameter(config.getBytes(StandardCharsets.UTF_8));
@@ -213,15 +213,15 @@ public void storeConfig(String sdkKey, String config) {
         fn.accept(sdkKeyAddress, configAddress);
     }
 
-    public void setPlatformData(String platformData) {
+    public synchronized void setPlatformData(String platformData) {
         unpinAll();
         int platformDataAddress = newUint8ArrayParameter(platformData.getBytes(StandardCharsets.UTF_8));
         Func setPlatformDataPtr = linker.get(store, "", "setPlatformDataUTF8").get().func();
         WasmFunctions.Consumer1<Integer> fn = WasmFunctions.consumer(store, setPlatformDataPtr, I32);
         fn.accept(platformDataAddress);
     }
 
-    public void setClientCustomData(String sdkKey, String customData) {
+    public synchronized void setClientCustomData(String sdkKey, String customData) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int customDataAddress = newUint8ArrayParameter(customData.getBytes(StandardCharsets.UTF_8));;
@@ -230,7 +230,7 @@ public void setClientCustomData(String sdkKey, String customData) {
         fn.accept(sdkKeyAddress, customDataAddress);
     }
 
-    public BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throws JsonProcessingException {
+    public synchronized BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throws JsonProcessingException {
         unpinAll();
         String userString = OBJECT_MAPPER.writeValueAsString(user);
 
@@ -249,30 +249,7 @@ public BucketedUserConfig generateBucketedConfig(String sdkKey, User user) throw
         return config;
     }
 
-    public String getVariable(String sdkKey, User user, String key, Variable.TypeEnum variableTypeEnum, boolean shouldTrackEvent) throws JsonProcessingException {
-        // need some kind of mutex?
-        String userString = OBJECT_MAPPER.writeValueAsString(user);
-
-        int wasmVariableType = this.variableTypeMap.get(variableTypeEnum);
-
-        unpinAll();
-        int sdkKeyAddress = getSDKKeyAddress(sdkKey);
-        int userAddress = newWasmString(userString);
-        int keyAddress = newWasmString(key);
-
-        Func getVariablePtr = linker.get(store, "", "variableForUser").get().func();
-        WasmFunctions.Function5<Integer, Integer, Integer, Integer, Integer, Integer> variableForUser = WasmFunctions.func(
-                store, getVariablePtr, I32, I32, I32, I32, I32, I32);
-
-        int resultAddress = variableForUser.call(sdkKeyAddress, userAddress, keyAddress, wasmVariableType, shouldTrackEvent ? 1 : 0);
-        if (resultAddress == 0){
-            return null;
-        }
-        String variableString = readWasmString(resultAddress);
-        return variableString;
-    }
-
-    public byte[] getVariableForUserProtobuf(byte[] serializedParams){
+    public synchronized byte[] getVariableForUserProtobuf(byte[] serializedParams){
         int paramsAddr = newUint8ArrayParameter(serializedParams);
 
         Func getVariablePtr = linker.get(store, "", "variableForUser_PB").get().func();
@@ -290,7 +267,7 @@ public byte[] getVariableForUserProtobuf(byte[] serializedParams){
         return varBytes;
     }
 
-    public void initEventQueue(String sdkKey, String options) {
+    public synchronized void initEventQueue(String sdkKey, String options) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int optionsAddress = newWasmString(options);
@@ -300,7 +277,7 @@ public void initEventQueue(String sdkKey, String options) {
         fn.accept(sdkKeyAddress, optionsAddress);
     }
 
-    public void queueEvent(String sdkKey, String user, String event) {
+    public synchronized void queueEvent(String sdkKey, String user, String event) {
         unpinAll();
         int sdkKeyAddress = newWasmString(sdkKey);
         int userAddress = getPinnedParameter(user);
@@ -311,7 +288,7 @@ public void queueEvent(String sdkKey, String user, String event) {
         fn.accept(sdkKeyAddress, userAddress, eventAddress);
     }
 
-    public void queueAggregateEvent(String sdkKey, String event, String variableVariationMap) {
+    public synchronized void queueAggregateEvent(String sdkKey, String event, String variableVariationMap) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int eventAddress = getPinnedParameter(event);
@@ -322,7 +299,7 @@ public void queueAggregateEvent(String sdkKey, String event, String variableVari
         fn.accept(sdkKeyAddress, eventAddress, variableVariationMapAddress);
     }
 
-    public FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingException {
+    public synchronized FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingException {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
 
@@ -345,7 +322,7 @@ public FlushPayload[] flushEventQueue(String sdkKey) throws JsonProcessingExcept
         return payloads;
     }
 
-    public void onPayloadFailure(String sdkKey, String payloadId, boolean retryable) {
+    public synchronized void onPayloadFailure(String sdkKey, String payloadId, boolean retryable) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int payloadIdAddress = newWasmString(payloadId);
@@ -355,7 +332,7 @@ public void onPayloadFailure(String sdkKey, String payloadId, boolean retryable)
         fn.accept(sdkKeyAddress, payloadIdAddress, retryable ? 1 : 0);
     }
 
-    public void onPayloadSuccess(String sdkKey, String payloadId) {
+    public synchronized void onPayloadSuccess(String sdkKey, String payloadId) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);
         int payloadIdAddress = newWasmString(payloadId);
@@ -365,7 +342,7 @@ public void onPayloadSuccess(String sdkKey, String payloadId) {
         fn.accept(sdkKeyAddress, payloadIdAddress);
     }
 
-    public int getEventQueueSize(String sdkKey) {
+    public synchronized int getEventQueueSize(String sdkKey) {
         unpinAll();
         int sdkKeyAddress = getSDKKeyAddress(sdkKey);