diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f307415..51b3f70 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,4 +22,4 @@ jobs: cache: maven - name: Verify - run: mvn --batch-mode verify + run: make verify diff --git a/Makefile b/Makefile index ab53d92..2e2a149 100644 --- a/Makefile +++ b/Makefile @@ -30,8 +30,8 @@ lint: test: mvn test -verify: - mvn verify +verify: lib + mvn --batch-mode verify run-dev: java -cp "$(APP_JAR):$(PORTICO_JAR)" com.yetanalytics.hlaxapi.App $(SIM_CONFIG) diff --git a/README.md b/README.md index 486c915..f1af085 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,12 @@ Example: } ``` +The frequency of the buffer being cleared and statement being actually posted can be changed with a java property when running the jar: + +``` +java -Dxapi.buffer.clear-rate=5000 ... +``` + ### Vendoring Portico `make lib` rebuilds and vendors only Portico's Java jar into this repository's local Maven file repository. Portico ships its own Ant wrapper at `codebase/ant`, so no separate Ant installation is required. diff --git a/config/xapi-config.json b/config/xapi-config.json index 609a55e..da1cf07 100644 --- a/config/xapi-config.json +++ b/config/xapi-config.json @@ -20,14 +20,21 @@ }, "result": { "response": "[<<[\"this\", [\"ToPosition\", \"X\"]]>>, <<[\"this\", [\"ToPosition\", \"Y\"]]>>]" + }, + "verb": { + "id": "http://yetanalytics.com/verbs/moved" + }, + "object": { + "id": "http://yetanalytics.com/objects/7" } } } ], "lrs": { - "host": "host string", - "key": "key string", - "secret": "secret string", - "batch": 4 + "host": "http://localhost:8080/xapi", + "key": "my_key", + "secret": "my_secret", + "batch": 35, + "maxRetries": 3 } } diff --git a/pom.xml b/pom.xml index 600d754..a344d9b 100644 --- a/pom.xml +++ b/pom.xml @@ -105,6 +105,12 @@ spring-context 6.1.18 + + + com.yetanalytics + xapi-tools + 0.0.4 + diff --git a/src/main/java/com/yetanalytics/hlaxapi/AppConfig.java b/src/main/java/com/yetanalytics/hlaxapi/AppConfig.java index 891ddc5..2945660 100644 --- a/src/main/java/com/yetanalytics/hlaxapi/AppConfig.java +++ b/src/main/java/com/yetanalytics/hlaxapi/AppConfig.java @@ -10,6 +10,7 @@ import org.apache.logging.log4j.Logger; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableScheduling; import com.yetanalytics.hlaxapi.config.ConfigParser; import com.yetanalytics.hlaxapi.config.XapiConfig; @@ -21,6 +22,7 @@ * Spring configuration class for beans related to HLA and simulation configuration. */ @Configuration +@EnableScheduling public class AppConfig { private static final Logger logger = LogManager.getLogger(AppConfig.class); diff --git a/src/main/java/com/yetanalytics/hlaxapi/HlaInterfaceImpl.java b/src/main/java/com/yetanalytics/hlaxapi/HlaInterfaceImpl.java index 09b6aab..0a19780 100755 --- a/src/main/java/com/yetanalytics/hlaxapi/HlaInterfaceImpl.java +++ b/src/main/java/com/yetanalytics/hlaxapi/HlaInterfaceImpl.java @@ -9,6 +9,8 @@ import org.apache.commons.lang3.StringUtils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; import com.yetanalytics.hlaxapi.config.XapiConfig; import com.yetanalytics.hlaxapi.config.model.StatementTrigger; @@ -57,9 +59,6 @@ import hla.rti1516e.exceptions.SaveInProgress; import hla.rti1516e.exceptions.UnsupportedCallbackModel; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; - @Component public class HlaInterfaceImpl extends NullFederateAmbassador implements HlaInterface { @@ -78,6 +77,9 @@ public class HlaInterfaceImpl extends NullFederateAmbassador implements HlaInter @Autowired private TriggerProcessor triggerProcessor; + @Autowired + private XapiClient xapiClient; + public void start() throws ConnectionFailed, InvalidLocalSettingsDesignator, RTIinternalError, NotConnected, ErrorReadingFDD, CouldNotOpenFDD, InconsistentFDD, RestoreInProgress, SaveInProgress, @@ -213,7 +215,7 @@ private void receiveInteraction(InteractionClassHandle interactionClass, Paramet logger.info("Received Interaction"); try { String interactionName = ambassador.getInteractionClassName(interactionClass); - logger.info("Interaction Handle: {}", interactionName); + logger.trace("Interaction Handle: {}", interactionName); String interactionKey = StringUtils.substringAfterLast(interactionName, "."); // Create Interaction-specific injection context to pass to trigger processor @@ -225,8 +227,13 @@ private void receiveInteraction(InteractionClassHandle interactionClass, Paramet .filter(trigger -> trigger.clazz.equals(interactionKey) && trigger.type.equals(StatementTrigger.Type.INTERACTION)) .forEach(trigger -> { - logger.info("Processing trigger for interaction {}", trigger.clazz); - triggerProcessor.processTrigger(trigger, context); + logger.trace("Processing trigger for interaction {}", trigger.clazz); + String xapi = triggerProcessor.processTrigger(trigger, context); + try { + xapiClient.sendStatementFromString(xapi); + } catch (Exception e) { + logger.error("Error parsing or posting statement {}", xapi, e); + } }); } catch (InvalidInteractionClassHandle | FederateNotExecutionMember | NotConnected | RTIinternalError e) { logger.error("Error ascertaining interaction details!", e); diff --git a/src/main/java/com/yetanalytics/hlaxapi/TriggerProcessor.java b/src/main/java/com/yetanalytics/hlaxapi/TriggerProcessor.java index b5d198e..c99afb1 100644 --- a/src/main/java/com/yetanalytics/hlaxapi/TriggerProcessor.java +++ b/src/main/java/com/yetanalytics/hlaxapi/TriggerProcessor.java @@ -48,7 +48,7 @@ public String processTrigger(StatementTrigger trigger, InjectionContext context) JsonNode processed = processNode(stmtNode, context, mapper); String output = mapper.writeValueAsString(processed); - logger.info("Processed statement output: {}", output); + logger.trace("Processed statement output: {}", output); return output; } catch (Exception e) { logger.debug("Could not process statement for trigger", e); diff --git a/src/main/java/com/yetanalytics/hlaxapi/XapiClient.java b/src/main/java/com/yetanalytics/hlaxapi/XapiClient.java new file mode 100644 index 0000000..d65ae05 --- /dev/null +++ b/src/main/java/com/yetanalytics/hlaxapi/XapiClient.java @@ -0,0 +1,98 @@ +package com.yetanalytics.hlaxapi; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonMappingException; +import com.yetanalytics.hlaxapi.config.XapiConfig; + +import com.yetanalytics.xapi.client.StatementClient; +import com.yetanalytics.xapi.exception.StatementClientException; +import com.yetanalytics.xapi.model.Statement; +import com.yetanalytics.xapi.client.LRS; + +import com.yetanalytics.xapi.util.Mapper; + +@Component +public class XapiClient { + + private static final Logger logger = LogManager.getLogger(XapiClient.class); + + private StatementClient client; + + private List buffer; + + private Integer batchSize; + + private Integer retryCount = 0; + + private Integer maxRetries; + + public XapiClient(XapiConfig xapiConfig) { + LRS lrs = new LRS( + xapiConfig.lrsConfig.host, + xapiConfig.lrsConfig.key, + xapiConfig.lrsConfig.secret, + xapiConfig.lrsConfig.batch + ); + client = new StatementClient(lrs); + buffer = new ArrayList(); + batchSize = xapiConfig.lrsConfig.batch; + maxRetries = xapiConfig.lrsConfig.maxRetries; + } + + private StatementClient getClient() { + return client; + } + + public void sendStatementFromString(String s) throws JsonMappingException, JsonProcessingException{ + Statement stmt = Mapper.getMapper().readValue(s, Statement.class); + addToBuffer(stmt); + } + + /** Synchronized buffer methods */ + + private synchronized void addToBuffer(Statement stmt){ + buffer.add(stmt); + } + + // Check and post buffer to LRS every 10 seconds (or ENV) if contains statements + @Scheduled(fixedRateString = "${xapi.buffer.clear-rate:10000}") + private synchronized void clearBuffer() { + logger.info("Buffer Size: {}", buffer.size()); + if (buffer.size() > 0){ + try { + List results = client.postStatements(buffer); + logger.info("Stored statements: {}", results); + buffer = new ArrayList(); + logger.info("Cleared Buffer"); + retryCount = 0; + } catch (StatementClientException e) { + logger.error("Error sending statements to LRS:", e); + if (retryCount < maxRetries) { + retryCount++; + logger.info("Retrying to send statements to LRS, attempt {}/{}", retryCount, maxRetries); + } else { + // TODO: For durability, we should write the buffer to a file or database or DLQ for retrying later + // might be a case for a light queueing system like RabbitMQ. Also failures will be reduced if we + // pre-validate statements before sending to the LRS. Also we may want to reduce the accrual of + // statements in the buffer if we have a DLQ strategy, that way less innocent statements are lost. + // For now, we will just clear the buffer after max retries. + // TODO: Additionally we should consider a cap on the buffer size, and if it exceeds that cap, + // we should start dropping statements or writing to a DLQ. + buffer = new ArrayList(); + retryCount = 0; + logger.info("Cleared Buffer after max retries, statement data lost!"); + } + } + } + } + +} diff --git a/src/main/java/com/yetanalytics/hlaxapi/config/model/LrsConfig.java b/src/main/java/com/yetanalytics/hlaxapi/config/model/LrsConfig.java index bf311af..7eae13a 100644 --- a/src/main/java/com/yetanalytics/hlaxapi/config/model/LrsConfig.java +++ b/src/main/java/com/yetanalytics/hlaxapi/config/model/LrsConfig.java @@ -5,9 +5,10 @@ public class LrsConfig { public String key; public String secret; public int batch; + public int maxRetries; @Override public String toString() { - return String.format("LrsConfig{host=%s,key=%s,batch=%d}", host, key, batch); + return String.format("LrsConfig{host=%s,key=%s,batch=%d,maxRetries=%d}", host, key, batch, maxRetries); } } diff --git a/src/test/java/com/yetanalytics/hlaxapi/XapiClientTest.java b/src/test/java/com/yetanalytics/hlaxapi/XapiClientTest.java new file mode 100644 index 0000000..c3f090b --- /dev/null +++ b/src/test/java/com/yetanalytics/hlaxapi/XapiClientTest.java @@ -0,0 +1,162 @@ +package com.yetanalytics.hlaxapi; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import org.junit.jupiter.api.Test; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.yetanalytics.hlaxapi.config.XapiConfig; +import com.yetanalytics.hlaxapi.config.model.LrsConfig; +import com.yetanalytics.xapi.client.LRS; +import com.yetanalytics.xapi.client.StatementClient; +import com.yetanalytics.xapi.exception.StatementClientException; +import com.yetanalytics.xapi.model.Statement; + +class XapiClientTest { + + private static final String STATEMENT_JSON = """ + { + "actor": { + "objectType": "Agent", + "name": "Test Pilot", + "mbox": "mailto:test@example.com" + }, + "verb": { + "id": "http://adlnet.gov/expapi/verbs/experienced" + }, + "object": { + "objectType": "Activity", + "id": "https://example.com/simulation/activity" + } + } + """; + + @Test + void buffersStatementFromJsonString() throws Exception { + XapiClient xapiClient = new XapiClient(config(4, 1)); + + xapiClient.sendStatementFromString(STATEMENT_JSON); + + assertEquals(1, buffer(xapiClient).size()); + } + + @Test + void rejectsInvalidStatementJson() { + XapiClient xapiClient = new XapiClient(config(4, 1)); + + assertThrows(JsonProcessingException.class, () -> xapiClient.sendStatementFromString("{")); + } + + @Test + void clearBufferPostsBufferedStatementsAndClearsBuffer() throws Exception { + XapiClient xapiClient = new XapiClient(config(4, 1)); + FakeStatementClient fakeClient = new FakeStatementClient(); + setClient(xapiClient, fakeClient); + + xapiClient.sendStatementFromString(STATEMENT_JSON); + clearBuffer(xapiClient); + + assertEquals(1, fakeClient.postedStatements.size()); + assertEquals(0, buffer(xapiClient).size()); + assertEquals(0, retryCount(xapiClient)); + } + + @Test + void clearBufferKeepsStatementsWhenClientErrorsBeforeMaxRetries() throws Exception { + XapiClient xapiClient = new XapiClient(config(4, 1)); + FakeStatementClient fakeClient = new FakeStatementClient(); + fakeClient.failuresRemaining = 1; + setClient(xapiClient, fakeClient); + + xapiClient.sendStatementFromString(STATEMENT_JSON); + clearBuffer(xapiClient); + + assertEquals(1, fakeClient.postAttempts); + assertEquals(1, buffer(xapiClient).size()); + assertEquals(1, retryCount(xapiClient)); + } + + @Test + void clearBufferClearsStatementsAfterMaxRetries() throws Exception { + XapiClient xapiClient = new XapiClient(config(4, 1)); + FakeStatementClient fakeClient = new FakeStatementClient(); + fakeClient.failuresRemaining = 2; + setClient(xapiClient, fakeClient); + + xapiClient.sendStatementFromString(STATEMENT_JSON); + clearBuffer(xapiClient); + clearBuffer(xapiClient); + + assertEquals(2, fakeClient.postAttempts); + assertEquals(0, buffer(xapiClient).size()); + assertEquals(0, retryCount(xapiClient)); + } + + private static XapiConfig config(int batch, int maxRetries) { + LrsConfig lrsConfig = new LrsConfig(); + lrsConfig.host = "https://example.com/xapi/"; + lrsConfig.key = "key"; + lrsConfig.secret = "secret"; + lrsConfig.batch = batch; + lrsConfig.maxRetries = maxRetries; + + XapiConfig xapiConfig = new XapiConfig(); + xapiConfig.lrsConfig = lrsConfig; + return xapiConfig; + } + + private static void clearBuffer(XapiClient xapiClient) throws Exception { + Method clearBuffer = XapiClient.class.getDeclaredMethod("clearBuffer"); + clearBuffer.setAccessible(true); + clearBuffer.invoke(xapiClient); + } + + private static void setClient(XapiClient xapiClient, StatementClient client) throws Exception { + Field clientField = XapiClient.class.getDeclaredField("client"); + clientField.setAccessible(true); + clientField.set(xapiClient, client); + } + + @SuppressWarnings("unchecked") + private static List buffer(XapiClient xapiClient) throws Exception { + Field bufferField = XapiClient.class.getDeclaredField("buffer"); + bufferField.setAccessible(true); + return (List) bufferField.get(xapiClient); + } + + private static int retryCount(XapiClient xapiClient) throws Exception { + Field retryCountField = XapiClient.class.getDeclaredField("retryCount"); + retryCountField.setAccessible(true); + return (Integer) retryCountField.get(xapiClient); + } + + private static class FakeStatementClient extends StatementClient { + private int failuresRemaining; + private int postAttempts; + private final List postedStatements = new ArrayList<>(); + + FakeStatementClient() { + super(new LRS("https://example.com/xapi/", "key", "secret", 4)); + } + + @Override + public List postStatements(List statements) { + postAttempts++; + if (failuresRemaining > 0) { + failuresRemaining--; + throw new StatementClientException("Client error"); + } + postedStatements.addAll(statements); + return statements.stream() + .map(statement -> UUID.randomUUID()) + .toList(); + } + } +}